private Map<String, Map<TopicPartition, Checkpoint>> readCheckpoints() {
        Map<String, Map<TopicPartition, Checkpoint>> checkpoints = new HashMap<>();
        Callback<ConsumerRecord<byte[], byte[]>> consumedCallback = (error, cpRecord) -> {
            if (error != null) {
                // if there is no authorization to READ from the topic, we must throw an error
                // to stop the KafkaBasedLog forever looping attempting to read to end
                checkpoints.clear();
                if (error instanceof RuntimeException) {
                    throw (RuntimeException) error;
                } else {
                    throw new RuntimeException(error);
                }
            } else {
                try {
                    Checkpoint cp = Checkpoint.deserializeRecord(cpRecord);
                    if (consumerGroups.contains(cp.consumerGroupId())) {
                        Map<TopicPartition, Checkpoint> cps = checkpoints.computeIfAbsent(cp.consumerGroupId(), ignored1 -> new HashMap<>());
                        cps.put(cp.topicPartition(), cp);
                    }
                } catch (SchemaException ex) {
                    log.warn("Ignored invalid checkpoint record at offset {}", cpRecord.offset(), ex);
                }
            }
        };

        try {
            long startTime = System.currentTimeMillis();
            readCheckpointsImpl(config, consumedCallback);
            log.debug("starting+stopping KafkaBasedLog took {}ms", System.currentTimeMillis() - startTime);
            loadSuccess = true;
        } catch (Exception error) {
            loadSuccess = false;
            if (error instanceof AuthorizationException) {
                log.warn("Not authorized to access checkpoints topic {} - " +
                        "this may degrade offset translation as only checkpoints " +
                        "for offsets which were mirrored after the task started will be emitted",
                        config.checkpointsTopic(), error);
            } else {
                log.info("Exception encountered loading checkpoints topic {} - " +
                        "this may degrade offset translation as only checkpoints " +
                        "for offsets which were mirrored after the task started will be emitted",
                        config.checkpointsTopic(), error);
            }
        }
        return checkpoints;
    }
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