private boolean handleVoteResponse(
        RaftResponse.Inbound responseMetadata,
        long currentTimeMs
    ) {
        int remoteNodeId = responseMetadata.source().id();
        VoteResponseData response = (VoteResponseData) responseMetadata.data();
        Errors topLevelError = Errors.forCode(response.errorCode());
        if (topLevelError != Errors.NONE) {
            return handleTopLevelError(topLevelError, responseMetadata);
        }

        if (!hasValidTopicPartition(response, log.topicPartition())) {
            return false;
        }

        VoteResponseData.PartitionData partitionResponse =
            response.topics().get(0).partitions().get(0);

        Errors error = Errors.forCode(partitionResponse.errorCode());
        OptionalInt responseLeaderId = optionalLeaderId(partitionResponse.leaderId());
        int responseEpoch = partitionResponse.leaderEpoch();

        final Endpoints leaderEndpoints;
        if (responseLeaderId.isPresent()) {
            if (response.nodeEndpoints().isEmpty()) {
                leaderEndpoints = partitionState.lastVoterSet().listeners(responseLeaderId.getAsInt());
            } else {
                leaderEndpoints = Endpoints.fromVoteResponse(
                    channel.listenerName(),
                    responseLeaderId.getAsInt(),
                    response.nodeEndpoints()
                );
            }
        } else {
            leaderEndpoints = Endpoints.empty();
        }

        Optional<Boolean> handled = maybeHandleCommonResponse(
            error,
            responseLeaderId,
            responseEpoch,
            leaderEndpoints,
            responseMetadata.source(),
            currentTimeMs
        );
        if (handled.isPresent()) {
            return handled.get();
        } else if (error == Errors.NONE) {
            if (quorum.isLeader()) {
                logger.debug("Ignoring vote response {} since we already became leader for epoch {}",
                    partitionResponse, quorum.epoch());
            } else if (quorum.isCandidate()) {
                CandidateState state = quorum.candidateStateOrThrow();
                if (partitionResponse.voteGranted()) {
                    state.recordGrantedVote(remoteNodeId);
                    maybeTransitionToLeader(state, currentTimeMs);
                } else {
                    state.recordRejectedVote(remoteNodeId);

                    // If our vote is rejected, we go immediately to the random backoff. This
                    // ensures that we are not stuck waiting for the election timeout when the
                    // vote has become gridlocked.
                    if (state.isVoteRejected() && !state.isBackingOff()) {
                        logger.info("Insufficient remaining votes to become leader (rejected by {}). " +
                            "We will backoff before retrying election again", state.rejectingVoters());

                        state.startBackingOff(
                            currentTimeMs,
                            binaryExponentialElectionBackoffMs(state.retries())
                        );
                    }
                }
            } else {
                logger.debug("Ignoring vote response {} since we are no longer a candidate in epoch {}",
                    partitionResponse, quorum.epoch());
            }
            return true;
        } else {
            return handleUnexpectedError(error, responseMetadata);
        }
    }
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