private void processCreateTopicResults(final CreateTopicsResult createTopicsResult,
                                           final Set<String> topicStillToCreate,
                                           final Set<String> createdTopics,
                                           final long deadline) {
        final Map<String, Throwable> lastErrorsSeenForTopic = new HashMap<>();
        final Map<String, KafkaFuture<Void>> createResultForTopic = createTopicsResult.values();
        while (!createResultForTopic.isEmpty()) {
            for (final String topicName : new HashSet<>(topicStillToCreate)) {
                if (!createResultForTopic.containsKey(topicName)) {
                    cleanUpCreatedTopics(createdTopics);
                    throw new IllegalStateException("Create topic results do not contain internal topic " + topicName
                        + " to setup. " + BUG_ERROR_MESSAGE);
                }
                final KafkaFuture<Void> createResult = createResultForTopic.get(topicName);
                if (createResult.isDone()) {
                    try {
                        createResult.get();
                        createdTopics.add(topicName);
                        topicStillToCreate.remove(topicName);
                    } catch (final ExecutionException executionException) {
                        final Throwable cause = executionException.getCause();
                        if (cause instanceof TopicExistsException) {
                            lastErrorsSeenForTopic.put(topicName, cause);
                            log.info("Internal topic {} already exists. Topic is probably marked for deletion. " +
                                "Will retry to create this topic later (to let broker complete async delete operation first)",
                                topicName);
                        } else if (cause instanceof TimeoutException) {
                            lastErrorsSeenForTopic.put(topicName, cause);
                            log.info("Creating internal topic {} timed out.", topicName);
                        } else {
                            cleanUpCreatedTopics(createdTopics);
                            log.error("Unexpected error during creation of internal topic: ", cause);
                            throw new StreamsException(
                                String.format("Could not create internal topic %s for the following reason: ", topicName),
                                cause
                            );
                        }
                    } catch (final InterruptedException interruptedException) {
                        throw new InterruptException(interruptedException);
                    } finally {
                        createResultForTopic.remove(topicName);
                    }
                }
            }

            maybeThrowTimeoutExceptionDuringSetup(
                topicStillToCreate,
                createdTopics,
                lastErrorsSeenForTopic,
                deadline
            );

            if (!createResultForTopic.isEmpty()) {
                Utils.sleep(100);
            }
        }
    }
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