You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "sumanpal97 (via GitHub)" <gi...@apache.org> on 2026/08/15 17:28:04 UTC
[PR] KAFKA-20895: Grant pre-vote once the leader has resigned [kafka]
sumanpal97 opened a new pull request, #23168:
URL: https://github.com/apache/kafka/pull/23168
On a clean shutdown the leader sends `EndQuorumEpoch` carrying a list of preferred successors ordered by log end offset, and each successor waits a backoff proportional to its position before standing for election. The first entry, being the most up to date, is expected to win.
It cannot. `FollowerState.canGrantVote()` rejects a pre-vote whenever `hasFetchedFromLeader` is true, without accounting for the leader having already announced that it is resigning. With more than three voters the first successor collects only its own vote and the resigned leader's, which is not a majority, so it loses and returns to follower. The second entry then wins, because by that point the first entry has transitioned away from follower and is able to vote. The pattern repeats for larger quorums: the third entry wins with six or seven voters, the fourth with eight or nine. Every graceful controller shutdown therefore wastes an election round, which is the degraded availability described in the JIRA.
This change tracks whether the leader of the current epoch has resigned, and lets a follower grant a pre-vote in that case even if it has already fetched, as long as the requester's log is up to date. Granting is no longer disruptive once the leader has stepped down, so the disruptive-candidate protection introduced by KIP-996 is not weakened for a healthy leader. Standard votes are unaffected, since the new condition sits behind the existing `isPreVote` guard.
The flag lives on `FollowerState` and needs no explicit reset: `QuorumState` allocates a fresh `FollowerState` on every transition, so it is naturally scoped to one epoch and leader, exactly like the existing `hasFetchedFromLeader`. It is set in `handleEndQuorumEpochRequest` only when the replica is still a follower and the request came from its own leader.
### Testing strategy
`RaftEventSimulationTest.preferredSuccessorWinsElectionAfterGracefulLeaderShutdown` is a multi-node simulation that starts a quorum, waits for a leader and for every replica to converge on the same log so the resigning leader is able to grant pre-votes, reads the ranking the leader will advertise via `LeaderState.nonLeaderVotersByDescendingFetchOffset()`, shuts the leader down gracefully, and asserts that the first entry of that advertised ranking becomes the new leader. The ranking is read from the leader rather than hardcoded, so the assertion does not depend on which node happens to be elected first.
It is parameterized over four through nine voters. Without the production change it fails for all six sizes, and the winner lands at exactly the position the JIRA predicts: the second entry for four and five voters, the third for six and seven, the fourth for eight and nine. With the change all six elect the first entry, and the election completes in roughly 30 ms instead of waiting out a backoff.
Nine voters is the upper bound because `strictExponentialElectionBackoffMs` right-shifts the configured maximum by the number of successors, so beyond that the base backoff falls below the simulation's polling interval and the successors no longer stand for election one at a time. The test also raises `election.backoff.max.ms` to the production default via a new mutable field on the simulation's `Cluster`; the other simulations lower it to keep their runtime down, but here the gap between successive candidates is the behavior under test.
An `@ParameterizedTest` was used rather than the class's `withVotersObservers` helper because that helper randomizes both the seed and the voter count on every run, whereas this scenario needs a deterministic case per quorum size.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20895: Grant pre-vote once the leader has resigned [kafka]
Posted by "sumanpal97 (via GitHub)" <gi...@apache.org>.
sumanpal97 commented on code in PR #23168:
URL: https://github.com/apache/kafka/pull/23168#discussion_r3806288670
##########
raft/src/main/java/org/apache/kafka/raft/FollowerState.java:
##########
@@ -232,17 +244,18 @@ public void setFetchingSnapshot(Optional<RawSnapshotWriter> newSnapshot) {
@Override
public boolean canGrantVote(ReplicaKey replicaKey, boolean isLogUpToDate, boolean isPreVote) {
- if (isPreVote && !hasFetchedFromLeader && isLogUpToDate) {
+ if (isPreVote && (!hasFetchedFromLeader || hasLeaderResigned) && isLogUpToDate) {
Review Comment:
Sure thing. Added. Please check
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20895: Grant pre-vote once the leader has resigned [kafka]
Posted by "kevin-wu24 (via GitHub)" <gi...@apache.org>.
kevin-wu24 commented on code in PR #23168:
URL: https://github.com/apache/kafka/pull/23168#discussion_r4379411741
##########
raft/src/test/java/org/apache/kafka/raft/FollowerStateTest.java:
##########
@@ -122,6 +122,24 @@ public void testPreVoteAfterSuccessfulFetchFromLeader(boolean isLogUpToDate) {
assertFalse(state.canGrantVote(ReplicaKey.of(10, ReplicaKey.NO_DIRECTORY_ID), isLogUpToDate, false));
}
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testPreVoteAfterLeaderResigned(boolean isLogUpToDate) {
+ FollowerState state = newFollowerState(Set.of(1, 2, 3));
+ state.resetFetchTimeoutForSuccessfulFetch(time.milliseconds());
+ state.setHasLeaderResigned();
+
+ assertEquals(isLogUpToDate, state.canGrantVote(ReplicaKey.of(1, ReplicaKey.NO_DIRECTORY_ID), isLogUpToDate, true));
Review Comment:
```suggestion
assertEquals(
isLogUpToDate,
state.canGrantVote(ReplicaKey.of(1, ReplicaKey.NO_DIRECTORY_ID), isLogUpToDate, true)
);
```
There is a convention in the `raft` module to keep lines under 100 characters in length. This applies to many places in this PR.
##########
raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientPreVoteTest.java:
##########
@@ -670,6 +672,54 @@ public void testFollowerGrantsPreVoteIfHasNotFetchedYet(KRaftVersion kraftVersio
assertTrue(context.client.quorum().isFollower());
}
+ @Test
+ public void testFollowerGrantsPreVoteAfterLeaderResigns() throws Exception {
+ int localId = randomReplicaId();
+ ReplicaKey local = replicaKey(localId, true);
+ ReplicaKey leader = replicaKey(localId + 1, true);
+ ReplicaKey otherNodeKey = replicaKey(localId + 2, true);
+ int epoch = 2;
+
+ RaftClientTestContext context = new RaftClientTestContext.Builder(
+ local.id(),
+ local.directoryId().get()
+ )
+ .withStartingVoters(VoterSetTest.voterSet(Stream.of(local, leader, otherNodeKey)), KRaftVersion.KRAFT_VERSION_0)
+ .withElectedLeader(epoch, leader.id())
+ .withRaftProtocol(RaftProtocol.KIP_1186_PROTOCOL)
+ .build();
+
+ context.assertElectedLeader(epoch, leader.id());
+
+ context.pollUntilRequest();
+ RaftRequest.Outbound fetchRequest = context.assertSentFetchRequest();
+ context.assertFetchRequestData(fetchRequest, epoch, 0L, 0, context.client.highWatermark());
+ context.deliverResponse(
+ fetchRequest.correlationId(),
+ fetchRequest.destination(),
+ context.fetchResponse(epoch, leader.id(), MemoryRecords.EMPTY, 0L, Errors.NONE)
+ );
+
+ // After fetching successfully from the leader, follower rejects PreVotes even if the remote log is up to date
+ VoteRequestData preVote = context.preVoteRequest(epoch, otherNodeKey, epoch, 1);
+ context.deliverRequest(preVote);
+ context.pollUntilResponse();
+ context.assertSentVoteResponse(Errors.NONE, epoch, OptionalInt.of(leader.id()), false);
+ assertTrue(context.client.quorum().isFollower());
+
+ // Leader resigns by sending EndQuorumEpoch. This request can move the node out of the follower state.
Review Comment:
```suggestion
// Leader resigns by sending EndQuorumEpoch. The local node is not the top preferred
// successor and the clock doesn't advance, so it stays a follower
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20895: Grant pre-vote once the leader has resigned [kafka]
Posted by "kevin-wu24 (via GitHub)" <gi...@apache.org>.
kevin-wu24 commented on code in PR #23168:
URL: https://github.com/apache/kafka/pull/23168#discussion_r4349303050
##########
raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientPreVoteTest.java:
##########
@@ -670,6 +670,54 @@ public void testFollowerGrantsPreVoteIfHasNotFetchedYet(KRaftVersion kraftVersio
assertTrue(context.client.quorum().isFollower());
}
+ @ParameterizedTest
+ @EnumSource(value = KRaftVersion.class)
+ public void testFollowerGrantsPreVoteAfterLeaderResigns(KRaftVersion kraftVersion) throws Exception {
+ int localId = randomReplicaId();
+ ReplicaKey local = replicaKey(localId, true);
+ ReplicaKey leader = replicaKey(localId + 1, true);
+ ReplicaKey otherNodeKey = replicaKey(localId + 2, true);
+ int epoch = 2;
+
+ RaftClientTestContext context = new RaftClientTestContext.Builder(
+ local.id(),
+ local.directoryId().get()
+ )
+ .withStartingVoters(VoterSetTest.voterSet(Stream.of(local, leader, otherNodeKey)), kraftVersion)
+ .withElectedLeader(epoch, leader.id())
+ .withRaftProtocol(KIP_996_PROTOCOL)
Review Comment:
Please use the most recent value for `RaftProtocol`. That contains pre-vote too.
##########
raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientPreVoteTest.java:
##########
@@ -670,6 +670,54 @@ public void testFollowerGrantsPreVoteIfHasNotFetchedYet(KRaftVersion kraftVersio
assertTrue(context.client.quorum().isFollower());
}
+ @ParameterizedTest
+ @EnumSource(value = KRaftVersion.class)
+ public void testFollowerGrantsPreVoteAfterLeaderResigns(KRaftVersion kraftVersion) throws Exception {
+ int localId = randomReplicaId();
+ ReplicaKey local = replicaKey(localId, true);
+ ReplicaKey leader = replicaKey(localId + 1, true);
+ ReplicaKey otherNodeKey = replicaKey(localId + 2, true);
+ int epoch = 2;
+
+ RaftClientTestContext context = new RaftClientTestContext.Builder(
+ local.id(),
+ local.directoryId().get()
+ )
+ .withStartingVoters(VoterSetTest.voterSet(Stream.of(local, leader, otherNodeKey)), kraftVersion)
+ .withElectedLeader(epoch, leader.id())
+ .withRaftProtocol(KIP_996_PROTOCOL)
+ .build();
+
+ context.assertElectedLeader(epoch, leader.id());
+
+ context.pollUntilRequest();
+ RaftRequest.Outbound fetchRequest = context.assertSentFetchRequest();
+ context.assertFetchRequestData(fetchRequest, epoch, 0L, 0, context.client.highWatermark());
+ context.deliverResponse(
+ fetchRequest.correlationId(),
+ fetchRequest.destination(),
+ context.fetchResponse(epoch, leader.id(), MemoryRecords.EMPTY, 0L, Errors.NONE)
+ );
+
+ // After fetching successfully from the leader, follower rejects PreVotes even if the remote log is up to date
+ context.deliverRequest(context.preVoteRequest(epoch, otherNodeKey, epoch, 1));
+ context.pollUntilResponse();
+ context.assertSentVoteResponse(Errors.NONE, epoch, OptionalInt.of(leader.id()), false);
+ assertTrue(context.client.quorum().isFollower());
+
+ // Leader resigns by sending EndQuorumEpoch
+ context.deliverRequest(context.endEpochRequest(epoch, leader.id(), List.of(otherNodeKey, local)));
+ context.pollUntilResponse();
+ context.assertSentEndQuorumEpochResponse(Errors.NONE, epoch, OptionalInt.of(leader.id()));
+ assertTrue(context.client.quorum().isFollower());
+
+ // Once the leader has resigned, follower grants PreVotes again
+ context.deliverRequest(context.preVoteRequest(epoch, otherNodeKey, epoch, 1));
+ context.pollUntilResponse();
+ context.assertSentVoteResponse(Errors.NONE, epoch, OptionalInt.of(leader.id()), true);
+ assertTrue(context.client.quorum().isFollower());
Review Comment:
What is the motivation behind checking that the local node is a follower repeatedly?
##########
raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientPreVoteTest.java:
##########
@@ -670,6 +670,54 @@ public void testFollowerGrantsPreVoteIfHasNotFetchedYet(KRaftVersion kraftVersio
assertTrue(context.client.quorum().isFollower());
}
+ @ParameterizedTest
+ @EnumSource(value = KRaftVersion.class)
+ public void testFollowerGrantsPreVoteAfterLeaderResigns(KRaftVersion kraftVersion) throws Exception {
+ int localId = randomReplicaId();
+ ReplicaKey local = replicaKey(localId, true);
+ ReplicaKey leader = replicaKey(localId + 1, true);
+ ReplicaKey otherNodeKey = replicaKey(localId + 2, true);
+ int epoch = 2;
+
+ RaftClientTestContext context = new RaftClientTestContext.Builder(
+ local.id(),
+ local.directoryId().get()
+ )
+ .withStartingVoters(VoterSetTest.voterSet(Stream.of(local, leader, otherNodeKey)), kraftVersion)
+ .withElectedLeader(epoch, leader.id())
+ .withRaftProtocol(KIP_996_PROTOCOL)
+ .build();
+
+ context.assertElectedLeader(epoch, leader.id());
+
+ context.pollUntilRequest();
+ RaftRequest.Outbound fetchRequest = context.assertSentFetchRequest();
+ context.assertFetchRequestData(fetchRequest, epoch, 0L, 0, context.client.highWatermark());
+ context.deliverResponse(
+ fetchRequest.correlationId(),
+ fetchRequest.destination(),
+ context.fetchResponse(epoch, leader.id(), MemoryRecords.EMPTY, 0L, Errors.NONE)
+ );
+
+ // After fetching successfully from the leader, follower rejects PreVotes even if the remote log is up to date
+ context.deliverRequest(context.preVoteRequest(epoch, otherNodeKey, epoch, 1));
+ context.pollUntilResponse();
+ context.assertSentVoteResponse(Errors.NONE, epoch, OptionalInt.of(leader.id()), false);
+ assertTrue(context.client.quorum().isFollower());
+
+ // Leader resigns by sending EndQuorumEpoch
+ context.deliverRequest(context.endEpochRequest(epoch, leader.id(), List.of(otherNodeKey, local)));
+ context.pollUntilResponse();
+ context.assertSentEndQuorumEpochResponse(Errors.NONE, epoch, OptionalInt.of(leader.id()));
+ assertTrue(context.client.quorum().isFollower());
+
+ // Once the leader has resigned, follower grants PreVotes again
Review Comment:
```suggestion
// Once the leader has resigned, the follower can now grant the same PreVote that was previously rejected
```
##########
raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientPreVoteTest.java:
##########
@@ -670,6 +670,54 @@ public void testFollowerGrantsPreVoteIfHasNotFetchedYet(KRaftVersion kraftVersio
assertTrue(context.client.quorum().isFollower());
}
+ @ParameterizedTest
+ @EnumSource(value = KRaftVersion.class)
+ public void testFollowerGrantsPreVoteAfterLeaderResigns(KRaftVersion kraftVersion) throws Exception {
Review Comment:
We do not need to parametrize over the `kraftVersion`. That feature and pre-vote are not related.
##########
raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientPreVoteTest.java:
##########
@@ -670,6 +670,54 @@ public void testFollowerGrantsPreVoteIfHasNotFetchedYet(KRaftVersion kraftVersio
assertTrue(context.client.quorum().isFollower());
}
+ @ParameterizedTest
+ @EnumSource(value = KRaftVersion.class)
+ public void testFollowerGrantsPreVoteAfterLeaderResigns(KRaftVersion kraftVersion) throws Exception {
+ int localId = randomReplicaId();
+ ReplicaKey local = replicaKey(localId, true);
+ ReplicaKey leader = replicaKey(localId + 1, true);
+ ReplicaKey otherNodeKey = replicaKey(localId + 2, true);
+ int epoch = 2;
+
+ RaftClientTestContext context = new RaftClientTestContext.Builder(
+ local.id(),
+ local.directoryId().get()
+ )
+ .withStartingVoters(VoterSetTest.voterSet(Stream.of(local, leader, otherNodeKey)), kraftVersion)
+ .withElectedLeader(epoch, leader.id())
+ .withRaftProtocol(KIP_996_PROTOCOL)
+ .build();
+
+ context.assertElectedLeader(epoch, leader.id());
+
+ context.pollUntilRequest();
+ RaftRequest.Outbound fetchRequest = context.assertSentFetchRequest();
+ context.assertFetchRequestData(fetchRequest, epoch, 0L, 0, context.client.highWatermark());
+ context.deliverResponse(
+ fetchRequest.correlationId(),
+ fetchRequest.destination(),
+ context.fetchResponse(epoch, leader.id(), MemoryRecords.EMPTY, 0L, Errors.NONE)
+ );
+
+ // After fetching successfully from the leader, follower rejects PreVotes even if the remote log is up to date
+ context.deliverRequest(context.preVoteRequest(epoch, otherNodeKey, epoch, 1));
Review Comment:
It might be good to store `context.preVoteRequest` in a variable. That way, it is clear to the test reader that the same pre-vote request first gets rejected, and then after the leader resigns, the same request now gets granted.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20895: Grant pre-vote once the leader has resigned [kafka]
Posted by "mimaison (via GitHub)" <gi...@apache.org>.
mimaison commented on code in PR #23168:
URL: https://github.com/apache/kafka/pull/23168#discussion_r3805800928
##########
raft/src/main/java/org/apache/kafka/raft/FollowerState.java:
##########
@@ -232,17 +244,18 @@ public void setFetchingSnapshot(Optional<RawSnapshotWriter> newSnapshot) {
@Override
public boolean canGrantVote(ReplicaKey replicaKey, boolean isLogUpToDate, boolean isPreVote) {
- if (isPreVote && !hasFetchedFromLeader && isLogUpToDate) {
+ if (isPreVote && (!hasFetchedFromLeader || hasLeaderResigned) && isLogUpToDate) {
Review Comment:
Should we have a test for that change in `FollowerStateTest`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20895: Grant pre-vote once the leader has resigned [kafka]
Posted by "sumanpal97 (via GitHub)" <gi...@apache.org>.
sumanpal97 commented on code in PR #23168:
URL: https://github.com/apache/kafka/pull/23168#discussion_r4347379990
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -195,6 +197,83 @@ void canElectNewLeaderAfterOldLeaderFailure() {
});
}
+ @ParameterizedTest
+ @ValueSource(ints = {4, 5, 6, 7, 8, 9})
+ void preferredSuccessorWinsElectionAfterGracefulLeaderShutdown(int numVoters) {
+ Cluster cluster = new Cluster(numVoters, 0, 1234L);
+ /* The other simulations lower this to keep their runtime down, but here the whole point is
+ * that the nominated successors stand for election one at a time, so use the production
+ * default to keep the gap between them wide enough to be meaningful.
+ */
+ cluster.electionBackoffMaxMs = QuorumConfig.DEFAULT_QUORUM_ELECTION_BACKOFF_MAX_MS;
+ MessageRouter router = new MessageRouter(cluster);
+ EventScheduler scheduler = schedulerWithDefaultInvariants(cluster);
+
+ cluster.startAll();
+ schedulePolling(scheduler, cluster, 3, 5);
+ scheduler.schedule(router::deliverAll, 0, 2, 1);
+ scheduler.runUntil(cluster::hasConsistentLeader);
+ // Every replica has fetched from the leader and is caught up with it
+ scheduler.runUntil(() -> cluster.allReachedHighWatermark(1));
+
+ int leaderId = cluster.latestLeader().orElseThrow(() ->
+ new AssertionError("Failed to find current leader")
+ );
+
+ // Read the ranking that the leader will put into its EndQuorumEpoch requests
+ LeaderState<Integer> leaderState = cluster
+ .nodeIfRunning(leaderId)
+ .orElseThrow(() -> new AssertionError("Leader " + leaderId + " is not running"))
+ .client
+ .quorum()
+ .leaderStateOrThrow();
+ List<Integer> preferredSuccessors = leaderState
+ .nonLeaderVotersByDescendingFetchOffset()
+ .stream()
+ .map(ReplicaKey::id)
+ .collect(Collectors.toList());
+
+ // A clean shutdown, so the leader resigns and nominates the successors above
+ long resignedAtMs = cluster.time.milliseconds();
+ cluster.shutdown(leaderId);
+
+ scheduler.runUntil(() ->
+ cluster.latestLeader().isPresent() && cluster.latestLeader().getAsInt() != leaderId
+ );
+
+ int newLeaderId = cluster.latestLeader().getAsInt();
+ long electionDurationMs = cluster.time.milliseconds() - resignedAtMs;
+ int position = preferredSuccessors.indexOf(newLeaderId);
+ int backoffBaseMs = cluster.electionBackoffMaxMs >> (preferredSuccessors.size() - 1);
+ long backoffOfWinnerMs = position <= 0 ? 0 : (long) backoffBaseMs << (position - 1);
Review Comment:
Good point. That calculation is gone now that the simulation test has been removed.
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -195,6 +197,83 @@ void canElectNewLeaderAfterOldLeaderFailure() {
});
}
+ @ParameterizedTest
+ @ValueSource(ints = {4, 5, 6, 7, 8, 9})
+ void preferredSuccessorWinsElectionAfterGracefulLeaderShutdown(int numVoters) {
+ Cluster cluster = new Cluster(numVoters, 0, 1234L);
+ /* The other simulations lower this to keep their runtime down, but here the whole point is
+ * that the nominated successors stand for election one at a time, so use the production
+ * default to keep the gap between them wide enough to be meaningful.
+ */
+ cluster.electionBackoffMaxMs = QuorumConfig.DEFAULT_QUORUM_ELECTION_BACKOFF_MAX_MS;
+ MessageRouter router = new MessageRouter(cluster);
+ EventScheduler scheduler = schedulerWithDefaultInvariants(cluster);
+
+ cluster.startAll();
+ schedulePolling(scheduler, cluster, 3, 5);
+ scheduler.schedule(router::deliverAll, 0, 2, 1);
+ scheduler.runUntil(cluster::hasConsistentLeader);
+ // Every replica has fetched from the leader and is caught up with it
+ scheduler.runUntil(() -> cluster.allReachedHighWatermark(1));
+
+ int leaderId = cluster.latestLeader().orElseThrow(() ->
+ new AssertionError("Failed to find current leader")
+ );
+
+ // Read the ranking that the leader will put into its EndQuorumEpoch requests
+ LeaderState<Integer> leaderState = cluster
+ .nodeIfRunning(leaderId)
+ .orElseThrow(() -> new AssertionError("Leader " + leaderId + " is not running"))
+ .client
+ .quorum()
+ .leaderStateOrThrow();
+ List<Integer> preferredSuccessors = leaderState
+ .nonLeaderVotersByDescendingFetchOffset()
+ .stream()
+ .map(ReplicaKey::id)
+ .collect(Collectors.toList());
+
+ // A clean shutdown, so the leader resigns and nominates the successors above
+ long resignedAtMs = cluster.time.milliseconds();
+ cluster.shutdown(leaderId);
+
+ scheduler.runUntil(() ->
+ cluster.latestLeader().isPresent() && cluster.latestLeader().getAsInt() != leaderId
+ );
+
+ int newLeaderId = cluster.latestLeader().getAsInt();
+ long electionDurationMs = cluster.time.milliseconds() - resignedAtMs;
+ int position = preferredSuccessors.indexOf(newLeaderId);
+ int backoffBaseMs = cluster.electionBackoffMaxMs >> (preferredSuccessors.size() - 1);
+ long backoffOfWinnerMs = position <= 0 ? 0 : (long) backoffBaseMs << (position - 1);
+
+ System.out.printf(
+ "%s voters: leader %s resigned nominating %s. Node %s won at position %s, whose " +
+ "election backoff is %s ms, and the election took %s ms%n",
+ numVoters,
+ leaderId,
+ preferredSuccessors,
+ newLeaderId,
+ position,
+ backoffOfWinnerMs,
+ electionDurationMs
+ );
Review Comment:
Removed along with the simulation test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20895: Grant pre-vote once the leader has resigned [kafka]
Posted by "sumanpal97 (via GitHub)" <gi...@apache.org>.
sumanpal97 commented on PR #23168:
URL: https://github.com/apache/kafka/pull/23168#issuecomment-5488728220
Hi @jsancio @kevin-wu24 , can you please take a look on the changes
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20895: Grant pre-vote once the leader has resigned [kafka]
Posted by "sumanpal97 (via GitHub)" <gi...@apache.org>.
sumanpal97 commented on code in PR #23168:
URL: https://github.com/apache/kafka/pull/23168#discussion_r4379098842
##########
raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientPreVoteTest.java:
##########
@@ -17,13 +17,14 @@
package org.apache.kafka.raft;
import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.message.VoteRequestData;
import org.apache.kafka.common.message.VoteResponseData;
import org.apache.kafka.common.protocol.ApiKeys;
import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.record.internal.MemoryRecords;
import org.apache.kafka.raft.RaftClientTestContext.RaftProtocol;
import org.apache.kafka.server.common.KRaftVersion;
-
Review Comment:
Added the blank line back.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20895: Grant pre-vote once the leader has resigned [kafka]
Posted by "sumanpal97 (via GitHub)" <gi...@apache.org>.
sumanpal97 commented on code in PR #23168:
URL: https://github.com/apache/kafka/pull/23168#discussion_r4347377940
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -195,6 +197,83 @@ void canElectNewLeaderAfterOldLeaderFailure() {
});
}
+ @ParameterizedTest
+ @ValueSource(ints = {4, 5, 6, 7, 8, 9})
Review Comment:
No longer applicable. The simulation test was removed in favour of the unit test.
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -195,6 +197,83 @@ void canElectNewLeaderAfterOldLeaderFailure() {
});
}
+ @ParameterizedTest
+ @ValueSource(ints = {4, 5, 6, 7, 8, 9})
+ void preferredSuccessorWinsElectionAfterGracefulLeaderShutdown(int numVoters) {
+ Cluster cluster = new Cluster(numVoters, 0, 1234L);
+ /* The other simulations lower this to keep their runtime down, but here the whole point is
+ * that the nominated successors stand for election one at a time, so use the production
+ * default to keep the gap between them wide enough to be meaningful.
+ */
Review Comment:
Thanks. Per Kevin's feedback, this simulation test has been removed, so the comment is gone too.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20895: Grant pre-vote once the leader has resigned [kafka]
Posted by "kevin-wu24 (via GitHub)" <gi...@apache.org>.
kevin-wu24 commented on code in PR #23168:
URL: https://github.com/apache/kafka/pull/23168#discussion_r4373283323
##########
raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientPreVoteTest.java:
##########
@@ -17,13 +17,14 @@
package org.apache.kafka.raft;
import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.message.VoteRequestData;
import org.apache.kafka.common.message.VoteResponseData;
import org.apache.kafka.common.protocol.ApiKeys;
import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.record.internal.MemoryRecords;
import org.apache.kafka.raft.RaftClientTestContext.RaftProtocol;
import org.apache.kafka.server.common.KRaftVersion;
-
Review Comment:
Please add this empty line back to pass the checkstyle.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20895: Grant pre-vote once the leader has resigned [kafka]
Posted by "mimaison (via GitHub)" <gi...@apache.org>.
mimaison commented on PR #23168:
URL: https://github.com/apache/kafka/pull/23168#issuecomment-5343891370
@jsancio @kevin-wu24 Can you take a look?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20895: Grant pre-vote once the leader has resigned [kafka]
Posted by "sumanpal97 (via GitHub)" <gi...@apache.org>.
sumanpal97 commented on code in PR #23168:
URL: https://github.com/apache/kafka/pull/23168#discussion_r4380290752
##########
raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientPreVoteTest.java:
##########
@@ -670,6 +672,54 @@ public void testFollowerGrantsPreVoteIfHasNotFetchedYet(KRaftVersion kraftVersio
assertTrue(context.client.quorum().isFollower());
}
+ @Test
+ public void testFollowerGrantsPreVoteAfterLeaderResigns() throws Exception {
+ int localId = randomReplicaId();
+ ReplicaKey local = replicaKey(localId, true);
+ ReplicaKey leader = replicaKey(localId + 1, true);
+ ReplicaKey otherNodeKey = replicaKey(localId + 2, true);
+ int epoch = 2;
+
+ RaftClientTestContext context = new RaftClientTestContext.Builder(
+ local.id(),
+ local.directoryId().get()
+ )
+ .withStartingVoters(VoterSetTest.voterSet(Stream.of(local, leader, otherNodeKey)), KRaftVersion.KRAFT_VERSION_0)
+ .withElectedLeader(epoch, leader.id())
+ .withRaftProtocol(RaftProtocol.KIP_1186_PROTOCOL)
+ .build();
+
+ context.assertElectedLeader(epoch, leader.id());
+
+ context.pollUntilRequest();
+ RaftRequest.Outbound fetchRequest = context.assertSentFetchRequest();
+ context.assertFetchRequestData(fetchRequest, epoch, 0L, 0, context.client.highWatermark());
+ context.deliverResponse(
+ fetchRequest.correlationId(),
+ fetchRequest.destination(),
+ context.fetchResponse(epoch, leader.id(), MemoryRecords.EMPTY, 0L, Errors.NONE)
+ );
+
+ // After fetching successfully from the leader, follower rejects PreVotes even if the remote log is up to date
+ VoteRequestData preVote = context.preVoteRequest(epoch, otherNodeKey, epoch, 1);
+ context.deliverRequest(preVote);
+ context.pollUntilResponse();
+ context.assertSentVoteResponse(Errors.NONE, epoch, OptionalInt.of(leader.id()), false);
+ assertTrue(context.client.quorum().isFollower());
+
+ // Leader resigns by sending EndQuorumEpoch. This request can move the node out of the follower state.
Review Comment:
Updated.
##########
raft/src/test/java/org/apache/kafka/raft/FollowerStateTest.java:
##########
@@ -122,6 +122,24 @@ public void testPreVoteAfterSuccessfulFetchFromLeader(boolean isLogUpToDate) {
assertFalse(state.canGrantVote(ReplicaKey.of(10, ReplicaKey.NO_DIRECTORY_ID), isLogUpToDate, false));
}
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testPreVoteAfterLeaderResigned(boolean isLogUpToDate) {
+ FollowerState state = newFollowerState(Set.of(1, 2, 3));
+ state.resetFetchTimeoutForSuccessfulFetch(time.milliseconds());
+ state.setHasLeaderResigned();
+
+ assertEquals(isLogUpToDate, state.canGrantVote(ReplicaKey.of(1, ReplicaKey.NO_DIRECTORY_ID), isLogUpToDate, true));
Review Comment:
Wrapped the new lines so they stay under 100 characters.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20895: Grant pre-vote once the leader has resigned [kafka]
Posted by "ahuang98 (via GitHub)" <gi...@apache.org>.
ahuang98 commented on code in PR #23168:
URL: https://github.com/apache/kafka/pull/23168#discussion_r4274605836
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -195,6 +197,83 @@ void canElectNewLeaderAfterOldLeaderFailure() {
});
}
+ @ParameterizedTest
+ @ValueSource(ints = {4, 5, 6, 7, 8, 9})
+ void preferredSuccessorWinsElectionAfterGracefulLeaderShutdown(int numVoters) {
+ Cluster cluster = new Cluster(numVoters, 0, 1234L);
+ /* The other simulations lower this to keep their runtime down, but here the whole point is
+ * that the nominated successors stand for election one at a time, so use the production
+ * default to keep the gap between them wide enough to be meaningful.
+ */
+ cluster.electionBackoffMaxMs = QuorumConfig.DEFAULT_QUORUM_ELECTION_BACKOFF_MAX_MS;
+ MessageRouter router = new MessageRouter(cluster);
+ EventScheduler scheduler = schedulerWithDefaultInvariants(cluster);
+
+ cluster.startAll();
+ schedulePolling(scheduler, cluster, 3, 5);
+ scheduler.schedule(router::deliverAll, 0, 2, 1);
+ scheduler.runUntil(cluster::hasConsistentLeader);
+ // Every replica has fetched from the leader and is caught up with it
+ scheduler.runUntil(() -> cluster.allReachedHighWatermark(1));
+
+ int leaderId = cluster.latestLeader().orElseThrow(() ->
+ new AssertionError("Failed to find current leader")
+ );
+
+ // Read the ranking that the leader will put into its EndQuorumEpoch requests
+ LeaderState<Integer> leaderState = cluster
+ .nodeIfRunning(leaderId)
+ .orElseThrow(() -> new AssertionError("Leader " + leaderId + " is not running"))
+ .client
+ .quorum()
+ .leaderStateOrThrow();
+ List<Integer> preferredSuccessors = leaderState
+ .nonLeaderVotersByDescendingFetchOffset()
+ .stream()
+ .map(ReplicaKey::id)
+ .collect(Collectors.toList());
+
+ // A clean shutdown, so the leader resigns and nominates the successors above
+ long resignedAtMs = cluster.time.milliseconds();
+ cluster.shutdown(leaderId);
+
+ scheduler.runUntil(() ->
+ cluster.latestLeader().isPresent() && cluster.latestLeader().getAsInt() != leaderId
+ );
+
+ int newLeaderId = cluster.latestLeader().getAsInt();
+ long electionDurationMs = cluster.time.milliseconds() - resignedAtMs;
+ int position = preferredSuccessors.indexOf(newLeaderId);
+ int backoffBaseMs = cluster.electionBackoffMaxMs >> (preferredSuccessors.size() - 1);
+ long backoffOfWinnerMs = position <= 0 ? 0 : (long) backoffBaseMs << (position - 1);
+
+ System.out.printf(
+ "%s voters: leader %s resigned nominating %s. Node %s won at position %s, whose " +
+ "election backoff is %s ms, and the election took %s ms%n",
+ numVoters,
+ leaderId,
+ preferredSuccessors,
+ newLeaderId,
+ position,
+ backoffOfWinnerMs,
+ electionDurationMs
+ );
Review Comment:
do we need this? seems to be covered by the assert below
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -590,6 +669,7 @@ private static class Cluster {
final Map<Integer, Node> voters = new HashMap<>();
final Map<Integer, PersistentState> nodes = new HashMap<>();
final Map<Integer, RaftNode> running = new HashMap<>();
+ int electionBackoffMaxMs = ELECTION_JITTER_MS;
Review Comment:
do we need this change?
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -195,6 +197,83 @@ void canElectNewLeaderAfterOldLeaderFailure() {
});
}
+ @ParameterizedTest
+ @ValueSource(ints = {4, 5, 6, 7, 8, 9})
Review Comment:
why not use `withVotersObservers`?
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -195,6 +197,83 @@ void canElectNewLeaderAfterOldLeaderFailure() {
});
}
+ @ParameterizedTest
+ @ValueSource(ints = {4, 5, 6, 7, 8, 9})
+ void preferredSuccessorWinsElectionAfterGracefulLeaderShutdown(int numVoters) {
+ Cluster cluster = new Cluster(numVoters, 0, 1234L);
+ /* The other simulations lower this to keep their runtime down, but here the whole point is
+ * that the nominated successors stand for election one at a time, so use the production
+ * default to keep the gap between them wide enough to be meaningful.
+ */
Review Comment:
nit: can we reword this slightly? I would prefer not using 'stand for election one at a time' as it gives the impression of strict 'one after the other'. maybe something like:
```
Keep the larger, default quorum.election.backoff.max.ms here to give successors enough time to be
considered one after another. This config sets how long each successor waits before it starts an election.
With a small value it is more likely two successors will have overlapping elections.
```
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -195,6 +197,83 @@ void canElectNewLeaderAfterOldLeaderFailure() {
});
}
+ @ParameterizedTest
+ @ValueSource(ints = {4, 5, 6, 7, 8, 9})
+ void preferredSuccessorWinsElectionAfterGracefulLeaderShutdown(int numVoters) {
+ Cluster cluster = new Cluster(numVoters, 0, 1234L);
+ /* The other simulations lower this to keep their runtime down, but here the whole point is
+ * that the nominated successors stand for election one at a time, so use the production
+ * default to keep the gap between them wide enough to be meaningful.
+ */
+ cluster.electionBackoffMaxMs = QuorumConfig.DEFAULT_QUORUM_ELECTION_BACKOFF_MAX_MS;
+ MessageRouter router = new MessageRouter(cluster);
+ EventScheduler scheduler = schedulerWithDefaultInvariants(cluster);
+
+ cluster.startAll();
+ schedulePolling(scheduler, cluster, 3, 5);
+ scheduler.schedule(router::deliverAll, 0, 2, 1);
+ scheduler.runUntil(cluster::hasConsistentLeader);
+ // Every replica has fetched from the leader and is caught up with it
+ scheduler.runUntil(() -> cluster.allReachedHighWatermark(1));
+
+ int leaderId = cluster.latestLeader().orElseThrow(() ->
+ new AssertionError("Failed to find current leader")
+ );
+
+ // Read the ranking that the leader will put into its EndQuorumEpoch requests
+ LeaderState<Integer> leaderState = cluster
+ .nodeIfRunning(leaderId)
+ .orElseThrow(() -> new AssertionError("Leader " + leaderId + " is not running"))
+ .client
+ .quorum()
+ .leaderStateOrThrow();
+ List<Integer> preferredSuccessors = leaderState
+ .nonLeaderVotersByDescendingFetchOffset()
+ .stream()
+ .map(ReplicaKey::id)
+ .collect(Collectors.toList());
+
+ // A clean shutdown, so the leader resigns and nominates the successors above
+ long resignedAtMs = cluster.time.milliseconds();
+ cluster.shutdown(leaderId);
+
+ scheduler.runUntil(() ->
+ cluster.latestLeader().isPresent() && cluster.latestLeader().getAsInt() != leaderId
+ );
+
+ int newLeaderId = cluster.latestLeader().getAsInt();
+ long electionDurationMs = cluster.time.milliseconds() - resignedAtMs;
+ int position = preferredSuccessors.indexOf(newLeaderId);
+ int backoffBaseMs = cluster.electionBackoffMaxMs >> (preferredSuccessors.size() - 1);
+ long backoffOfWinnerMs = position <= 0 ? 0 : (long) backoffBaseMs << (position - 1);
Review Comment:
also if the calculation were to change in the production code then this becomes out of date
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20895: Grant pre-vote once the leader has resigned [kafka]
Posted by "kevin-wu24 (via GitHub)" <gi...@apache.org>.
kevin-wu24 commented on code in PR #23168:
URL: https://github.com/apache/kafka/pull/23168#discussion_r4273989281
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -195,6 +197,83 @@ void canElectNewLeaderAfterOldLeaderFailure() {
});
}
+ @ParameterizedTest
+ @ValueSource(ints = {4, 5, 6, 7, 8, 9})
Review Comment:
I do not think we should have a simulation test for this. This issue is more of a performance issue, whereas the simulation tests are conceptually more about ensuring the invariants of the KRaft protocol are upheld. Additionally, simulation tests are more expensive to run than the unit tests. For example, this test could fail if you decided to drop some messages, which happens in the real world a lot.
See my other comment for how you can make the unit test model this situation without using a simulation test.
##########
raft/src/main/java/org/apache/kafka/raft/FollowerState.java:
##########
@@ -154,6 +158,14 @@ public void overrideFetchTimeout(long currentTimeMs, long timeoutMs) {
fetchTimer.reset(timeoutMs);
}
+ /**
+ * Record that the leader of this epoch has resigned, after receiving an `EndQuorumEpoch`
+ * request from it.
+ */
+ public void setLeaderHasResigned() {
Review Comment:
nit: `setHasLeaderResigned`
##########
raft/src/test/java/org/apache/kafka/raft/FollowerStateTest.java:
##########
@@ -122,6 +122,24 @@ public void testPreVoteAfterSuccessfulFetchFromLeader(boolean isLogUpToDate) {
assertFalse(state.canGrantVote(ReplicaKey.of(10, ReplicaKey.NO_DIRECTORY_ID), isLogUpToDate, false));
}
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testPreVoteAfterLeaderResigned(boolean isLogUpToDate) {
+ FollowerState state = newFollowerState(Set.of(1, 2, 3));
+ state.resetFetchTimeoutForSuccessfulFetch(time.milliseconds());
+ state.setLeaderHasResigned();
+
+ assertEquals(isLogUpToDate, state.canGrantVote(ReplicaKey.of(1, ReplicaKey.NO_DIRECTORY_ID), isLogUpToDate, true));
+ assertEquals(isLogUpToDate, state.canGrantVote(ReplicaKey.of(2, ReplicaKey.NO_DIRECTORY_ID), isLogUpToDate, true));
+ assertEquals(isLogUpToDate, state.canGrantVote(ReplicaKey.of(3, ReplicaKey.NO_DIRECTORY_ID), isLogUpToDate, true));
+ assertEquals(isLogUpToDate, state.canGrantVote(ReplicaKey.of(10, ReplicaKey.NO_DIRECTORY_ID), isLogUpToDate, true));
+
+ assertFalse(state.canGrantVote(ReplicaKey.of(1, ReplicaKey.NO_DIRECTORY_ID), isLogUpToDate, false));
+ assertFalse(state.canGrantVote(ReplicaKey.of(2, ReplicaKey.NO_DIRECTORY_ID), isLogUpToDate, false));
+ assertFalse(state.canGrantVote(ReplicaKey.of(3, ReplicaKey.NO_DIRECTORY_ID), isLogUpToDate, false));
+ assertFalse(state.canGrantVote(ReplicaKey.of(10, ReplicaKey.NO_DIRECTORY_ID), isLogUpToDate, false));
+ }
Review Comment:
In my opinion, a better test would have the local follower node start as rejecting preVotes from a remote node whose log is up to date because `hasFetchedFromLeader` is true. Then, the follower is delivered an `EndQuorumEpochRequest` to model the leader resigning. Finally, this test should show that a preVote is now granted by the local follower.
Writing a unit test like this also makes it so we do not need the simulation test below IMO. It would go in `KafkaRaftClientPreVoteTest`.
What do you think?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20895: Grant pre-vote once the leader has resigned [kafka]
Posted by "sumanpal97 (via GitHub)" <gi...@apache.org>.
sumanpal97 commented on code in PR #23168:
URL: https://github.com/apache/kafka/pull/23168#discussion_r4367579415
##########
raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientPreVoteTest.java:
##########
@@ -670,6 +670,54 @@ public void testFollowerGrantsPreVoteIfHasNotFetchedYet(KRaftVersion kraftVersio
assertTrue(context.client.quorum().isFollower());
}
+ @ParameterizedTest
+ @EnumSource(value = KRaftVersion.class)
+ public void testFollowerGrantsPreVoteAfterLeaderResigns(KRaftVersion kraftVersion) throws Exception {
+ int localId = randomReplicaId();
+ ReplicaKey local = replicaKey(localId, true);
+ ReplicaKey leader = replicaKey(localId + 1, true);
+ ReplicaKey otherNodeKey = replicaKey(localId + 2, true);
+ int epoch = 2;
+
+ RaftClientTestContext context = new RaftClientTestContext.Builder(
+ local.id(),
+ local.directoryId().get()
+ )
+ .withStartingVoters(VoterSetTest.voterSet(Stream.of(local, leader, otherNodeKey)), kraftVersion)
+ .withElectedLeader(epoch, leader.id())
+ .withRaftProtocol(KIP_996_PROTOCOL)
+ .build();
+
+ context.assertElectedLeader(epoch, leader.id());
+
+ context.pollUntilRequest();
+ RaftRequest.Outbound fetchRequest = context.assertSentFetchRequest();
+ context.assertFetchRequestData(fetchRequest, epoch, 0L, 0, context.client.highWatermark());
+ context.deliverResponse(
+ fetchRequest.correlationId(),
+ fetchRequest.destination(),
+ context.fetchResponse(epoch, leader.id(), MemoryRecords.EMPTY, 0L, Errors.NONE)
+ );
+
+ // After fetching successfully from the leader, follower rejects PreVotes even if the remote log is up to date
+ context.deliverRequest(context.preVoteRequest(epoch, otherNodeKey, epoch, 1));
+ context.pollUntilResponse();
+ context.assertSentVoteResponse(Errors.NONE, epoch, OptionalInt.of(leader.id()), false);
+ assertTrue(context.client.quorum().isFollower());
+
+ // Leader resigns by sending EndQuorumEpoch
+ context.deliverRequest(context.endEpochRequest(epoch, leader.id(), List.of(otherNodeKey, local)));
+ context.pollUntilResponse();
+ context.assertSentEndQuorumEpochResponse(Errors.NONE, epoch, OptionalInt.of(leader.id()));
+ assertTrue(context.client.quorum().isFollower());
+
+ // Once the leader has resigned, follower grants PreVotes again
+ context.deliverRequest(context.preVoteRequest(epoch, otherNodeKey, epoch, 1));
+ context.pollUntilResponse();
+ context.assertSentVoteResponse(Errors.NONE, epoch, OptionalInt.of(leader.id()), true);
+ assertTrue(context.client.quorum().isFollower());
Review Comment:
A PreVote response does not change the node's role, so those checks were unnecessary. I kept only the check after EndQuorumEpoch, because that request can move the node out of the follower state.
##########
raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientPreVoteTest.java:
##########
@@ -670,6 +670,54 @@ public void testFollowerGrantsPreVoteIfHasNotFetchedYet(KRaftVersion kraftVersio
assertTrue(context.client.quorum().isFollower());
}
+ @ParameterizedTest
+ @EnumSource(value = KRaftVersion.class)
+ public void testFollowerGrantsPreVoteAfterLeaderResigns(KRaftVersion kraftVersion) throws Exception {
+ int localId = randomReplicaId();
+ ReplicaKey local = replicaKey(localId, true);
+ ReplicaKey leader = replicaKey(localId + 1, true);
+ ReplicaKey otherNodeKey = replicaKey(localId + 2, true);
+ int epoch = 2;
+
+ RaftClientTestContext context = new RaftClientTestContext.Builder(
+ local.id(),
+ local.directoryId().get()
+ )
+ .withStartingVoters(VoterSetTest.voterSet(Stream.of(local, leader, otherNodeKey)), kraftVersion)
+ .withElectedLeader(epoch, leader.id())
+ .withRaftProtocol(KIP_996_PROTOCOL)
+ .build();
+
+ context.assertElectedLeader(epoch, leader.id());
+
+ context.pollUntilRequest();
+ RaftRequest.Outbound fetchRequest = context.assertSentFetchRequest();
+ context.assertFetchRequestData(fetchRequest, epoch, 0L, 0, context.client.highWatermark());
+ context.deliverResponse(
+ fetchRequest.correlationId(),
+ fetchRequest.destination(),
+ context.fetchResponse(epoch, leader.id(), MemoryRecords.EMPTY, 0L, Errors.NONE)
+ );
+
+ // After fetching successfully from the leader, follower rejects PreVotes even if the remote log is up to date
+ context.deliverRequest(context.preVoteRequest(epoch, otherNodeKey, epoch, 1));
Review Comment:
Agree. Stored the PreVote request and reused it, so the rejected and granted requests are the same object.
##########
raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientPreVoteTest.java:
##########
@@ -670,6 +670,54 @@ public void testFollowerGrantsPreVoteIfHasNotFetchedYet(KRaftVersion kraftVersio
assertTrue(context.client.quorum().isFollower());
}
+ @ParameterizedTest
+ @EnumSource(value = KRaftVersion.class)
+ public void testFollowerGrantsPreVoteAfterLeaderResigns(KRaftVersion kraftVersion) throws Exception {
+ int localId = randomReplicaId();
+ ReplicaKey local = replicaKey(localId, true);
+ ReplicaKey leader = replicaKey(localId + 1, true);
+ ReplicaKey otherNodeKey = replicaKey(localId + 2, true);
+ int epoch = 2;
+
+ RaftClientTestContext context = new RaftClientTestContext.Builder(
+ local.id(),
+ local.directoryId().get()
+ )
+ .withStartingVoters(VoterSetTest.voterSet(Stream.of(local, leader, otherNodeKey)), kraftVersion)
+ .withElectedLeader(epoch, leader.id())
+ .withRaftProtocol(KIP_996_PROTOCOL)
Review Comment:
Switched the test to KIP_1186_PROTOCOL, the latest protocol, which still includes pre-vote.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20895: Grant pre-vote once the leader has resigned [kafka]
Posted by "sumanpal97 (via GitHub)" <gi...@apache.org>.
sumanpal97 commented on code in PR #23168:
URL: https://github.com/apache/kafka/pull/23168#discussion_r4347376107
##########
raft/src/test/java/org/apache/kafka/raft/FollowerStateTest.java:
##########
@@ -122,6 +122,24 @@ public void testPreVoteAfterSuccessfulFetchFromLeader(boolean isLogUpToDate) {
assertFalse(state.canGrantVote(ReplicaKey.of(10, ReplicaKey.NO_DIRECTORY_ID), isLogUpToDate, false));
}
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testPreVoteAfterLeaderResigned(boolean isLogUpToDate) {
+ FollowerState state = newFollowerState(Set.of(1, 2, 3));
+ state.resetFetchTimeoutForSuccessfulFetch(time.milliseconds());
+ state.setLeaderHasResigned();
+
+ assertEquals(isLogUpToDate, state.canGrantVote(ReplicaKey.of(1, ReplicaKey.NO_DIRECTORY_ID), isLogUpToDate, true));
+ assertEquals(isLogUpToDate, state.canGrantVote(ReplicaKey.of(2, ReplicaKey.NO_DIRECTORY_ID), isLogUpToDate, true));
+ assertEquals(isLogUpToDate, state.canGrantVote(ReplicaKey.of(3, ReplicaKey.NO_DIRECTORY_ID), isLogUpToDate, true));
+ assertEquals(isLogUpToDate, state.canGrantVote(ReplicaKey.of(10, ReplicaKey.NO_DIRECTORY_ID), isLogUpToDate, true));
+
+ assertFalse(state.canGrantVote(ReplicaKey.of(1, ReplicaKey.NO_DIRECTORY_ID), isLogUpToDate, false));
+ assertFalse(state.canGrantVote(ReplicaKey.of(2, ReplicaKey.NO_DIRECTORY_ID), isLogUpToDate, false));
+ assertFalse(state.canGrantVote(ReplicaKey.of(3, ReplicaKey.NO_DIRECTORY_ID), isLogUpToDate, false));
+ assertFalse(state.canGrantVote(ReplicaKey.of(10, ReplicaKey.NO_DIRECTORY_ID), isLogUpToDate, false));
+ }
Review Comment:
Agreed. I added testFollowerGrantsPreVoteAfterLeaderResigns to KafkaRaftClientPreVoteTest. After a successful fetch, the follower rejects a PreVote from an up-to-date node. It then receives an EndQuorumEpochRequest from the leader and grants the same PreVote. I kept the FollowerStateTest case as a state-level check, since Mickael asked for it. Happy to drop it if you'd prefer.
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -195,6 +197,83 @@ void canElectNewLeaderAfterOldLeaderFailure() {
});
}
+ @ParameterizedTest
+ @ValueSource(ints = {4, 5, 6, 7, 8, 9})
Review Comment:
Makes sense, thanks. I removed the simulation test and replaced it with a unit test in KafkaRaftClientPreVoteTest.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20895: Grant pre-vote once the leader has resigned [kafka]
Posted by "sumanpal97 (via GitHub)" <gi...@apache.org>.
sumanpal97 commented on code in PR #23168:
URL: https://github.com/apache/kafka/pull/23168#discussion_r4347382007
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -590,6 +669,7 @@ private static class Cluster {
final Map<Integer, Node> voters = new HashMap<>();
final Map<Integer, PersistentState> nodes = new HashMap<>();
final Map<Integer, RaftNode> running = new HashMap<>();
+ int electionBackoffMaxMs = ELECTION_JITTER_MS;
Review Comment:
Reverted. It was only needed by the simulation test, so RaftEventSimulationTest now matches trunk.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20895: Grant pre-vote once the leader has resigned [kafka]
Posted by "kevin-wu24 (via GitHub)" <gi...@apache.org>.
kevin-wu24 commented on code in PR #23168:
URL: https://github.com/apache/kafka/pull/23168#discussion_r4348795252
##########
raft/src/test/java/org/apache/kafka/raft/FollowerStateTest.java:
##########
@@ -122,6 +122,24 @@ public void testPreVoteAfterSuccessfulFetchFromLeader(boolean isLogUpToDate) {
assertFalse(state.canGrantVote(ReplicaKey.of(10, ReplicaKey.NO_DIRECTORY_ID), isLogUpToDate, false));
}
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testPreVoteAfterLeaderResigned(boolean isLogUpToDate) {
+ FollowerState state = newFollowerState(Set.of(1, 2, 3));
+ state.resetFetchTimeoutForSuccessfulFetch(time.milliseconds());
+ state.setLeaderHasResigned();
+
+ assertEquals(isLogUpToDate, state.canGrantVote(ReplicaKey.of(1, ReplicaKey.NO_DIRECTORY_ID), isLogUpToDate, true));
+ assertEquals(isLogUpToDate, state.canGrantVote(ReplicaKey.of(2, ReplicaKey.NO_DIRECTORY_ID), isLogUpToDate, true));
+ assertEquals(isLogUpToDate, state.canGrantVote(ReplicaKey.of(3, ReplicaKey.NO_DIRECTORY_ID), isLogUpToDate, true));
+ assertEquals(isLogUpToDate, state.canGrantVote(ReplicaKey.of(10, ReplicaKey.NO_DIRECTORY_ID), isLogUpToDate, true));
+
+ assertFalse(state.canGrantVote(ReplicaKey.of(1, ReplicaKey.NO_DIRECTORY_ID), isLogUpToDate, false));
+ assertFalse(state.canGrantVote(ReplicaKey.of(2, ReplicaKey.NO_DIRECTORY_ID), isLogUpToDate, false));
+ assertFalse(state.canGrantVote(ReplicaKey.of(3, ReplicaKey.NO_DIRECTORY_ID), isLogUpToDate, false));
+ assertFalse(state.canGrantVote(ReplicaKey.of(10, ReplicaKey.NO_DIRECTORY_ID), isLogUpToDate, false));
+ }
Review Comment:
> I kept the FollowerStateTest case as a state-level check, since Mickael asked for it.
Yeah that is fine to keep.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20895: Grant pre-vote once the leader has resigned [kafka]
Posted by "sumanpal97 (via GitHub)" <gi...@apache.org>.
sumanpal97 commented on PR #23168:
URL: https://github.com/apache/kafka/pull/23168#issuecomment-6386087072
@kevin-wu24 Updated PR Description as well.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org