void readCheckpointsImpl(MirrorCheckpointTaskConfig config, Callback<ConsumerRecord<byte[], byte[]>> consumedCallback) {
        try {
            cpAdmin = new TopicAdmin(
                    config.targetAdminConfig("checkpoint-target-admin"),
                    config.forwardingAdmin(config.targetAdminConfig("checkpoint-target-admin")));

            backingStore = KafkaBasedLog.withExistingClients(
                    config.checkpointsTopic(),
                    MirrorUtils.newConsumer(config.targetConsumerConfig(CHECKPOINTS_TARGET_CONSUMER_ROLE)),
                    null,
                    cpAdmin,
                    consumedCallback,
                    Time.SYSTEM,
                    ignored -> {
                    },
                    topicPartition -> topicPartition.partition() == 0);

            backingStore.start(true);
            backingStore.stop();
        } finally {
            releaseResources();
        }
    }
Reads once the Kafka log for checkpoints and populates a map of
checkpoints per consumer group.

The Kafka log is closed after the initial load and only the in memory map is
used after start.
connect/mirror/src/main/java/org/apache/kafka/connect/mirror/CheckpointStore.java