private void cleanUpCreatedTopics(final Set<String> topicsToCleanUp) {
        log.info("Starting to clean up internal topics {}.", topicsToCleanUp);

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

        final Set<String> topicsStillToCleanup = new HashSet<>(topicsToCleanUp);
        while (!topicsStillToCleanup.isEmpty()) {
            log.info("Going to cleanup internal topics: " + topicsStillToCleanup);
            final DeleteTopicsResult deleteTopicsResult = adminClient.deleteTopics(topicsStillToCleanup);
            final Map<String, KafkaFuture<Void>> deleteResultForTopic = deleteTopicsResult.topicNameValues();
            while (!deleteResultForTopic.isEmpty()) {
                for (final String topicName : new HashSet<>(topicsStillToCleanup)) {
                    if (!deleteResultForTopic.containsKey(topicName)) {
                        throw new IllegalStateException("Delete topic results do not contain internal topic " + topicName
                            + " to clean up. " + BUG_ERROR_MESSAGE);
                    }
                    final KafkaFuture<Void> deleteResult = deleteResultForTopic.get(topicName);
                    if (deleteResult.isDone()) {
                        try {
                            deleteResult.get();
                            topicsStillToCleanup.remove(topicName);
                        } catch (final ExecutionException executionException) {
                            final Throwable cause = executionException.getCause();
                            if (cause instanceof UnknownTopicOrPartitionException) {
                                log.info("Internal topic {} to clean up is missing", topicName);
                            } else if (cause instanceof LeaderNotAvailableException) {
                                log.info("The leader of internal topic {} to clean up is not available.", topicName);
                            } else if (cause instanceof TimeoutException) {
                                log.info("Cleaning up internal topic {} timed out.", topicName);
                            } else {
                                log.error("Unexpected error during cleanup of internal topics: ", cause);
                                throw new StreamsException(
                                    String.format("Could not clean up internal topics %s, because during the cleanup " +
                                            "of topic %s the following error occurred: ",
                                        topicsStillToCleanup, topicName),
                                    cause
                                );
                            }
                        } catch (final InterruptedException interruptedException) {
                            throw new InterruptException(interruptedException);
                        } finally {
                            deleteResultForTopic.remove(topicName);
                        }
                    }
                }

                maybeThrowTimeoutException(
                    Collections.singletonList(topicsStillToCleanup),
                    deadline,
                    String.format("Could not cleanup internal topics within %d milliseconds. This can happen if the " +
                            "Kafka cluster is temporarily not available or the broker did not complete topic creation " +
                            "before the cleanup. The following internal topics could not be cleaned up: %s",
                        retryTimeoutMs, topicsStillToCleanup)
                );

                if (!deleteResultForTopic.isEmpty()) {
                    Utils.sleep(100);
                }
            }

            maybeSleep(
                Collections.singletonList(topicsStillToCleanup),
                deadline,
                "validated"
            );
        }

        log.info("Completed cleanup of internal topics {}.", topicsToCleanUp);
    }
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