private FetchSnapshotResponseData handleFetchSnapshotRequest(
        RaftRequest.Inbound requestMetadata,
        long currentTimeMs
    ) {
        FetchSnapshotRequestData data = (FetchSnapshotRequestData) requestMetadata.data();

        if (!hasValidClusterId(data.clusterId())) {
            return new FetchSnapshotResponseData().setErrorCode(Errors.INCONSISTENT_CLUSTER_ID.code());
        }

        if (data.topics().size() != 1 && data.topics().get(0).partitions().size() != 1) {
            return FetchSnapshotResponse.withTopLevelError(Errors.INVALID_REQUEST);
        }

        Optional<FetchSnapshotRequestData.PartitionSnapshot> partitionSnapshotOpt = FetchSnapshotRequest
            .forTopicPartition(data, log.topicPartition());
        if (!partitionSnapshotOpt.isPresent()) {
            // The Raft client assumes that there is only one topic partition.
            TopicPartition unknownTopicPartition = new TopicPartition(
                data.topics().get(0).name(),
                data.topics().get(0).partitions().get(0).partition()
            );

            return RaftUtil.singletonFetchSnapshotResponse(
                requestMetadata.listenerName(),
                requestMetadata.apiVersion(),
                unknownTopicPartition,
                quorum.leaderIdOrSentinel(),
                quorum.leaderEndpoints(),
                responsePartitionSnapshot -> responsePartitionSnapshot
                    .setErrorCode(Errors.UNKNOWN_TOPIC_OR_PARTITION.code())
            );
        }

        FetchSnapshotRequestData.PartitionSnapshot partitionSnapshot = partitionSnapshotOpt.get();
        Optional<Errors> leaderValidation = validateLeaderOnlyRequest(
            partitionSnapshot.currentLeaderEpoch()
        );
        if (leaderValidation.isPresent()) {
            return RaftUtil.singletonFetchSnapshotResponse(
                requestMetadata.listenerName(),
                requestMetadata.apiVersion(),
                log.topicPartition(),
                quorum.leaderIdOrSentinel(),
                quorum.leaderEndpoints(),
                responsePartitionSnapshot -> addQuorumLeader(responsePartitionSnapshot)
                    .setErrorCode(leaderValidation.get().code())
            );
        }

        OffsetAndEpoch snapshotId = new OffsetAndEpoch(
            partitionSnapshot.snapshotId().endOffset(),
            partitionSnapshot.snapshotId().epoch()
        );

        Optional<RawSnapshotReader> snapshotOpt = log.readSnapshot(snapshotId);
        if (!snapshotOpt.isPresent() || snapshotId.equals(BOOTSTRAP_SNAPSHOT_ID)) {
            // The bootstrap checkpoint should not be replicated. The first leader will
            // make sure that the content of the bootstrap checkpoint is included in the
            // partition log
            return RaftUtil.singletonFetchSnapshotResponse(
                requestMetadata.listenerName(),
                requestMetadata.apiVersion(),
                log.topicPartition(),
                quorum.leaderIdOrSentinel(),
                quorum.leaderEndpoints(),
                responsePartitionSnapshot -> addQuorumLeader(responsePartitionSnapshot)
                    .setErrorCode(Errors.SNAPSHOT_NOT_FOUND.code())
            );
        }

        RawSnapshotReader snapshot = snapshotOpt.get();
        long snapshotSize = snapshot.sizeInBytes();
        if (partitionSnapshot.position() < 0 || partitionSnapshot.position() >= snapshotSize) {
            return RaftUtil.singletonFetchSnapshotResponse(
                requestMetadata.listenerName(),
                requestMetadata.apiVersion(),
                log.topicPartition(),
                quorum.leaderIdOrSentinel(),
                quorum.leaderEndpoints(),
                responsePartitionSnapshot -> addQuorumLeader(responsePartitionSnapshot)
                    .setErrorCode(Errors.POSITION_OUT_OF_RANGE.code())
            );
        }

        if (partitionSnapshot.position() > Integer.MAX_VALUE) {
            throw new IllegalStateException(
                String.format(
                    "Trying to fetch a snapshot with size (%d) and a position (%d) larger than %d",
                    snapshotSize,
                    partitionSnapshot.position(),
                    Integer.MAX_VALUE
                )
            );
        }

        int maxSnapshotSize;
        try {
            maxSnapshotSize = Math.toIntExact(snapshotSize);
        } catch (ArithmeticException e) {
            maxSnapshotSize = Integer.MAX_VALUE;
        }

        UnalignedRecords records = snapshot.slice(partitionSnapshot.position(), Math.min(data.maxBytes(), maxSnapshotSize));

        LeaderState<T> state = quorum.leaderStateOrThrow();
        state.updateCheckQuorumForFollowingVoter(
            ReplicaKey.of(
                data.replicaId(),
                partitionSnapshot.replicaDirectoryId()
            ),
            currentTimeMs
        );

        return RaftUtil.singletonFetchSnapshotResponse(
            requestMetadata.listenerName(),
            requestMetadata.apiVersion(),
            log.topicPartition(),
            quorum.leaderIdOrSentinel(),
            quorum.leaderEndpoints(),
            responsePartitionSnapshot -> {
                addQuorumLeader(responsePartitionSnapshot)
                    .snapshotId()
                    .setEndOffset(snapshotId.offset())
                    .setEpoch(snapshotId.epoch());

                return responsePartitionSnapshot
                    .setSize(snapshotSize)
                    .setPosition(partitionSnapshot.position())
                    .setUnalignedRecords(records);
            }
        );
    }
Handle a FetchSnapshot request, similar to the Fetch request but we use {@link UnalignedRecords}
in response because the records are not necessarily offset-aligned.

This API may return the following errors:

- {@link Errors#INCONSISTENT_CLUSTER_ID} if the cluster id is presented in request
    but different from this node
- {@link Errors#BROKER_NOT_AVAILABLE} if this node is currently shutting down
- {@link Errors#FENCED_LEADER_EPOCH} if the epoch is smaller than this node's epoch
- {@link Errors#INVALID_REQUEST} if the request epoch is larger than the leader's current epoch
    or if either the fetch offset or the last fetched epoch is invalid
- {@link Errors#SNAPSHOT_NOT_FOUND} if the request snapshot id does not exists
- {@link Errors#POSITION_OUT_OF_RANGE} if the request snapshot offset out of range
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java