KAFKA-17642: PreVote response handling and ProspectiveState #18240

ahuang98 · 2024-12-18T00:02:14Z

KIP: https://cwiki.apache.org/confluence/display/KAFKA/KIP-996%3A+Pre-Vote
Jira: https://issues.apache.org/jira/browse/KAFKA-16164
Implements items 2-4 which cover response handling and new ProspectiveState:

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

jsancio

Thank @ahuang98 . Here is a partial review.

raft/src/main/java/org/apache/kafka/raft/EpochElection.java

raft/src/main/java/org/apache/kafka/raft/ProspectiveState.java

raft/src/main/java/org/apache/kafka/raft/NomineeState.java

raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java

raft/src/main/java/org/apache/kafka/raft/QuorumState.java

raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java

raft/src/main/java/org/apache/kafka/raft/QuorumState.java

jsancio

Thanks for the changes @ahuang98 . Quick review of the changes to **/src/main.

raft/src/main/java/org/apache/kafka/raft/CandidateState.java

raft/src/main/java/org/apache/kafka/raft/UnattachedState.java

raft/src/main/java/org/apache/kafka/raft/internals/EpochElection.java

raft/src/main/java/org/apache/kafka/raft/internals/KafkaRaftMetrics.java

ahuang98 · 2024-12-30T07:39:52Z

raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java

+//            if (quorum.isVoter()) {
+                // canElectNewLeaderAfterOldLeaderPartitioned fails if we do not bump epoch since it is possible
+                // that the replica ends up as follower in the same epoch.
+                // resigned(leaderId=local) -> prospective(leaderId=local) -> follower(leaderId=local) which is illegal
+//                transitionToProspective(quorum.epoch() + 1, currentTimeMs);
+//                transitionToCandidate(currentTimeMs);
+//            } else {


the existing raft event simulation tests picked up on a new bug in pollResigned - if we simply replace the transitionToCandidate(currentTimeMs) with transitionToProspective(currentTimeMs), a cordoned leader in epoch 5 could resign in epoch 5, transition to prospective in epoch 5 (with leaderId=localId), fail election and then attempt to become follower of itself in epoch 5.

so far, these are the alternatives which seem reasonable to me:

resigned voter in epoch X should transition to prospective in epoch X+1

cons: need to create a special code path just for this case to allow becoming prospective in epoch+1 (would also add trivial complexity for determining if votedKey or leaderId should be kept from prior transition). transitioning to prospective in epoch + 1 is almost as disruptive as transitioning directly to candidate since it involves an epoch bump

pro: probably the option which follows intentions of past logic most closely

resigned voter in epoch X should simply transition to unattached in epoch X+1 (current version)

con: resigned replica has to wait two election timeouts after resignation to become prospective

pro: simplified logic. unless this is the only replica eligible for leadership in the quorum (e.g. due to network partitioning), the impact of waiting two election timeouts after resignation is small - all other replicas should be starting their own elections within a single fetch timeout/election timeout

resigned voter in epoch X instead waits a smaller backoffTimeMs before transitioning to unattached in epoch X+1

con: scope creep - what should this backoff be? additional changes to resignedState

pro: resigned voter waits less time before becoming eligible to start a new election.

the existing raft event simulation tests picked up on a new bug in pollResigned

What is the exact error? Let's add an unittest to one of the KafkaRaftClient*Test suite that shows the bug.

attempt to become follower of itself in epoch 5.

Let's add a check to transtitionToFollower that checks that leaderId is not equal to localId.

It makes sense to me that after the resign state the replica should always increase its epoch. The replica resigned from leadership at epoch X so eventually the epoch will be at least X + 1. Did you consider transitioning to candidate and relaxing the transition functions to allow both resigned and prospective to transition to candidate?

discussed offline, transitionToUnattached has existing logic for assigning election timeouts which we can borrow - we can just add an additional if clause that if we came from resignedState, assign electionTimeout to resignedState.electionTimeout which is effectively 0

jsancio

@ahuang98 review to reply to your comments.

raft/src/main/java/org/apache/kafka/raft/QuorumState.java

raft/src/main/java/org/apache/kafka/raft/UnattachedState.java

jsancio · 2024-12-30T15:47:39Z

raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java

