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.getProperties(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());
    }
Sets up internal topics.

Either the given topic are all created or the method fails with an exception.

@param topicConfigs internal topics to setup
streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopicManager.java