private boolean hasConsistentLeader(int epoch, OptionalInt leaderId) {
        // Only elected leaders are sent in the request/response header, so if we have an elected
        // leaderId, it should be consistent with what is in the message.
        if (leaderId.isPresent() && leaderId.getAsInt() == quorum.localIdOrSentinel()) {
            // The response indicates that we should be the leader, so we verify that is the case
            return quorum.isLeader();
        } else {
            return epoch != quorum.epoch()
                || !leaderId.isPresent()
                || !quorum.leaderId().isPresent()
                || leaderId.equals(quorum.leaderId());
        }
    }
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