private VoteResponseData handleVoteRequest(
        RaftRequest.Inbound requestMetadata
    ) {
        VoteRequestData request = (VoteRequestData) requestMetadata.data();

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

        if (!hasValidTopicPartition(request, log.topicPartition())) {
            // Until we support multi-raft, we treat individual topic partition mismatches as invalid requests
            return new VoteResponseData().setErrorCode(Errors.INVALID_REQUEST.code());
        }

        VoteRequestData.PartitionData partitionRequest =
            request.topics().get(0).partitions().get(0);

        int candidateId = partitionRequest.candidateId();
        int candidateEpoch = partitionRequest.candidateEpoch();

        int lastEpoch = partitionRequest.lastOffsetEpoch();
        long lastEpochEndOffset = partitionRequest.lastOffset();
        if (lastEpochEndOffset < 0 || lastEpoch < 0 || lastEpoch >= candidateEpoch) {
            return buildVoteResponse(
                requestMetadata.listenerName(),
                requestMetadata.apiVersion(),
                Errors.INVALID_REQUEST,
                false
            );
        }

        Optional<Errors> errorOpt = validateVoterOnlyRequest(candidateId, candidateEpoch);
        if (errorOpt.isPresent()) {
            return buildVoteResponse(
                requestMetadata.listenerName(),
                requestMetadata.apiVersion(),
                errorOpt.get(),
                false
            );
        }

        if (candidateEpoch > quorum.epoch()) {
            transitionToUnattached(candidateEpoch);
        }

        // Check that the request was intended for this replica
        Optional<ReplicaKey> voterKey = RaftUtil.voteRequestVoterKey(request, partitionRequest);
        if (!isValidVoterKey(voterKey)) {
            logger.info(
                "Candidate sent a voter key ({}) in the VOTE request that doesn't match the " +
                "local key ({}, {}); rejecting the vote",
                voterKey,
                nodeId,
                nodeDirectoryId
            );
            // The request is not intended to this replica since the replica keys don't match
            return buildVoteResponse(
                requestMetadata.listenerName(),
                requestMetadata.apiVersion(),
                Errors.INVALID_VOTER_KEY,
                false
            );
        }

        OffsetAndEpoch lastEpochEndOffsetAndEpoch = new OffsetAndEpoch(lastEpochEndOffset, lastEpoch);
        ReplicaKey candidateKey = ReplicaKey.of(
            candidateId,
            partitionRequest.candidateDirectoryId()
        );
        boolean voteGranted = quorum.canGrantVote(
            candidateKey,
            lastEpochEndOffsetAndEpoch.compareTo(endOffset()) >= 0
        );

        if (voteGranted && quorum.isUnattached()) {
            transitionToVoted(candidateKey, candidateEpoch);
        }

        logger.info("Vote request {} with epoch {} is {}", request, candidateEpoch, voteGranted ? "granted" : "rejected");
        return buildVoteResponse(
            requestMetadata.listenerName(),
            requestMetadata.apiVersion(),
            Errors.NONE,
            voteGranted
        );
    }
Handle a Vote request. 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 last epoch or offset are invalid
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java