class StoreChangelogReader implements ChangelogReader {

    public void restore() {
        final long start = time.milliseconds();
        try {
            if (!consumer.subscription().isEmpty()) {
                throw new IllegalStateException(String.format("Restore consumer should have not subscribed to any partitions (%s) beforehand", consumer.subscription()));
            }
            final Map<TopicPartition, Long> endOffsets = consumer.endOffsets(stateRestorers.keySet());

            // remove any partitions where we already have all of the data
            final Map<TopicPartition, StateRestorer> needsRestoring = new HashMap<>();
            for (final Map.Entry<TopicPartition, Long> entry : endOffsets.entrySet()) {
                TopicPartition topicPartition = entry.getKey();
                Long offset = entry.getValue();
                final StateRestorer restorer = stateRestorers.get(topicPartition);
                if (restorer.checkpoint() >= offset) {
                    restorer.setRestoredOffset(restorer.checkpoint());
                } else {
                    needsRestoring.put(topicPartition, restorer);
                }
            }

            log.info("{} Starting restoring state stores from changelog topics {}", logPrefix, needsRestoring.keySet());

            consumer.assign(needsRestoring.keySet());
            final List<StateRestorer> needsPositionUpdate = new ArrayList<>();
            for (final StateRestorer restorer : needsRestoring.values()) {
                if (restorer.checkpoint() != StateRestorer.NO_CHECKPOINT) {
                    consumer.seek(restorer.partition(), restorer.checkpoint());
                    logRestoreOffsets(restorer.partition(),
                                      restorer.checkpoint(),
                                      endOffsets.get(restorer.partition()));
                    restorer.setStartingOffset(consumer.position(restorer.partition()));
                } else {
                    consumer.seekToBeginning(Collections.singletonList(restorer.partition()));
                    needsPositionUpdate.add(restorer);
                }
            }

            for (final StateRestorer restorer : needsPositionUpdate) {
                final long position = consumer.position(restorer.partition());
                restorer.setStartingOffset(position);
                logRestoreOffsets(restorer.partition(),
                                  position,
                                  endOffsets.get(restorer.partition()));
            }

            final Set<TopicPartition> partitions = new HashSet<>(needsRestoring.keySet());
            while (!partitions.isEmpty()) {
                final ConsumerRecords<byte[], byte[]> allRecords = consumer.poll(10);
                final Iterator<TopicPartition> partitionIterator = partitions.iterator();
                while (partitionIterator.hasNext()) {
                    restorePartition(endOffsets, allRecords, partitionIterator);
                }
            }
        } finally {
            consumer.assign(Collections.<TopicPartition>emptyList());
            log.debug("{} Took {} ms to restore all active states", logPrefix, time.milliseconds() - start);
        }
    }

    public  StoreChangelogReader(final String threadId, final Consumer<byte[], byte[]> consumer, final Time time, final long partitionValidationTimeoutMs);
    public  StoreChangelogReader(final Consumer<byte[], byte[]> consumer, final Time time, final long partitionValidationTimeoutMs);

}