+//            if (quorum.isVoter()) {
+                // canElectNewLeaderAfterOldLeaderPartitioned fails if we do not bump epoch since it is possible
+                // that the replica ends up as follower in the same epoch.
+                // resigned(leaderId=local) -> prospective(leaderId=local) -> follower(leaderId=local) which is illegal
+//                transitionToProspective(quorum.epoch() + 1, currentTimeMs);
+//                transitionToCandidate(currentTimeMs);
+//            } else {


the existing raft event simulation tests picked up on a new bug in pollResigned

What is the exact error? Let's add an unittest to one of the KafkaRaftClient*Test suite that shows the bug.

attempt to become follower of itself in epoch 5.

Let's add a check to transtitionToFollower that checks that leaderId is not equal to localId.

It makes sense to me that after the resign state the replica should always increase its epoch. The replica resigned from leadership at epoch X so eventually the epoch will be at least X + 1. Did you consider transitioning to candidate and relaxing the transition functions to allow both resigned and prospective to transition to candidate?

ahuang98 · 2024-12-30T17:29:24Z

It makes sense to me that after the resign state the replica should always increase its epoch. The replica resigned from leadership at epoch X so eventually the epoch will be at least X + 1. Did you consider transitioning to candidate and relaxing the transition functions to allow both resigned and prospective to transition to candidate?

yes, I decided not to list that as an option because I felt it was equal to if not worse than the option of having resigned transition to prospective in epoch X + 1. personally I felt it was nicer to have less edge cases to the invariant that only prospective should transition to candidate

jsancio

Reviewed the changes to src/main.

jsancio · 2024-12-30T20:34:54Z

raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java

+                    prospective.epochElection().rejectingVoters()
+                );
+                prospectiveTransitionAfterElectionLoss(prospective, currentTimeMs);
+            }
        }


Let's add an else case and throw an illegal state exception.

jsancio · 2024-12-30T20:38:21Z

raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java

