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

Code before:
            throw new IllegalStateException(
                String.format(
                    "Cannot add voted key (%s) to current state (%s) in epoch %d since it matches the local " +
                    "broker.id",
                    candidateKey,
                    state,
                    epoch
                )
            );
        } else if (localId.isEmpty()) {
            throw new IllegalStateException("Cannot add voted state without a replica id");
        } else if (epoch != currentEpoch || !isProspectiveNotVoted()) {
            throw new IllegalStateException(
                String.format(
                    "Cannot add voted key (%s) to current state (%s) in epoch %d",
                    candidateKey,
                    state,
                    epoch
                )
            );
        }

        // Note that we reset the election timeout after voting for a candidate because we
        // know that the candidate has at least as good of a chance of getting elected as us
        durableTransitionTo(
            new ProspectiveState(
                time,
                localIdOrThrow(),
                epoch,
                state.election().optionalLeaderId(),
                state.leaderEndpoints(),
                Optional.of(candidateKey),
                partitionState.lastVoterSet(),
                state.highWatermark(),
                randomElectionTimeoutMs(),
                logContext
            )
        );
    }

    /**
     * Become a follower of an elected leader so that we can begin fetching.
     */
    public void transitionToFollower(int epoch, int leaderId, Endpoints endpoints) {
        int currentEpoch = state.epoch();
        if (endpoints.isEmpty()) {
            throw new IllegalArgumentException(
                String.format(
                    "Cannot transition to Follower with leader %s and epoch %s without a leader endpoint",
                    leaderId,
                    epoch

Review comment (bug):
Outside the scope of this PR but this is not true when there is a network partition. We should not reset the election timer if a vote is granted. The worst case, if the replica gets this wrong, is that the replica transitions back to the follower or unattached state.

Code after:
        int currentEpoch = state.epoch();
        if (localId.isPresent() && candidateKey.id() == localId.getAsInt()) {
            throw new IllegalStateException(
                String.format(
                    "Cannot add voted key (%s) to current state (%s) in epoch %d since it matches the local " +
                    "broker.id",
                    candidateKey,
                    state,
                    epoch
                )
            );
        } else if (localId.isEmpty()) {
            throw new IllegalStateException("Cannot add voted state without a replica id");
        } else if (epoch != currentEpoch || !isProspectiveNotVoted()) {
            throw new IllegalStateException(
                String.format(
                    "Cannot add voted key (%s) to current state (%s) in epoch %d",
                    candidateKey,
                    state,
                    epoch
                )
            );
        }

        // Note that we reset the election timeout after voting for a candidate because we
        // know that the candidate has at least as good of a chance of getting elected as us
        durableTransitionTo(
            new ProspectiveState(
                time,
                localIdOrThrow(),
                epoch,
                state.election().optionalLeaderId(),
                state.leaderEndpoints(),
                Optional.of(candidateKey),
                partitionState.lastVoterSet(),
                state.highWatermark(),
                randomElectionTimeoutMs(),
                logContext
            )
        );
    }

    /**
     * Become a follower of an elected leader so that we can begin fetching.
     */
    public void transitionToFollower(int epoch, int leaderId, Endpoints endpoints) {
        int currentEpoch = state.epoch();
        if (endpoints.isEmpty()) {
            throw new IllegalArgumentException(
                String.format(
                    "Cannot transition to Follower with leader %s and epoch %s without a leader endpoint",