public Set<String> upstreamClusters() throws InterruptedException {
        return listTopics().stream()
            .filter(this::isHeartbeatTopic)
            .flatMap(x -> allSources(x).stream())
            .collect(Collectors.toSet());
    }
Finds upstream clusters, which may be multiple hops away, based on incoming heartbeats.
connect/mirror-client/src/main/java/org/apache/kafka/connect/mirror/MirrorClient.java