int countHopsForTopic(String topic, String sourceClusterAlias) {
        int hops = 0;
        Set<String> visited = new HashSet<>();
        while (true) {
            hops++;
            String source = replicationPolicy.topicSource(topic);
            if (source == null) {
                return -1;
            }
            if (source.equals(sourceClusterAlias)) {
                return hops;
            }
            if (visited.contains(source)) {
                // Extra check for IdentityReplicationPolicy and similar impls that cannot prevent cycles.
                // We assume we're stuck in a cycle and will never find sourceClusterAlias.
                return -1;
            }
            visited.add(source);
            topic = replicationPolicy.upstreamTopic(topic);
        }
    }
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