private RepartitionTopics prepareRepartitionTopics(final Cluster metadata) {
        final RepartitionTopics repartitionTopics = new RepartitionTopics(
            taskManager.topologyMetadata(),
            internalTopicManager,
            copartitionedTopicsEnforcer,
            metadata,
            logPrefix
        );
        repartitionTopics.setup();
        final boolean isMissingInputTopics = !repartitionTopics.missingSourceTopicExceptions().isEmpty();
        if (isMissingInputTopics) {
            if (!taskManager.topologyMetadata().hasNamedTopologies()) {
                final String errorMsg = String.format("Missing source topics. %s", repartitionTopics.missingSourceTopics());
                log.error(errorMsg);
                throw new MissingSourceTopicException(errorMsg);
            } else {
                nonFatalExceptionsToHandle.addAll(repartitionTopics.missingSourceTopicExceptions());
            }
        }
        return repartitionTopics;
    }
Computes and assembles all repartition topic metadata then creates the topics if necessary. Also verifies
that all user input topics of each topology have been created ahead of time. If any such source topics are
missing from a NamedTopology, the assignor will skip distributing its tasks until they have been created
and invoke the exception handler (without killing the thread) once for each topology to alert the user of
the missing topics.
<p>
For regular applications without named topologies, the assignor will instead send a shutdown signal to
all clients so the user can identify and resolve the problem.

@return application metadata such as partition info of repartition topics, missing external topics, etc
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java