+     * 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(long currentTimeMs) {


How about passing the NomineeState object, checking the subtype of that object and casting to the appropriate subtype.

Like the following?

private void maybeHandleElectionLoss(NomineeState state, long currentTimeMs) { if (state instanceof CandidateState) { CandidateState candidate = (CandidateState) state; ... else if (state instanceof ProspectiveState) { ProspectiveState prospective = (ProspectiveState) state; ...

Is the intention of the additional parameter to make it clear this method should be called on NomineeState? This seems a bit redundant with the existing QuorumState helpers (e.g. isCandidate() and candidateStateOrThrow()).

Yes. We discussed this offline.

jsancio · 2024-12-30T20:44:33Z

raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java

+            if (prospective.epochElection().isVoteRejected()) {
+                logger.info(
+                    "Insufficient remaining votes to become candidate (rejected by {}). ",
+                    prospective.epochElection().rejectingVoters()


How about just printing the entire epochElection? It may be useful to know the state of the entire voter set not just the rejecting voters.

jsancio · 2024-12-30T20:45:25Z

raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java

+                logger.info(
+                    "Insufficient remaining votes to become leader (rejected by {}). " +
+                    "We will backoff before retrying election again",
+                    candidate.epochElection().rejectingVoters()


How about just printing the entire epochElection? It may be useful to know the state of the entire voter set not just the rejecting voters.

raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java

raft/src/main/java/org/apache/kafka/raft/QuorumState.java

raft/src/main/java/org/apache/kafka/raft/internals/EpochElection.java

jsancio · 2025-01-01T21:06:15Z

raft/src/main/java/org/apache/kafka/raft/internals/EpochElection.java

+        @Override
+        public String toString() {
+            return String.format(
+                "VoterState(%s, state=%s)",


"VoterState(replicaKey=%s, state=%s)",

ReplicaKey's toString method contains the class name so I didn't want to be redundant - String.format("ReplicaKey(id=%d, directoryId=%s)", id, directoryId);

jsancio · 2025-01-01T21:08:36Z

raft/src/main/java/org/apache/kafka/raft/internals/EpochElection.java

+        return String.format(
+            "EpochElection(%s)",
+            voterStates.values().stream()
+                .map(VoterState::toString)
+                .collect(
+                    Collectors.joining(", "))
+        );


Why not just print the map?

return String.format( "EpochElection(voterStates=%s)", voterStates );

it felt redundant to print the keys given that the replica ids are also contained in the values. since this is would only be used for debugging though, I'll take your suggestion and just print the entire map

jsancio

Quick feedback.

jsancio · 2025-01-02T14:48:09Z

raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java

@@ -667,7 +668,8 @@ private boolean maybeTransitionToLeader(CandidateState state, long currentTimeMs
    }

    private boolean maybeTransitionToCandidate(ProspectiveState state, long currentTimeMs) {
-        if (state.epochElection().isVoteGranted()) {
+        // If replica is the only voter, it should transition to candidate immediately
+        if (state.epochElection().isVoteGranted() || quorum.isOnlyVoter()) {


If the quorum has a size of one and since the replica votes for itself when transitioning to prospective, isVoteGranted() should always return true. If so, the replica doesn't need to check if it is the only voter.

Let's confirm we have a test for this in KafkaRaftClientTest. If not, let's add a test.

Let's also confirm that we have a test for this in ProspectiveStateTest and CandidateStateTest. If not, let's add tests for these cases.

Added four tests for this, starting at testInitializeAsOnlyVoterWithEmptyElectionState

confirm that we have a test for this in ProspectiveStateTest and CandidateStateTest

Confirmed!

jsancio · 2025-01-02T14:51:23Z

raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java

-            transitionToCandidate(currentTimeMs);
+        // When there is only a single voter, become prospective immediately.
+        // transitionToProspective will handle short-circuiting voter to candidate state
+        if (quorum.isOnlyVoter() && !quorum.isNomineeState() && !quorum.isLeader()) {


Why do you check that is not leader? In KRaft a replica should never start as a leader. KRaft throws and illegal state exception if it starts as leader. See line 545 above.

if (quorum.isLeader()) { throw new IllegalStateException("Voter cannot initialize as a Leader");

Discussed offline, technically the replica can transition to leader due to the above conditional.
We can improve this conditional by directly checking if the replica is Unattached or Follower, and merge this conditional into the above conditional

ahuang98 · 2025-01-03T19:49:45Z

raft/src/test/java/org/apache/kafka/raft/QuorumStateTest.java

@@ -154,6 +155,163 @@ private ReplicaKey replicaKey(int id, boolean withDirectoryId) {
        return ReplicaKey.of(id, directoryId);
    }

+    @ParameterizedTest


I've organized QuorumStateTest in the following way - misc tests were pulled to the front. All other tests are organized under banners (e.g. Initialization tests, Tests of transitions from state X)

ahuang98 · 2025-01-03T19:51:26Z

raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientTest.java

@@ -302,7 +303,7 @@ public void testGrantVotesWhenShuttingDown(boolean withKip853Rpc) throws Excepti

    @ParameterizedTest
    @ValueSource(booleans = { true, false })
-    public void testInitializeAsResignedAndBecomeCandidate(boolean withKip853Rpc) throws Exception {


The diff is misleading here. This test was just removed because I found it was a duplicate of testInitializeAsResignedLeaderFromStateStore

ahuang98 · 2025-01-03T19:54:13Z

raft/src/test/java/org/apache/kafka/raft/internals/KafkaRaftMetricsTest.java

@@ -162,16 +218,18 @@ public void shouldRecordVoterQuorumState(KRaftVersion kraftVersion) {
            getMetric(metrics, "current-vote-directory-id").metricValue()
        );
        assertEquals((double) 1, getMetric(metrics, "current-epoch").metricValue());
-        assertEquals((double) -1L, getMetric(metrics, "high-watermark").metricValue());
+        assertEquals((double) -1L, getMetric(metrics, "high-watermark").metricValue()); // todo, bug fix


@jsancio the HW drops to -1L after candidate transitions to leader - if you agree this is a bug I'll file a Jira for this

ahuang98 added 8 commits December 17, 2024 15:55

initial commit

16217d0

still need to add more test cases to QuorumStateTest

c19bda5

match latest version of kip

8a28714

fix quorumstate description

dbb5501

fix compilation after rebase

98e64eb

add raftProtocol testing matrix, need to fix quorumstatetest next

cc6fe6a

wip, still needs more cleanup

1839d7c

interface changes

b8a4002

github-actions bot added triage PRs from the community core Kafka Broker kraft clients labels Dec 18, 2024

needs additional revisions to voted states

e621e78

jsancio reviewed Dec 18, 2024

View reviewed changes

github-actions bot removed the triage PRs from the community label Dec 19, 2024

ahuang98 added 2 commits December 18, 2024 23:32

addressing comments plus quorumstatetest

a213374

more tests

317026d

ahuang98 commented Dec 20, 2024

View reviewed changes

raft/src/main/java/org/apache/kafka/raft/QuorumState.java Show resolved Hide resolved

more comments and more test fixes

3097832

ahuang98 commented Dec 23, 2024

View reviewed changes

raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java Outdated Show resolved Hide resolved

some cleanup

b01aa11

ahuang98 marked this pull request as ready for review December 23, 2024 16:05

ahuang98 commented Dec 23, 2024

View reviewed changes

raft/src/main/java/org/apache/kafka/raft/QuorumState.java Show resolved Hide resolved

from latest discussions

f9d1812

jsancio reviewed Dec 24, 2024

View reviewed changes

jsancio added the ci-approved label Dec 24, 2024

ahuang98 added 2 commits December 28, 2024 23:48

addressing comments and non-simulation test failures

3268784

resigned edge case

1972b58

ahuang98 commented Dec 30, 2024

View reviewed changes

jsancio reviewed Dec 30, 2024

View reviewed changes

comments and test fixes

8e851ed

jsancio reviewed Jan 1, 2025

View reviewed changes

more src main change requests

8c68a9f

jsancio reviewed Jan 2, 2025

View reviewed changes

ahuang98 added 4 commits January 2, 2025 12:31

only voter logic

2de09bc

nits

d034c83

some test improvements, onlyvoter fix

578e4c3

more test revisions

2c32866

ahuang98 commented Jan 3, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-17642: PreVote response handling and ProspectiveState #18240

KAFKA-17642: PreVote response handling and ProspectiveState #18240

ahuang98 commented Dec 18, 2024

jsancio left a comment

jsancio left a comment

ahuang98 Dec 30, 2024 •

edited

Loading

jsancio Dec 30, 2024 •

edited

Loading

ahuang98 Dec 30, 2024

jsancio left a comment

jsancio Dec 30, 2024 •

edited

Loading

ahuang98 commented Dec 30, 2024

jsancio left a comment

jsancio Dec 30, 2024

jsancio Dec 30, 2024

ahuang98 Jan 2, 2025

jsancio Jan 2, 2025

jsancio Dec 30, 2024

jsancio Dec 30, 2024

jsancio Jan 1, 2025

ahuang98 Jan 2, 2025

jsancio Jan 1, 2025

ahuang98 Jan 2, 2025

jsancio left a comment

jsancio Jan 2, 2025

ahuang98 Jan 2, 2025 •

edited

Loading

jsancio Jan 2, 2025

ahuang98 Jan 2, 2025 •

edited

Loading

ahuang98 Jan 3, 2025

ahuang98 Jan 3, 2025

ahuang98 Jan 3, 2025

KAFKA-17642: PreVote response handling and ProspectiveState #18240

Are you sure you want to change the base?

KAFKA-17642: PreVote response handling and ProspectiveState #18240

Conversation

ahuang98 commented Dec 18, 2024

Committer Checklist (excluded from commit message)

jsancio left a comment

Choose a reason for hiding this comment

jsancio left a comment

Choose a reason for hiding this comment

ahuang98 Dec 30, 2024 • edited Loading

Choose a reason for hiding this comment

jsancio Dec 30, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jsancio left a comment

Choose a reason for hiding this comment

jsancio Dec 30, 2024 • edited Loading

Choose a reason for hiding this comment

ahuang98 commented Dec 30, 2024

jsancio left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jsancio left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ahuang98 Jan 2, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ahuang98 Jan 2, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ahuang98 Dec 30, 2024 •

edited

Loading

jsancio Dec 30, 2024 •

edited

Loading

jsancio Dec 30, 2024 •

edited

Loading

ahuang98 Jan 2, 2025 •

edited

Loading

ahuang98 Jan 2, 2025 •

edited

Loading