class InternalTopicManager {

    public Map<String, Integer> getNumPartitions(final Set<String> topics) {
        for (int i = 0; i < MAX_TOPIC_READY_TRY; i++) {
            try {
                final MetadataResponse metadata = streamsKafkaClient.fetchMetadata();
                final Map<String, Integer> existingTopicPartitions = fetchExistingPartitionCountByTopic(metadata);
                existingTopicPartitions.keySet().retainAll(topics);

                return existingTopicPartitions;
            } catch (StreamsException ex) {
                log.warn("Could not get number of partitions: " + ex.getMessage() + " Retry #" + i);
            }
            // backoff
            time.sleep(100L);
        }
        throw new StreamsException("Could not get number of partitions.");
    }

    public  InternalTopicManager(final StreamsKafkaClient streamsKafkaClient, final int replicationFactor,
                                final long windowChangeLogAdditionalRetention, final Time time);

}