class InternalTopicManager {

    public void makeReady(final Map<InternalTopicConfig, Integer> topics) {
        for (int i = 0; i < MAX_TOPIC_READY_TRY; i++) {
            try {
                final MetadataResponse metadata = streamsKafkaClient.fetchMetadata();
                final Map<String, Integer> existingTopicPartitions = fetchExistingPartitionCountByTopic(metadata);
                final Map<InternalTopicConfig, Integer> topicsToBeCreated = validateTopicPartitions(topics, existingTopicPartitions);
                if (metadata.brokers().size() < replicationFactor) {
                    throw new StreamsException("Found only " + metadata.brokers().size() + " brokers, " +
                        " but replication factor is " + replicationFactor + "." +
                        " Decrease replication factor for internal topics via StreamsConfig parameter \"replication.factor\""  +
                        " or add more brokers to your cluster.");
                }
                streamsKafkaClient.createTopics(topicsToBeCreated, replicationFactor, windowChangeLogAdditionalRetention, metadata);
                return;
            } catch (StreamsException ex) {
                log.warn("Could not create internal topics: " + ex.getMessage() + " Retry #" + i);
            }
            // backoff
            time.sleep(100L);
        }
        throw new StreamsException("Could not create internal topics.");
    }

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

}