### Question:
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); } time.sleep(100L); } throw new StreamsException("Could not create internal topics."); } InternalTopicManager(final StreamsKafkaClient streamsKafkaClient, final int replicationFactor,
                                final long windowChangeLogAdditionalRetention, final Time time); void makeReady(final Map<InternalTopicConfig, Integer> topics); Map<String, Integer> getNumPartitions(final Set<String> topics); void close(); static final String CLEANUP_POLICY_PROP; static final String RETENTION_MS; static final Long WINDOW_CHANGE_LOG_ADDITIONAL_RETENTION_DEFAULT; }### Answer:
@Test public void shouldCreateRequiredTopics() throws Exception { InternalTopicManager internalTopicManager = new InternalTopicManager(streamsKafkaClient, 1, WINDOW_CHANGE_LOG_ADDITIONAL_RETENTION_DEFAULT, time); internalTopicManager.makeReady(Collections.singletonMap(new InternalTopicConfig(topic, Collections.singleton(InternalTopicConfig.CleanupPolicy.compact), null), 1)); }
@Test public void shouldNotCreateTopicIfExistsWithDifferentPartitions() throws Exception { InternalTopicManager internalTopicManager = new InternalTopicManager(streamsKafkaClient, 1, WINDOW_CHANGE_LOG_ADDITIONAL_RETENTION_DEFAULT, time); boolean exceptionWasThrown = false; try { internalTopicManager.makeReady(Collections.singletonMap(new InternalTopicConfig(topic, Collections.singleton(InternalTopicConfig.CleanupPolicy.compact), null), 2)); } catch (StreamsException e) { exceptionWasThrown = true; } Assert.assertTrue(exceptionWasThrown); }