private void maybeLoadSnapshot() {
        if ((nextOffset == STARTING_NEXT_OFFSET || nextOffset < log.startOffset()) && log.latestSnapshot().isPresent()) {
            RawSnapshotReader rawSnapshot = log.latestSnapshot().get();
            // Clear the current state
            synchronized (kraftVersionHistory) {
                kraftVersionHistory.clear();
            }
            synchronized (voterSetHistory) {
                voterSetHistory.clear();
            }

            // Load the snapshot since the listener is at the start of the log or the log doesn't have the next entry.
            try (SnapshotReader<?> reader = RecordsSnapshotReader.of(
                    rawSnapshot,
                    serde,
                    bufferSupplier,
                    maxBatchSizeBytes,
                    true // Validate batch CRC
                )
            ) {
                logger.info(
                    "Loading snapshot ({}) since log start offset ({}) is greater than the internal listener's next offset ({})",
                    reader.snapshotId(),
                    log.startOffset(),
                    nextOffset
                );
                OptionalLong currentOffset = OptionalLong.of(reader.lastContainedLogOffset());
                while (reader.hasNext()) {
                    Batch<?> batch = reader.next();
                    handleBatch(batch, currentOffset);
                }

                nextOffset = reader.lastContainedLogOffset() + 1;
            }
        } else if (nextOffset == STARTING_NEXT_OFFSET) {
            // Listener just started and there are no snapshots; set the nextOffset to
            // 0 to start reading the log
            nextOffset = SMALLEST_LOG_OFFSET;
        }
    }
Returns the finalized kraft version at a given offset.

@param offset the offset (inclusive)
@return the finalized kraft version if one exist, otherwise 0
raft/src/main/java/org/apache/kafka/raft/internals/KRaftControlRecordStateMachine.java