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 shouldThrowTimeoutExceptionIfTopicExistsDuringSetup() {
        setupTopicInMockAdminClient(topic1, Collections.emptyMap());
        final MockTime time = new MockTime(
            (Integer) config.get(StreamsConfig.consumerPrefix(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG)) / 15
        );
        final InternalTopicManager internalTopicManager =
            new InternalTopicManager(time, mockAdminClient, new StreamsConfig(config));
        final InternalTopicConfig internalTopicConfig = setupRepartitionTopicConfig(topic1, 1);

        final TimeoutException exception = assertThrows(
            TimeoutException.class,
            () -> internalTopicManager.setup(Collections.singletonMap(topic1, internalTopicConfig))
        );

        assertThat(
            exception.getMessage(),
            is("Could not create internal topics within " +
                    (Integer) config.get(StreamsConfig.consumerPrefix(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG)) / 2 +
                " milliseconds. This can happen if the Kafka cluster is temporarily not available or a topic is marked" +
                    " for deletion and the broker did not complete its deletion within the timeout." +
                    " The last errors seen per topic are:" +
                    " {" + topic1 + "=org.apache.kafka.common.errors.TopicExistsException: Topic test_topic exists already.}")
        );
    }