public void setup(final Map<String, InternalTopicConfig> topicConfigs) {
        log.info("Starting to setup internal topics {}.", topicConfigs.keySet());

        final long now = time.milliseconds();
        final long deadline = now + retryTimeoutMs;

        final Map<String, Map<String, String>> streamsSideTopicConfigs = topicConfigs.values().stream()
            .collect(Collectors.toMap(
                InternalTopicConfig::name,
                topicConfig -> topicConfig.properties(defaultTopicConfigs, windowChangeLogAdditionalRetention)
            ));
        final Set<String> createdTopics = new HashSet<>();
        final Set<String> topicStillToCreate = new HashSet<>(topicConfigs.keySet());
        while (!topicStillToCreate.isEmpty()) {
            final Set<NewTopic> newTopics = topicStillToCreate.stream()
                .map(topicName -> new NewTopic(
                        topicName,
                        topicConfigs.get(topicName).numberOfPartitions(),
                        Optional.of(replicationFactor)
                    ).configs(streamsSideTopicConfigs.get(topicName))
                ).collect(Collectors.toSet());

            log.info("Going to create internal topics: " + newTopics);
            final CreateTopicsResult createTopicsResult = adminClient.createTopics(newTopics);

            processCreateTopicResults(createTopicsResult, topicStillToCreate, createdTopics, deadline);

            maybeSleep(Collections.singletonList(topicStillToCreate), deadline, "created");
        }

        log.info("Completed setup of internal topics {}.", topicConfigs.keySet());
    }
@Test
    public void shouldThrowWhenCreateTopicsThrowsUnexpectedException() {
        final AdminClient admin = mock(AdminClient.class);
        final StreamsConfig streamsConfig = new StreamsConfig(config);
        final InternalTopicManager topicManager = new InternalTopicManager(time, admin, streamsConfig);
        final InternalTopicConfig internalTopicConfig = setupRepartitionTopicConfig(topic1, 1);
        final KafkaFutureImpl<TopicMetadataAndConfig> createTopicFailFuture = new KafkaFutureImpl<>();
        createTopicFailFuture.completeExceptionally(new IllegalStateException("Nobody expects the Spanish inquisition"));
        final NewTopic newTopic = newTopic(topic1, internalTopicConfig, streamsConfig);
        when(admin.createTopics(mkSet(newTopic)))
            .thenAnswer(answer -> new MockCreateTopicsResult(mkMap(
                mkEntry(topic1, createTopicFailFuture)
            )));

        assertThrows(StreamsException.class, () -> topicManager.setup(mkMap(
            mkEntry(topic1, internalTopicConfig)
        )));
    }