Set<String> allSources(String topic) {
        Set<String> sources = new HashSet<>();
        String source = replicationPolicy.topicSource(topic);
        while (source != null && !sources.contains(source)) {
            // The extra Set.contains above is for ReplicationPolicies that cannot prevent cycles.
            sources.add(source);
            topic = replicationPolicy.upstreamTopic(topic);
            source = replicationPolicy.topicSource(topic);
        }
        return sources;
    }
Translates a remote consumer group's offsets into corresponding local offsets. Topics are automatically
renamed according to the ReplicationPolicy.
@param consumerGroupId The group ID of remote consumer group
@param remoteClusterAlias The alias of remote cluster
@param timeout The maximum time to block when consuming from the checkpoints topic
connect/mirror-client/src/main/java/org/apache/kafka/connect/mirror/MirrorClient.java