Repository: apache/kafka
File: raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java (Java)

Code before:
                    partitionResponse,
                    quorum.epoch()
                );
            }
            return true;
        } else {
            return handleUnexpectedError(error, responseMetadata);
        }
    }

    /**
     * On election loss, if replica is prospective it will transition to unattached or follower state.
     * If replica is candidate, it will start backing off.
     */
    private void maybeHandleElectionLoss(NomineeState state, long currentTimeMs) {
        if (state instanceof CandidateState candidate) {
            if (candidate.epochElection().isVoteRejected() && !candidate.isBackingOff()) {
                logger.info(
                    "Insufficient remaining votes to become leader. Candidate will wait the remaining election " +
                        "timeout ({}) before transitioning back to Prospective. Current epoch election state is {}.",
                    candidate.remainingElectionTimeMs(currentTimeMs),
                    candidate.epochElection()
                );
                // Mark that the candidate is now backing off. After election timeout expires the candidate will
                // transition back to prospective
                candidate.startBackingOff();
            }
        } else if (state instanceof ProspectiveState prospective) {
            if (prospective.epochElection().isVoteRejected()) {
                logger.info(
                    "Insufficient remaining votes to become candidate. Current epoch election state is {}. ",
                    prospective.epochElection()
                );
                prospectiveTransitionAfterElectionLoss(prospective, currentTimeMs);
            }
        } else {
            throw new IllegalStateException(
                "Expected to be a NomineeState (Prospective or Candidate), but current state is " + state
            );
        }
    }

    private int strictExponentialElectionBackoffMs(int positionInSuccessors, int totalNumSuccessors) {
        if (positionInSuccessors == 0) {
            return 0;
        } else if (positionInSuccessors < 0 || positionInSuccessors >= totalNumSuccessors) {
            return quorumConfig.electionBackoffMaxMs();
        }

        int retryBackOffBaseMs = quorumConfig.electionBackoffMaxMs() >> (totalNumSuccessors - 1);
        return Math.min(quorumConfig.electionBackoffMaxMs(), retryBackOffBaseMs << (positionInSuccessors - 1));

Review comment (refactor):
It looks like we recored the `isBackingOff` state simply so we can log this message at most once per epoch. Maybe we can remove this state if we change `EpochElection#recordVote` so that it returns true if the vote resulted in the election getting granted or rejected.

The other option is to log this message multiple times. Once for each vote response after the election was lost. I am okay with this since election should be infrequent and election are throttle by the election timeout.

Code after:
                    partitionResponse,
                    quorum.epoch()
                );
            }
            return true;
        } else {
            return handleUnexpectedError(error, responseMetadata);
        }
    }

    /**
     * On election loss, if replica is prospective it will transition to unattached or follower state.
     * If replica is candidate, it will start backing off.
     */
    private void maybeHandleElectionLoss(NomineeState state, long currentTimeMs) {
        if (state instanceof CandidateState candidate) {
            if (candidate.epochElection().isVoteRejected()) {
                logger.info(
                    "Insufficient remaining votes to become leader. Candidate will wait the remaining election " +
                        "timeout ({}) before transitioning back to Prospective. Current epoch election state is {}.",
                    candidate.remainingElectionTimeMs(currentTimeMs),
                    candidate.epochElection()
                );
            }
        } else if (state instanceof ProspectiveState prospective) {
            if (prospective.epochElection().isVoteRejected()) {
                logger.info(
                    "Insufficient remaining votes to become candidate. Current epoch election state is {}. ",
                    prospective.epochElection()
                );
                prospectiveTransitionAfterElectionLoss(prospective, currentTimeMs);
            }
        } else {
            throw new IllegalStateException(
                "Expected to be a NomineeState (Prospective or Candidate), but current state is " + state
            );
        }
    }

    private int strictExponentialElectionBackoffMs(int positionInSuccessors, int totalNumSuccessors) {
        if (positionInSuccessors == 0) {
            return 0;
        } else if (positionInSuccessors < 0 || positionInSuccessors >= totalNumSuccessors) {
            return quorumConfig.electionBackoffMaxMs();
        }

        int retryBackOffBaseMs = quorumConfig.electionBackoffMaxMs() >> (totalNumSuccessors - 1);
        return Math.min(quorumConfig.electionBackoffMaxMs(), retryBackOffBaseMs << (positionInSuccessors - 1));
    }

    private BeginQuorumEpochResponseData buildBeginQuorumEpochResponse(