class StoreChangelogReader implements ChangelogReader {

    @Override
    public void validatePartitionExists(final TopicPartition topicPartition, final String storeName) {
        final long start = time.milliseconds();
        // fetch all info on all topics to avoid multiple remote calls
        if (partitionInfo.isEmpty()) {
            try {
                partitionInfo.putAll(consumer.listTopics());
            } catch (final TimeoutException e) {
                log.warn("{} Could not list topics so will fall back to partition by partition fetching", logPrefix);
            }
        }

        final long endTime = time.milliseconds() + partitionValidationTimeoutMs;
        while (!hasPartition(topicPartition) && time.milliseconds() < endTime) {
            try {
                final List<PartitionInfo> partitions = consumer.partitionsFor(topicPartition.topic());
                if (partitions != null) {
                    partitionInfo.put(topicPartition.topic(), partitions);
                }
            } catch (final TimeoutException e) {
                throw new StreamsException(String.format("Could not fetch partition info for topic: %s before expiration of the configured request timeout",
                                                         topicPartition.topic()));
            }
        }

        if (!hasPartition(topicPartition)) {
            throw new StreamsException(String.format("Store %s's change log (%s) does not contain partition %s",
                                                     storeName, topicPartition.topic(), topicPartition.partition()));
        }
        log.debug("{} Took {} ms to validate that partition {} exists", logPrefix, time.milliseconds() - start, topicPartition);
    }

    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);

}