private Optional<Boolean> maybeHandleCommonResponse(
        Errors error,
        OptionalInt leaderId,
        int epoch,
        Endpoints leaderEndpoints,
        Node source,
        long currentTimeMs
    ) {
        if (leaderEndpoints.isEmpty() && leaderId.isPresent()) {
            // The response didn't include the leader endpoints because it is from a replica
            // that doesn't support reconfiguration. Look up the leader endpoint in the
            // voter set.
            leaderEndpoints = partitionState
                .lastVoterSet()
                .listeners(leaderId.getAsInt());
        }

        if (epoch < quorum.epoch() || error == Errors.UNKNOWN_LEADER_EPOCH) {
            // We have a larger epoch, so the response is no longer relevant
            return Optional.of(true);
        } else if (epoch > quorum.epoch()
            || error == Errors.FENCED_LEADER_EPOCH
            || error == Errors.NOT_LEADER_OR_FOLLOWER) {

            // The response indicates that the request had a stale epoch, but we need
            // to validate the epoch from the response against our current state.
            maybeTransition(leaderId, epoch, leaderEndpoints, currentTimeMs);
            return Optional.of(true);
        } else if (epoch == quorum.epoch()
            && leaderId.isPresent()
            && !quorum.hasLeader()) {

            // Since we are transitioning to Follower, we will only forward the
            // request to the handler if there is no error. Otherwise, we will let
            // the request be retried immediately (if needed) after the transition.
            // This handling allows an observer to discover the leader and append
            // to the log in the same Fetch request.
            transitionToFollower(epoch, leaderId.getAsInt(), leaderEndpoints, currentTimeMs);
            if (error == Errors.NONE) {
                return Optional.empty();
            } else {
                return Optional.of(true);
            }
        } else if (error == Errors.BROKER_NOT_AVAILABLE) {
            return Optional.of(false);
        } else if (error == Errors.INVALID_VOTER_KEY) {
            // The voter key in the request for VOTE and BEGIN_QUORUM_EPOCH doesn't match the
            // receiver's replica key
            logger.info(
                "Voter key for VOTE or BEGIN_QUORUM_EPOCH request didn't match the receiver's " +
                "replica key: {}",
                source
            );
            return Optional.of(true);
        } else if (error == Errors.INVALID_REQUEST) {
            throw new IllegalStateException("Received unexpected invalid request error");
        }

        return Optional.empty();
    }
Handle response errors that are common across request types.

@param error Error from the received response
@param leaderId Optional leaderId from the response
@param epoch Epoch received from the response
@param leaderEndpoints the endpoints of the leader from the response
@param source the node that sent the response
@param currentTimeMs Current epoch time in milliseconds
@return Optional value indicating whether the error was handled here and the outcome of
   that handling. Specifically:

   - Optional.empty means that the response was not handled here and the custom
       API handler should be applied
   - Optional.of(true) indicates that the response was successfully handled here and
       the node can become ready
   - Optional.of(false) indicates that the response was handled here, but that the
       node should got in to backoff
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java