public Map<TopicPartition, OffsetAndMetadata> remoteConsumerOffsets(String consumerGroupId,
            String remoteClusterAlias, Duration timeout) {
        long deadline = System.currentTimeMillis() + timeout.toMillis();
        Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();

        try (KafkaConsumer<byte[], byte[]> consumer = new KafkaConsumer<>(consumerConfig,
                new ByteArrayDeserializer(), new ByteArrayDeserializer())) {
            // checkpoint topics are not "remote topics", as they are not replicated. So we don't need
            // to use ReplicationPolicy to create the checkpoint topic here.
            String checkpointTopic = replicationPolicy.checkpointsTopic(remoteClusterAlias);
            List<TopicPartition> checkpointAssignment =
                Collections.singletonList(new TopicPartition(checkpointTopic, 0));
            consumer.assign(checkpointAssignment);
            consumer.seekToBeginning(checkpointAssignment);
            while (System.currentTimeMillis() < deadline && !endOfStream(consumer, checkpointAssignment)) {
                ConsumerRecords<byte[], byte[]> records = consumer.poll(timeout);
                for (ConsumerRecord<byte[], byte[]> record : records) {
                    try {
                        Checkpoint checkpoint = Checkpoint.deserializeRecord(record);
                        if (checkpoint.consumerGroupId().equals(consumerGroupId)) {
                            offsets.put(checkpoint.topicPartition(), checkpoint.offsetAndMetadata());
                        }
                    } catch (SchemaException e) {
                        log.info("Could not deserialize record. Skipping.", e);
                    }
                }
            }
            log.info("Consumed {} checkpoint records for {} from {}.", offsets.size(),
                consumerGroupId, checkpointTopic);
        }
        return offsets;
    }
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