private Set<String> computeMissingExternalSourceTopics(final TopicsInfo topicsInfo,
                                                           final Cluster clusterMetadata) {
        final Set<String> missingExternalSourceTopics = new HashSet<>(topicsInfo.sourceTopics);
        missingExternalSourceTopics.removeAll(topicsInfo.repartitionSourceTopics.keySet());
        missingExternalSourceTopics.removeAll(clusterMetadata.topics());
        return missingExternalSourceTopics;
    }
@param clusterMetadata  cluster metadata, eg which topics exist on the brokers
streams/src/main/java/org/apache/kafka/streams/processor/internals/RepartitionTopics.java