private Map<String, InternalTopicConfig> computeRepartitionTopicConfig(final Cluster clusterMetadata) {
        final Set<TopicsInfo> allTopicsInfo = new HashSet<>();
        final Map<String, InternalTopicConfig> allRepartitionTopicConfigs = new HashMap<>();
        for (final Map.Entry<String, Map<Subtopology, TopicsInfo>> topologyEntry : topologyMetadata.topologyToSubtopologyTopicsInfoMap().entrySet()) {
            final String topologyName = topologyMetadata.hasNamedTopologies() ? topologyEntry.getKey() : null;

            final Set<TopicsInfo> topicsInfoForTopology = new HashSet<>();
            final Set<String> missingSourceTopicsForTopology = new HashSet<>();
            final Map<String, InternalTopicConfig> repartitionTopicConfigsForTopology = new HashMap<>();

            for (final Map.Entry<Subtopology, TopicsInfo> subtopologyEntry : topologyEntry.getValue().entrySet()) {
                final TopicsInfo topicsInfo = subtopologyEntry.getValue();

                topicsInfoForTopology.add(topicsInfo);
                repartitionTopicConfigsForTopology.putAll(
                    topicsInfo.repartitionSourceTopics
                        .values()
                        .stream()
                        .collect(Collectors.toMap(InternalTopicConfig::name, topicConfig -> topicConfig)));

                final Set<String> missingSourceTopicsForSubtopology = computeMissingExternalSourceTopics(topicsInfo, clusterMetadata);
                missingSourceTopicsForTopology.addAll(missingSourceTopicsForSubtopology);
                if (!missingSourceTopicsForSubtopology.isEmpty()) {
                    final Subtopology subtopology = subtopologyEntry.getKey();
                    missingInputTopicsBySubtopology.put(subtopology, missingSourceTopicsForSubtopology);
                    log.error("Subtopology {} has missing source topics {} and will be excluded from the current assignment, "
                        + "this can be due to the consumer client's metadata being stale or because they have "
                        + "not been created yet. Please verify that you have created all input topics; if they "
                        + "do exist, you just need to wait for the metadata to be updated, at which time a new "
                        + "rebalance will be kicked off automatically and the topology will be retried at that time.",
                        subtopology.nodeGroupId, missingSourceTopicsForSubtopology);
                }
            }

            if (missingSourceTopicsForTopology.isEmpty()) {
                allRepartitionTopicConfigs.putAll(repartitionTopicConfigsForTopology);
                allTopicsInfo.addAll(topicsInfoForTopology);
            } else {
                log.debug("Skipping repartition topic validation for entire topology {} due to missing source topics {}",
                    topologyName, missingSourceTopicsForTopology);
            }
        }
        setRepartitionSourceTopicPartitionCount(allRepartitionTopicConfigs, allTopicsInfo, clusterMetadata);

        return allRepartitionTopicConfigs;
    }
@param clusterMetadata  cluster metadata, eg which topics exist on the brokers
streams/src/main/java/org/apache/kafka/streams/processor/internals/RepartitionTopics.java