You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "kevin-wu24 (via GitHub)" <gi...@apache.org> on 2025/02/20 20:22:32 UTC
[PR] KAFKA-16580: Enable dynamic quorum reconfiguration for raft simulation tests pt 1 [kafka]
kevin-wu24 opened a new pull request, #18987:
URL: https://github.com/apache/kafka/pull/18987
### What
- Remove global concept of `voters` from `Cluster`, instead it is now an accurate view of the initial voters.
- Enable KIP-853 by writing a bootstrap snapshot to each of the initial voters similar to tests that use `RaftClientTestContext`.
- Refactor the `MajorityReachedHighWatermark` invariant for KIP-853: since the voter set can change, need to look at the leader's local voter set at the HWM to determine if a majority have reached it.
- Added `RaftTestUtils` to house shared functions between these simulation tests and existing integration tests.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests pt 1 [kafka]
Posted by "ahuang98 (via GitHub)" <gi...@apache.org>.
ahuang98 commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r1982280759
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -550,14 +576,17 @@ void canAddObserverAndRemovePartitionedVoter(
initializeClusterAndStartAppending(cluster, router, scheduler, 10);
- // Partition a random voter, and add an observer to replace it
- RaftNode voterToRemove = cluster.running.get(random.nextInt(numVoters));
- router.filter(voterToRemove.id(), new DropAllTraffic());
+ // Partition a random voter. Add first observer as new voter
+ int voterIdToRemove = cluster.running.get(random.nextInt(numVoters)).id();
+ router.filter(voterIdToRemove, new DropAllTraffic());
+ expectedVoterIds.add(numVoters);
Review Comment:
nit: would be more readable if you used a new var like `firstsObserverId` and assigned to value of numVoters at beginning of 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-16580: Enable dynamic quorum reconfiguration for raft simulation tests pt 1 [kafka]
Posted by "kevin-wu24 (via GitHub)" <gi...@apache.org>.
kevin-wu24 commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r1982298361
##########
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java:
##########
@@ -3637,6 +3637,11 @@ QuorumState quorum() {
return quorum;
}
+ // Visible only for test
+ KRaftControlRecordStateMachine partitionState() {
Review Comment:
Either way we have to expose some additional public methods in `KafkaRaftClient`, and I'm not sure there's a great way around that atm.
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests pt 1 [kafka]
Posted by "ahuang98 (via GitHub)" <gi...@apache.org>.
ahuang98 commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r1970761435
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -1127,14 +1331,75 @@ private MajorityReachedHighWatermark(Cluster cluster) {
@Override
public void verify() {
- cluster.leaderHighWatermark().ifPresent(highWatermark -> {
- long numReachedHighWatermark = cluster.nodes.entrySet().stream()
- .filter(entry -> cluster.voters.containsKey(entry.getKey()))
- .filter(entry -> entry.getValue().log.endOffset().offset() >= highWatermark)
- .count();
- assertTrue(
- numReachedHighWatermark >= cluster.majoritySize(),
- "Insufficient nodes have reached current high watermark");
+ if (cluster.withKip853) {
+ /*
+ * For clusters running in KIP-853 mode, we check that a majority of at least one of:
+ * 1. the leader's voter set at the HWM
Review Comment:
I see, the sentence continues on L1339
wonder if something like the following might be more clear? maybe you can fix my language a bit too
```suggestion
* For clusters running in KIP-853 mode, we consider two possibilities for what the majority of the cluster is:
* 1. the majority based off the current value of lastVoterSet()
* 2. the majority based off what the leader's voter set was at the HWM (i.e. the prior lastVoterSet() value)
* This is because verify() may be called after the leader's in-memory voter set was updated but before the voter change was committed. A perfect invariant checker would use what has been committed to the log but that information is not readily exposed for testing.
```
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests pt 1 [kafka]
Posted by "kevin-wu24 (via GitHub)" <gi...@apache.org>.
kevin-wu24 commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r1970773425
##########
raft/src/main/java/org/apache/kafka/raft/VoterSet.java:
##########
@@ -343,6 +343,10 @@ public ReplicaKey voterKey() {
return voterKey;
}
+ public int voterId() {
Review Comment:
I think I just have to refactor `voterSetFromIds` to not be a one liner, and yes we can.
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests pt 1 [kafka]
Posted by "kevin-wu24 (via GitHub)" <gi...@apache.org>.
kevin-wu24 commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r1980456640
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -297,18 +309,83 @@ void canMakeProgressIfMajorityIsReachable(
scheduler.runUntil(() -> cluster.allReachedHighWatermark(2 * restoredLogEndOffset));
}
+ @Property(tries = 100, afterFailure = AfterFailureMode.SAMPLE_ONLY)
+ void canAddVotersIfMajorityIsReachable(
+ @ForAll int seed,
+ @ForAll @IntRange(min = 1, max = 3) int numObservers
+ ) {
+ int numVoters = 5;
+ Random random = new Random(seed);
+ Cluster cluster = new Cluster(numVoters, numObservers, random, true);
+ MessageRouter router = new MessageRouter(cluster);
+ EventScheduler scheduler = schedulerWithDefaultInvariants(cluster);
+ Set<Integer> expectedVoterIds = new HashSet<>(cluster.initialVoters.keySet());
+
+ initializeClusterAndStartAppending(cluster, router, scheduler, 10);
+
+ // Partition the nodes into two sets. Nodes are reachable within each set,
+ // but the two sets cannot communicate with each other. We should be able
+ // to make progress even if an election is needed in the larger set.
+ int firstObserverId = numVoters;
+ router.filter(
+ 0,
+ new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(2, 3, 4, firstObserverId)))
+ );
+ router.filter(
+ 1,
+ new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(2, 3, 4, firstObserverId)))
+ );
+ router.filter(2, new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(0, 1))));
+ router.filter(3, new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(0, 1))));
+ router.filter(4, new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(0, 1))));
+ router.filter(firstObserverId, new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(0, 1))));
+
+ long partitionLogEndOffset = cluster.maxLogEndOffset();
+ scheduler.runUntil(() -> cluster.anyReachedHighWatermark(2 * partitionLogEndOffset));
+
+ long minorityHighWatermark = cluster.maxHighWatermarkReached(Set.of(0, 1));
+ long majorityHighWatermark = cluster.maxHighWatermarkReached(Set.of(2, 3, 4));
+
+ assertTrue(
Review Comment:
I just copied this from the above test: `canMakeProgressIfMajorityIsReachable`.
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests pt 1 [kafka]
Posted by "ahuang98 (via GitHub)" <gi...@apache.org>.
ahuang98 commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r1980407781
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -117,40 +125,101 @@ public class RaftEventSimulationTest {
void canElectInitialLeader(
@ForAll int seed,
@ForAll @IntRange(min = 1, max = 5) int numVoters,
- @ForAll @IntRange(min = 0, max = 5) int numObservers
+ @ForAll @IntRange(min = 0, max = 5) int numObservers,
+ @ForAll boolean withKip853
) {
Random random = new Random(seed);
- Cluster cluster = new Cluster(numVoters, numObservers, random);
+ Cluster cluster = new Cluster(numVoters, numObservers, random, withKip853);
MessageRouter router = new MessageRouter(cluster);
EventScheduler scheduler = schedulerWithDefaultInvariants(cluster);
- cluster.startAll();
- schedulePolling(scheduler, cluster, 3, 5);
- scheduler.schedule(router::deliverAll, 0, 2, 1);
- scheduler.schedule(new SequentialAppendAction(cluster), 0, 2, 3);
- scheduler.runUntil(cluster::hasConsistentLeader);
- scheduler.runUntil(() -> cluster.allReachedHighWatermark(10));
+ initializeClusterAndStartAppending(cluster, router, scheduler, 2);
+ }
+
+ @Property(tries = 100, afterFailure = AfterFailureMode.SAMPLE_ONLY)
+ void canElectInitialLeaderAndAddVoter(
+ @ForAll int seed,
+ @ForAll @IntRange(min = 1, max = 5) int numVoters,
+ @ForAll @IntRange(min = 1, max = 5) int numObservers
+ ) {
+ Random random = new Random(seed);
+ Cluster cluster = new Cluster(numVoters, numObservers, random, true);
+ MessageRouter router = new MessageRouter(cluster);
+ EventScheduler scheduler = schedulerWithKip853Invariants(cluster);
+
+ initializeClusterAndStartAppending(cluster, router, scheduler, 10);
+ int firstObserverId = numVoters;
+ scheduler.schedule(new AddVoterAction(cluster, cluster.running.get(firstObserverId)), 0, 5, 3);
+ scheduler.runUntil(() -> cluster.leaderWithMaxEpoch().get().client.partitionState().lastVoterSet().size() == numVoters + 1);
+ VoterSet latestVoterSet = cluster.leaderWithMaxEpoch().get().client.partitionState().lastVoterSet();
+ scheduler.runUntil(() -> cluster.allHaveLatestVoterSet(latestVoterSet));
+ }
+
+ @Property(tries = 100, afterFailure = AfterFailureMode.SAMPLE_ONLY)
+ void canAddAndRemoveVoters(
Review Comment:
> I remove a random node so that the tries property means we will remove a leader with probability = 1/numVoters on each try.
which test will do this? this one just removes all observers and I see another which partitions a voter (would lose leadership) then removes it
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests pt 1 [kafka]
Posted by "kevin-wu24 (via GitHub)" <gi...@apache.org>.
kevin-wu24 commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r1980422775
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -117,40 +125,101 @@ public class RaftEventSimulationTest {
void canElectInitialLeader(
@ForAll int seed,
@ForAll @IntRange(min = 1, max = 5) int numVoters,
- @ForAll @IntRange(min = 0, max = 5) int numObservers
+ @ForAll @IntRange(min = 0, max = 5) int numObservers,
+ @ForAll boolean withKip853
) {
Random random = new Random(seed);
- Cluster cluster = new Cluster(numVoters, numObservers, random);
+ Cluster cluster = new Cluster(numVoters, numObservers, random, withKip853);
MessageRouter router = new MessageRouter(cluster);
EventScheduler scheduler = schedulerWithDefaultInvariants(cluster);
- cluster.startAll();
- schedulePolling(scheduler, cluster, 3, 5);
- scheduler.schedule(router::deliverAll, 0, 2, 1);
- scheduler.schedule(new SequentialAppendAction(cluster), 0, 2, 3);
- scheduler.runUntil(cluster::hasConsistentLeader);
- scheduler.runUntil(() -> cluster.allReachedHighWatermark(10));
+ initializeClusterAndStartAppending(cluster, router, scheduler, 2);
+ }
+
+ @Property(tries = 100, afterFailure = AfterFailureMode.SAMPLE_ONLY)
+ void canElectInitialLeaderAndAddVoter(
+ @ForAll int seed,
+ @ForAll @IntRange(min = 1, max = 5) int numVoters,
+ @ForAll @IntRange(min = 1, max = 5) int numObservers
+ ) {
+ Random random = new Random(seed);
+ Cluster cluster = new Cluster(numVoters, numObservers, random, true);
+ MessageRouter router = new MessageRouter(cluster);
+ EventScheduler scheduler = schedulerWithKip853Invariants(cluster);
+
+ initializeClusterAndStartAppending(cluster, router, scheduler, 10);
+ int firstObserverId = numVoters;
+ scheduler.schedule(new AddVoterAction(cluster, cluster.running.get(firstObserverId)), 0, 5, 3);
+ scheduler.runUntil(() -> cluster.leaderWithMaxEpoch().get().client.partitionState().lastVoterSet().size() == numVoters + 1);
+ VoterSet latestVoterSet = cluster.leaderWithMaxEpoch().get().client.partitionState().lastVoterSet();
+ scheduler.runUntil(() -> cluster.allHaveLatestVoterSet(latestVoterSet));
+ }
+
+ @Property(tries = 100, afterFailure = AfterFailureMode.SAMPLE_ONLY)
+ void canAddAndRemoveVoters(
Review Comment:
Ah this must've changed since my last push. Let me add a test for specifically removing the leader.
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests [kafka]
Posted by "kevin-wu24 (via GitHub)" <gi...@apache.org>.
kevin-wu24 commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r2006278691
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -1127,14 +1331,75 @@ private MajorityReachedHighWatermark(Cluster cluster) {
@Override
public void verify() {
- cluster.leaderHighWatermark().ifPresent(highWatermark -> {
- long numReachedHighWatermark = cluster.nodes.entrySet().stream()
- .filter(entry -> cluster.voters.containsKey(entry.getKey()))
- .filter(entry -> entry.getValue().log.endOffset().offset() >= highWatermark)
- .count();
- assertTrue(
- numReachedHighWatermark >= cluster.majoritySize(),
- "Insufficient nodes have reached current high watermark");
+ if (cluster.withKip853) {
+ /*
+ * For clusters running in KIP-853 mode, we check that a majority of at least one of:
+ * 1. the leader's voter set at the HWM
+ * 2. the leader's lastVoterSet()
+ * has reached the HWM. We need to perform a more elaborate check here because in clusters where
+ * an Add/RemoveVoter request increases/decreases the majority of voters value by 1, the leader
+ * could have used either majority value to update its HWM value. This is because depending on
+ * whether the leader read the most recent VotersRecord prior to updating its HWM value, the number
+ * of nodes (the majority) used to calculate that HWM value is different. This matters for invariant
+ * checking because we perform this verification on every message delivery.
+ * */
+ cluster.leaderWithMaxEpoch().ifPresent(leaderNode -> {
+ leaderNode.client.highWatermark().ifPresent(highWatermark -> {
+ VoterSet voterSet = leaderNode.client.partitionState().lastVoterSet();
+ long numReachedHighWatermark = numReachedHighWatermark(highWatermark, voterSet.voterIds());
+ if (numReachedHighWatermark < cluster.majoritySize(voterSet.size())) {
+ leaderNode.client.partitionState().voterSetAtOffset(highWatermark - 1).ifPresent(otherVoterSet -> {
Review Comment:
~~I agree that this minimal voter set check is valid, but I think it is too relaxed for a cluster with dynamic voters. Specifically, I have concerns about this statement, but maybe I'm misunderstanding what is going on:
> KRaft is allowed to commit up to the HWM between voterSet1 or voterSet2, with either voter set.
From my understanding of the code, once KRaft becomes aware of voterSet2, which happens at the following point in execution because `partitionState.lastVoterSet()` is set to voterSet2: `appendAsLeader -> partitionState.updateState()`, it should be using voterSet2 to calculate the HWM thereafter when it handles `FetchRequests` in subsequent events.
For example, in the case of adding a voter, voterSet1 is the minimal voter set with majority size X, and voterSet2 is the uncommitted voter set with majority size X + 1. This means when verifying the HWM value, we would be checking that X nodes have replicated up to the HWM offset, not X + 1 nodes, which should only happen if the HWM value was actually calculated with voterSet1.~~
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests pt 1 [kafka]
Posted by "kevin-wu24 (via GitHub)" <gi...@apache.org>.
kevin-wu24 commented on PR #18987:
URL: https://github.com/apache/kafka/pull/18987#issuecomment-2688961782
https://github.com/apache/kafka/pull/18987#discussion_r1971722138 contains the most recent observation on what is going on when we're checking the `MajorityReachedHighWatermark` invariant when adding/removing a voter.
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests pt 1 [kafka]
Posted by "kevin-wu24 (via GitHub)" <gi...@apache.org>.
kevin-wu24 commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r1976011663
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -117,40 +125,101 @@ public class RaftEventSimulationTest {
void canElectInitialLeader(
@ForAll int seed,
@ForAll @IntRange(min = 1, max = 5) int numVoters,
- @ForAll @IntRange(min = 0, max = 5) int numObservers
+ @ForAll @IntRange(min = 0, max = 5) int numObservers,
+ @ForAll boolean withKip853
) {
Random random = new Random(seed);
- Cluster cluster = new Cluster(numVoters, numObservers, random);
+ Cluster cluster = new Cluster(numVoters, numObservers, random, withKip853);
MessageRouter router = new MessageRouter(cluster);
EventScheduler scheduler = schedulerWithDefaultInvariants(cluster);
- cluster.startAll();
- schedulePolling(scheduler, cluster, 3, 5);
- scheduler.schedule(router::deliverAll, 0, 2, 1);
- scheduler.schedule(new SequentialAppendAction(cluster), 0, 2, 3);
- scheduler.runUntil(cluster::hasConsistentLeader);
- scheduler.runUntil(() -> cluster.allReachedHighWatermark(10));
+ initializeClusterAndStartAppending(cluster, router, scheduler, 2);
+ }
+
+ @Property(tries = 100, afterFailure = AfterFailureMode.SAMPLE_ONLY)
+ void canElectInitialLeaderAndAddVoter(
+ @ForAll int seed,
+ @ForAll @IntRange(min = 1, max = 5) int numVoters,
+ @ForAll @IntRange(min = 1, max = 5) int numObservers
+ ) {
+ Random random = new Random(seed);
+ Cluster cluster = new Cluster(numVoters, numObservers, random, true);
+ MessageRouter router = new MessageRouter(cluster);
+ EventScheduler scheduler = schedulerWithKip853Invariants(cluster);
+
+ initializeClusterAndStartAppending(cluster, router, scheduler, 10);
+ int firstObserverId = numVoters;
+ scheduler.schedule(new AddVoterAction(cluster, cluster.running.get(firstObserverId)), 0, 5, 3);
+ scheduler.runUntil(() -> cluster.leaderWithMaxEpoch().get().client.partitionState().lastVoterSet().size() == numVoters + 1);
+ VoterSet latestVoterSet = cluster.leaderWithMaxEpoch().get().client.partitionState().lastVoterSet();
+ scheduler.runUntil(() -> cluster.allHaveLatestVoterSet(latestVoterSet));
+ }
+
+ @Property(tries = 100, afterFailure = AfterFailureMode.SAMPLE_ONLY)
+ void canAddAndRemoveVoters(
+ @ForAll int seed,
+ @ForAll @IntRange(min = 1, max = 5) int numVoters,
+ @ForAll @IntRange(min = 1, max = 10) int numObservers
+ ) {
+ Random random = new Random(seed);
+ Cluster cluster = new Cluster(numVoters, numObservers, random, true);
+ MessageRouter router = new MessageRouter(cluster);
+ EventScheduler scheduler = schedulerWithKip853Invariants(cluster);
+
+ initializeClusterAndStartAppending(cluster, router, scheduler, 10);
+
+ // Schedule all the observers to be added as voters
+ for (int voterIdToAdd = numVoters; voterIdToAdd < numVoters + numObservers; voterIdToAdd++) {
+ int nextVoterSetSize = voterIdToAdd + 1;
+ scheduler.schedule(new AddVoterAction(cluster, cluster.running.get(voterIdToAdd)), 0, 5, 3);
+ scheduler.runUntil(() -> cluster.leaderWithMaxEpoch().get().client.partitionState().lastVoterSet().size() == nextVoterSetSize);
+ }
+ int voterSetSize = numVoters + numObservers;
+
+ // Schedule all voters besides the first to be removed
+ for (int voterIdToRemove = numVoters + numObservers - 1; voterIdToRemove > 0; voterIdToRemove--) {
+ int nextVoterSetSize = voterSetSize - 1;
+ scheduler.schedule(new RemoveVoterAction(cluster, cluster.running.get(voterIdToRemove)), 0, 5, 3);
+ if (cluster.leaderWithMaxEpoch().isEmpty()) {
+ scheduler.runUntil(cluster::hasConsistentLeader);
+ }
+ scheduler.runUntil(() -> cluster.leaderWithMaxEpoch().get().client.partitionState().lastVoterSet().size() == nextVoterSetSize);
+ }
+ }
+
+ @Property(tries = 100, afterFailure = AfterFailureMode.SAMPLE_ONLY)
+ void canElectInitialLeaderAndRemoveVoter(
Review Comment:
I can remove this and `canElectInitialLeaderAndAddVoter`. They were mostly like sanity check tests when setting the code changes up.
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests pt 1 [kafka]
Posted by "kevin-wu24 (via GitHub)" <gi...@apache.org>.
kevin-wu24 commented on PR #18987:
URL: https://github.com/apache/kafka/pull/18987#issuecomment-2682522812
@ahuang98 thanks for the quick review. I took another look into that invariant and I'll address the last two comments here since the situation is a bit complicated. @jsancio too since this may be missing in the current KIP-853 implementation. Let me know what y'all think:
I agree, we should only be checking this invariant using the leader's `lastVoterSet()`. When I remove the extra case checking the committed voter set (voter set at HWM - 1), we fail this invariant check in the following addVoter scenario:
- We're adding a voter that will increase the number of nodes that are considered a majority (i.e. old voter set = {0}, new voter set = {0, 1})
- The events involved are: `SequentialAppendAction`, an event which repeatedly gets executed, and my implementation of `AddVoterAction`. `SequentialAppendAction` just increments a counter and writes these records to the log using `client.prepareAppend` and `client.schedulePreparedAppend`. `AddVoterAction` creates two RaftMessages and adds them to the leader's message queue, the `AddVoterRequest` and `API_VERSIONS_RESPONSE`. The problem is that I use `CompleteableFuture#whenComplete` to set the `correlationId` of the `API_VERSIONS_RESPONSE`, which is asynchronous and exit `run()` after these two messages are added to the leader's queue (i.e. the `AddVoterRequest#run()` can finish execution without completing the future, and perhaps more importantly, before the new VoterSet is added to the log, allowing subsequent `SequentialAddActions` to execute even though they shouldn't be allowed to yet).
- This section of the KIP says we should not commit records while AddVoterRequest is being handled: https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=217391519#KIP853:KRaftControllerMembershipChanges-Handling.3
- I believe the easiest way to solve this is to not allow `AddVoterAction#run()` to return until we get back an `AddVoterResponse` from the leader.
- I want to make sure that this isn't possible in the current implementation. From looking at the raft code and the fact that this is possible in the simulation tests it looks like there is no guard against this in the raft layer (i.e. my proposed fix for the simulation test I think is considered a higher layer fix, since I'm just not allowing the handling of `AddVoterRequest` and `prepareAppend + schedulePrepareAppend` to happen concurrently, but I'm not seeing anything that prevents this in the source right now).
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests [kafka]
Posted by "kevin-wu24 (via GitHub)" <gi...@apache.org>.
kevin-wu24 commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r2006278691
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -1127,14 +1331,75 @@ private MajorityReachedHighWatermark(Cluster cluster) {
@Override
public void verify() {
- cluster.leaderHighWatermark().ifPresent(highWatermark -> {
- long numReachedHighWatermark = cluster.nodes.entrySet().stream()
- .filter(entry -> cluster.voters.containsKey(entry.getKey()))
- .filter(entry -> entry.getValue().log.endOffset().offset() >= highWatermark)
- .count();
- assertTrue(
- numReachedHighWatermark >= cluster.majoritySize(),
- "Insufficient nodes have reached current high watermark");
+ if (cluster.withKip853) {
+ /*
+ * For clusters running in KIP-853 mode, we check that a majority of at least one of:
+ * 1. the leader's voter set at the HWM
+ * 2. the leader's lastVoterSet()
+ * has reached the HWM. We need to perform a more elaborate check here because in clusters where
+ * an Add/RemoveVoter request increases/decreases the majority of voters value by 1, the leader
+ * could have used either majority value to update its HWM value. This is because depending on
+ * whether the leader read the most recent VotersRecord prior to updating its HWM value, the number
+ * of nodes (the majority) used to calculate that HWM value is different. This matters for invariant
+ * checking because we perform this verification on every message delivery.
+ * */
+ cluster.leaderWithMaxEpoch().ifPresent(leaderNode -> {
+ leaderNode.client.highWatermark().ifPresent(highWatermark -> {
+ VoterSet voterSet = leaderNode.client.partitionState().lastVoterSet();
+ long numReachedHighWatermark = numReachedHighWatermark(highWatermark, voterSet.voterIds());
+ if (numReachedHighWatermark < cluster.majoritySize(voterSet.size())) {
+ leaderNode.client.partitionState().voterSetAtOffset(highWatermark - 1).ifPresent(otherVoterSet -> {
Review Comment:
~~I agree that this minimal voter set check is valid, but I think it is too relaxed for a cluster with dynamic voters. Specifically, I have concerns about this statement, but maybe I'm misunderstanding what is going on:~~
> KRaft is allowed to commit up to the HWM between voterSet1 or voterSet2, with either voter set.
~~From my understanding of the code, once KRaft becomes aware of voterSet2, which happens at the following point in execution because `partitionState.lastVoterSet()` is set to voterSet2: `appendAsLeader -> partitionState.updateState()`, it should be using voterSet2 to calculate the HWM thereafter when it handles `FetchRequests` in subsequent events.~~
~~For example, in the case of adding a voter, voterSet1 is the minimal voter set with majority size X, and voterSet2 is the uncommitted voter set with majority size X + 1. This means when verifying the HWM value, we would be checking that X nodes have replicated up to the HWM offset, not X + 1 nodes, which should only happen if the HWM value was actually calculated with voterSet1.~~
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -1127,14 +1331,75 @@ private MajorityReachedHighWatermark(Cluster cluster) {
@Override
public void verify() {
- cluster.leaderHighWatermark().ifPresent(highWatermark -> {
- long numReachedHighWatermark = cluster.nodes.entrySet().stream()
- .filter(entry -> cluster.voters.containsKey(entry.getKey()))
- .filter(entry -> entry.getValue().log.endOffset().offset() >= highWatermark)
- .count();
- assertTrue(
- numReachedHighWatermark >= cluster.majoritySize(),
- "Insufficient nodes have reached current high watermark");
+ if (cluster.withKip853) {
+ /*
+ * For clusters running in KIP-853 mode, we check that a majority of at least one of:
+ * 1. the leader's voter set at the HWM
+ * 2. the leader's lastVoterSet()
+ * has reached the HWM. We need to perform a more elaborate check here because in clusters where
+ * an Add/RemoveVoter request increases/decreases the majority of voters value by 1, the leader
+ * could have used either majority value to update its HWM value. This is because depending on
+ * whether the leader read the most recent VotersRecord prior to updating its HWM value, the number
+ * of nodes (the majority) used to calculate that HWM value is different. This matters for invariant
+ * checking because we perform this verification on every message delivery.
+ * */
+ cluster.leaderWithMaxEpoch().ifPresent(leaderNode -> {
+ leaderNode.client.highWatermark().ifPresent(highWatermark -> {
+ VoterSet voterSet = leaderNode.client.partitionState().lastVoterSet();
+ long numReachedHighWatermark = numReachedHighWatermark(highWatermark, voterSet.voterIds());
+ if (numReachedHighWatermark < cluster.majoritySize(voterSet.size())) {
+ leaderNode.client.partitionState().voterSetAtOffset(highWatermark - 1).ifPresent(otherVoterSet -> {
Review Comment:
~~If the main concern is to avoid doing a separate check when the cluster has KIP-853 enabled, I think I can just remove the if statement, using `lastVoterSet()` with a static voter set always works (honestly not sure why this was even there). This way we are not relying on the implementation detail that each voter set can only differ by 1 voter.~~
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests [kafka]
Posted by "kevin-wu24 (via GitHub)" <gi...@apache.org>.
kevin-wu24 commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r2006031583
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -475,6 +563,99 @@ void canRecoverFromSingleNodeCommittedDataLoss(
scheduler.runUntil(() -> cluster.allReachedHighWatermark(highWatermarkBeforeRestart + 10));
}
+ @Property(tries = 100, afterFailure = AfterFailureMode.SAMPLE_ONLY)
+ void canAddObserverAndRemovePartitionedVoter(
+ @ForAll int seed,
+ @ForAll @IntRange(min = 3, max = 5) int numVoters
+ ) {
+ Random random = new Random(seed);
+ Cluster cluster = new Cluster(numVoters, 1, random, true);
+ MessageRouter router = new MessageRouter(cluster);
+ EventScheduler scheduler = schedulerWithDefaultInvariants(cluster);
+ Set<Integer> expectedVoterIds = new HashSet<>(cluster.initialVoters.keySet());
+
+ initializeClusterAndStartAppending(cluster, router, scheduler, 10);
+
+ // Partition a random voter. Add first observer as new voter
+ int voterIdToRemove = cluster.running.get(random.nextInt(numVoters)).id();
+ router.filter(voterIdToRemove, new DropAllTraffic());
+ expectedVoterIds.add(numVoters);
+ addVoter(cluster, scheduler, numVoters, expectedVoterIds);
+
+ // Remove the partitioned voter
+ expectedVoterIds.remove(voterIdToRemove);
+ removeVoter(cluster, scheduler, voterIdToRemove, expectedVoterIds);
+ long highWatermark = cluster.maxHighWatermarkReached();
+ scheduler.runUntil(() -> cluster.allReachedHighWatermark(highWatermark + 10, expectedVoterIds));
+ }
+
+ @Property(tries = 100, afterFailure = AfterFailureMode.SAMPLE_ONLY)
+ void canRemoveAndAddBackVoter(
+ @ForAll int seed,
+ @ForAll @IntRange(min = 3, max = 5) int numVoters,
+ @ForAll @IntRange(min = 0, max = 2) int numObservers
+ ) {
+ Random random = new Random(seed);
+ Cluster cluster = new Cluster(numVoters, numObservers, random, true);
+ MessageRouter router = new MessageRouter(cluster);
+ EventScheduler scheduler = schedulerWithDefaultInvariants(cluster);
+ Set<Integer> expectedVoterIds = new HashSet<>(cluster.initialVoters.keySet());
+
+ initializeClusterAndStartAppending(cluster, router, scheduler, 10);
+
+ // Partition a random voter and remove it
+ int voterId = cluster.running.get(random.nextInt(numVoters)).id();
+ router.filter(voterId, new DropAllTraffic());
+ expectedVoterIds.remove(voterId);
+ removeVoter(cluster, scheduler, voterId, expectedVoterIds);
+
+ // Restore the network and add the voter back
+ router.filter(voterId, new PermitAllTraffic());
+ expectedVoterIds.add(voterId);
+ addVoter(cluster, scheduler, voterId, expectedVoterIds);
+
+ long highWatermark = cluster.maxHighWatermarkReached();
+ scheduler.runUntil(() -> cluster.allReachedHighWatermark(highWatermark + 10));
+ }
+
+ /**
+ * This method initializes the cluster and starts appending to the log.
+ * @param appendActionPeriodMs - the period at which the append action should be scheduled, in milliseconds
+ * This value is important for tests that add voters, since if the cluster is appending
+ * records too quickly, it may prevent the new voter from catching up to the leader's LEO
+ * so that it can be added to the voter set.
+ */
+ private void initializeClusterAndStartAppending(Cluster cluster, MessageRouter router, EventScheduler scheduler, int appendActionPeriodMs) {
+ cluster.startAll();
+ schedulePolling(scheduler, cluster, 3, 5);
+ scheduler.schedule(router::deliverAll, 0, 2, 5);
+ scheduler.schedule(new SequentialAppendAction(cluster), 0, appendActionPeriodMs, 3);
+ scheduler.runUntil(cluster::hasConsistentLeader);
+ scheduler.runUntil(() -> cluster.anyReachedHighWatermark(10));
+ }
+
+ private void addVoter(Cluster cluster, EventScheduler scheduler, int idToAdd, Set<Integer> expectedVoterIds) {
+ scheduler.scheduleUntil(
+ new AddVoterAction(cluster, cluster.running.get(idToAdd)),
+ () -> cluster.leaderHasCommittedVoterSet(expectedVoterIds),
+ 0,
+ 5,
+ 3
+ );
Review Comment:
I remember running into a lot of issues when trying to use `runUntil` directly. Mainly because we need to do two things:
1. repeatedly schedule the operation that wraps an Add/RemoveVoter RPC completes successfully
2. stop scheduling that operation
It's not sufficient to use the existing `EventScheduler#schedule`, because that schedules the RPC from that point thereafter. What happens if I want to test a case like removing and adding back the same voter?
Maybe the naming `scheduleUntil` is not clear, but it means that we schedule the event until some condition is met, then we stop scheduling that event.
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests pt 1 [kafka]
Posted by "kevin-wu24 (via GitHub)" <gi...@apache.org>.
kevin-wu24 commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r1970056520
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -1127,14 +1331,75 @@ private MajorityReachedHighWatermark(Cluster cluster) {
@Override
public void verify() {
- cluster.leaderHighWatermark().ifPresent(highWatermark -> {
- long numReachedHighWatermark = cluster.nodes.entrySet().stream()
- .filter(entry -> cluster.voters.containsKey(entry.getKey()))
- .filter(entry -> entry.getValue().log.endOffset().offset() >= highWatermark)
- .count();
- assertTrue(
- numReachedHighWatermark >= cluster.majoritySize(),
- "Insufficient nodes have reached current high watermark");
+ if (cluster.withKip853) {
+ /*
+ * For clusters running in KIP-853 mode, we check that a majority of at least one of:
+ * 1. the leader's voter set at the HWM
+ * 2. the leader's lastVoterSet()
+ * has reached the HWM. We need to perform a more elaborate check here because in clusters where
+ * an Add/RemoveVoter request increases/decreases the majority of voters value by 1, the leader
+ * could have used either majority value to update its HWM value. This is because depending on
+ * whether the leader read the most recent VotersRecord prior to updating its HWM value, the number
+ * of nodes (the majority) used to calculate that HWM value is different. This matters for invariant
+ * checking because we perform this verification on every message delivery.
+ * */
+ cluster.leaderWithMaxEpoch().ifPresent(leaderNode -> {
+ leaderNode.client.highWatermark().ifPresent(highWatermark -> {
+ VoterSet voterSet = leaderNode.client.partitionState().lastVoterSet();
+ long numReachedHighWatermark = numReachedHighWatermark(highWatermark, voterSet.voterIds());
+ if (numReachedHighWatermark < cluster.majoritySize(voterSet.size())) {
+ leaderNode.client.partitionState().voterSetAtOffset(highWatermark - 1).ifPresent(otherVoterSet -> {
Review Comment:
An "event" from the perspective of the simulation tests is the execution of one `Runnable`. This detail is important because `SequentialAppendAction` will directly call `client.prepareAppend` and `client.schedulePreparedAppend` which can wakeup the message queue (this is where my implementation of `AddVoterAction` adds the voter-to-be-added's `AddVoterRequest` and `API_VERSIONS_RESPONSE`).
This check is incorrect, and we should rely on only the last voter set. More explained below
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests pt 1 [kafka]
Posted by "ahuang98 (via GitHub)" <gi...@apache.org>.
ahuang98 commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r1970747972
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -487,6 +555,12 @@ private EventScheduler schedulerWithDefaultInvariants(Cluster cluster) {
return scheduler;
}
+ private EventScheduler schedulerWithKip853Invariants(Cluster cluster) {
+ EventScheduler scheduler = schedulerWithDefaultInvariants(cluster);
+ scheduler.addInvariant(new AtMostOneUncommittedVoterSet(cluster));
Review Comment:
seems like an issue if it breaks non-KIP-853 tests - maybe worth at least adding it to the default invariants locally and running everything to see if any tests break - technically this new version of the invariant is more lenient than the old version right?
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests pt 1 [kafka]
Posted by "kevin-wu24 (via GitHub)" <gi...@apache.org>.
kevin-wu24 commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r1970766229
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -487,6 +555,12 @@ private EventScheduler schedulerWithDefaultInvariants(Cluster cluster) {
return scheduler;
}
+ private EventScheduler schedulerWithKip853Invariants(Cluster cluster) {
+ EventScheduler scheduler = schedulerWithDefaultInvariants(cluster);
+ scheduler.addInvariant(new AtMostOneUncommittedVoterSet(cluster));
Review Comment:
I will have to refactor the `voterSetFromIds()` method to actually return something a non-empty optional if we add this invariant for non-KIP-853 tests, as that would throw an exception during verification.
My main reason for not adding it initially is that it is trivially invariant for a cluster not running in KIP-853 mode, and that under the current implementation of the verification, we have to iterate over the offsets in the log between the HWM and the offset where the leader's `lastVoterSet()` is. Also, the current implementation has a typo, which I'll fix.
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests pt 1 [kafka]
Posted by "kevin-wu24 (via GitHub)" <gi...@apache.org>.
kevin-wu24 commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r1980466623
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -475,6 +537,104 @@ void canRecoverFromSingleNodeCommittedDataLoss(
scheduler.runUntil(() -> cluster.allReachedHighWatermark(highWatermarkBeforeRestart + 10));
}
+ @Property(tries = 100, afterFailure = AfterFailureMode.SAMPLE_ONLY)
+ void canAddObserverAndRemovePartitionedVoter(
+ @ForAll int seed,
+ @ForAll @IntRange(min = 3, max = 5) int numVoters
+ ) {
+ Random random = new Random(seed);
+ Cluster cluster = new Cluster(numVoters, 1, random, true);
+ MessageRouter router = new MessageRouter(cluster);
+ EventScheduler scheduler = schedulerWithDefaultInvariants(cluster);
+ Set<Integer> expectedVoterIds = new HashSet<>(cluster.initialVoters.keySet());
+
+ initializeClusterAndStartAppending(cluster, router, scheduler, 10);
+
+ // Partition a random voter, and add an observer to replace it
+ RaftNode voterToRemove = cluster.running.get(random.nextInt(numVoters));
+ router.filter(voterToRemove.id(), new DropAllTraffic());
+ addVoter(cluster, scheduler, numVoters, expectedVoterIds);
Review Comment:
see https://github.com/apache/kafka/pull/18987#discussion_r1980455784
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests [kafka]
Posted by "kevin-wu24 (via GitHub)" <gi...@apache.org>.
kevin-wu24 commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r2006049419
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -645,39 +837,56 @@ private static class Cluster {
final AtomicInteger correlationIdCounter = new AtomicInteger();
final MockTime time = new MockTime();
final String clusterId = Uuid.randomUuid().toString();
- final Map<Integer, Node> voters = new HashMap<>();
- final Map<Integer, PersistentState> nodes = new HashMap<>();
+ final Map<Integer, Node> initialVoters = new HashMap<>();
+ final Map<Integer, PersistentState> persistentStates = new HashMap<>();
final Map<Integer, RaftNode> running = new HashMap<>();
+ final boolean withKip853;
- private Cluster(int numVoters, int numObservers, Random random) {
+ private Cluster(int numVoters, int numObservers, Random random, boolean withKip853) {
this.random = random;
+ this.withKip853 = withKip853;
for (int nodeId = 0; nodeId < numVoters; nodeId++) {
- voters.put(nodeId, nodeFromId(nodeId));
- nodes.put(nodeId, new PersistentState(nodeId));
+ initialVoters.put(nodeId, nodeFromId(nodeId));
+ persistentStates.put(nodeId, new PersistentState(nodeId));
}
for (int nodeIdDelta = 0; nodeIdDelta < numObservers; nodeIdDelta++) {
int nodeId = numVoters + nodeIdDelta;
- nodes.put(nodeId, new PersistentState(nodeId));
+ persistentStates.put(nodeId, new PersistentState(nodeId));
}
}
Set<InetSocketAddress> endpointsFromIds(Set<Integer> nodeIds) {
- return voters
+ return running
.values()
.stream()
- .filter(node -> nodeIds.contains(node.id()))
+ .filter(node -> nodeIds.contains(node.nodeId))
.map(Cluster::nodeAddress)
.collect(Collectors.toSet());
}
+ Optional<VoterSet> initialVoterSetFromIds() {
Review Comment:
Yeah, the IDs from `initialVoterSet`. Initially, this function returned an empty `Optional` when KIP-853 was not enabled, which is what Alyssa was referring to.
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests [kafka]
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r2003697404
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -1127,14 +1331,75 @@ private MajorityReachedHighWatermark(Cluster cluster) {
@Override
public void verify() {
- cluster.leaderHighWatermark().ifPresent(highWatermark -> {
- long numReachedHighWatermark = cluster.nodes.entrySet().stream()
- .filter(entry -> cluster.voters.containsKey(entry.getKey()))
- .filter(entry -> entry.getValue().log.endOffset().offset() >= highWatermark)
- .count();
- assertTrue(
- numReachedHighWatermark >= cluster.majoritySize(),
- "Insufficient nodes have reached current high watermark");
+ if (cluster.withKip853) {
+ /*
+ * For clusters running in KIP-853 mode, we check that a majority of at least one of:
+ * 1. the leader's voter set at the HWM
+ * 2. the leader's lastVoterSet()
+ * has reached the HWM. We need to perform a more elaborate check here because in clusters where
+ * an Add/RemoveVoter request increases/decreases the majority of voters value by 1, the leader
+ * could have used either majority value to update its HWM value. This is because depending on
+ * whether the leader read the most recent VotersRecord prior to updating its HWM value, the number
Review Comment:
What is the summary of the discussion?
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests [kafka]
Posted by "kevin-wu24 (via GitHub)" <gi...@apache.org>.
kevin-wu24 commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r2006031583
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -475,6 +563,99 @@ void canRecoverFromSingleNodeCommittedDataLoss(
scheduler.runUntil(() -> cluster.allReachedHighWatermark(highWatermarkBeforeRestart + 10));
}
+ @Property(tries = 100, afterFailure = AfterFailureMode.SAMPLE_ONLY)
+ void canAddObserverAndRemovePartitionedVoter(
+ @ForAll int seed,
+ @ForAll @IntRange(min = 3, max = 5) int numVoters
+ ) {
+ Random random = new Random(seed);
+ Cluster cluster = new Cluster(numVoters, 1, random, true);
+ MessageRouter router = new MessageRouter(cluster);
+ EventScheduler scheduler = schedulerWithDefaultInvariants(cluster);
+ Set<Integer> expectedVoterIds = new HashSet<>(cluster.initialVoters.keySet());
+
+ initializeClusterAndStartAppending(cluster, router, scheduler, 10);
+
+ // Partition a random voter. Add first observer as new voter
+ int voterIdToRemove = cluster.running.get(random.nextInt(numVoters)).id();
+ router.filter(voterIdToRemove, new DropAllTraffic());
+ expectedVoterIds.add(numVoters);
+ addVoter(cluster, scheduler, numVoters, expectedVoterIds);
+
+ // Remove the partitioned voter
+ expectedVoterIds.remove(voterIdToRemove);
+ removeVoter(cluster, scheduler, voterIdToRemove, expectedVoterIds);
+ long highWatermark = cluster.maxHighWatermarkReached();
+ scheduler.runUntil(() -> cluster.allReachedHighWatermark(highWatermark + 10, expectedVoterIds));
+ }
+
+ @Property(tries = 100, afterFailure = AfterFailureMode.SAMPLE_ONLY)
+ void canRemoveAndAddBackVoter(
+ @ForAll int seed,
+ @ForAll @IntRange(min = 3, max = 5) int numVoters,
+ @ForAll @IntRange(min = 0, max = 2) int numObservers
+ ) {
+ Random random = new Random(seed);
+ Cluster cluster = new Cluster(numVoters, numObservers, random, true);
+ MessageRouter router = new MessageRouter(cluster);
+ EventScheduler scheduler = schedulerWithDefaultInvariants(cluster);
+ Set<Integer> expectedVoterIds = new HashSet<>(cluster.initialVoters.keySet());
+
+ initializeClusterAndStartAppending(cluster, router, scheduler, 10);
+
+ // Partition a random voter and remove it
+ int voterId = cluster.running.get(random.nextInt(numVoters)).id();
+ router.filter(voterId, new DropAllTraffic());
+ expectedVoterIds.remove(voterId);
+ removeVoter(cluster, scheduler, voterId, expectedVoterIds);
+
+ // Restore the network and add the voter back
+ router.filter(voterId, new PermitAllTraffic());
+ expectedVoterIds.add(voterId);
+ addVoter(cluster, scheduler, voterId, expectedVoterIds);
+
+ long highWatermark = cluster.maxHighWatermarkReached();
+ scheduler.runUntil(() -> cluster.allReachedHighWatermark(highWatermark + 10));
+ }
+
+ /**
+ * This method initializes the cluster and starts appending to the log.
+ * @param appendActionPeriodMs - the period at which the append action should be scheduled, in milliseconds
+ * This value is important for tests that add voters, since if the cluster is appending
+ * records too quickly, it may prevent the new voter from catching up to the leader's LEO
+ * so that it can be added to the voter set.
+ */
+ private void initializeClusterAndStartAppending(Cluster cluster, MessageRouter router, EventScheduler scheduler, int appendActionPeriodMs) {
+ cluster.startAll();
+ schedulePolling(scheduler, cluster, 3, 5);
+ scheduler.schedule(router::deliverAll, 0, 2, 5);
+ scheduler.schedule(new SequentialAppendAction(cluster), 0, appendActionPeriodMs, 3);
+ scheduler.runUntil(cluster::hasConsistentLeader);
+ scheduler.runUntil(() -> cluster.anyReachedHighWatermark(10));
+ }
+
+ private void addVoter(Cluster cluster, EventScheduler scheduler, int idToAdd, Set<Integer> expectedVoterIds) {
+ scheduler.scheduleUntil(
+ new AddVoterAction(cluster, cluster.running.get(idToAdd)),
+ () -> cluster.leaderHasCommittedVoterSet(expectedVoterIds),
+ 0,
+ 5,
+ 3
+ );
Review Comment:
I remember running into a lot of issues when trying to use `runUntil` directly. Mainly because we need to do two things:
1. repeatedly schedule the operation until an RPC completes successfully
2. stop scheduling that operation
It's not sufficient to use the existing `EventScheduler#schedule`, because that schedules the RPC from that point thereafter. What happens if I want to test a case like removing and adding back the same voter?
Maybe the naming `scheduleUntil` is not clear, but it means that we schedule the event until some condition is met, then we stop scheduling that event.
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests [kafka]
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r2004145500
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -475,6 +563,99 @@ void canRecoverFromSingleNodeCommittedDataLoss(
scheduler.runUntil(() -> cluster.allReachedHighWatermark(highWatermarkBeforeRestart + 10));
}
+ @Property(tries = 100, afterFailure = AfterFailureMode.SAMPLE_ONLY)
+ void canAddObserverAndRemovePartitionedVoter(
+ @ForAll int seed,
+ @ForAll @IntRange(min = 3, max = 5) int numVoters
+ ) {
+ Random random = new Random(seed);
+ Cluster cluster = new Cluster(numVoters, 1, random, true);
+ MessageRouter router = new MessageRouter(cluster);
+ EventScheduler scheduler = schedulerWithDefaultInvariants(cluster);
+ Set<Integer> expectedVoterIds = new HashSet<>(cluster.initialVoters.keySet());
+
+ initializeClusterAndStartAppending(cluster, router, scheduler, 10);
+
+ // Partition a random voter. Add first observer as new voter
+ int voterIdToRemove = cluster.running.get(random.nextInt(numVoters)).id();
+ router.filter(voterIdToRemove, new DropAllTraffic());
+ expectedVoterIds.add(numVoters);
+ addVoter(cluster, scheduler, numVoters, expectedVoterIds);
+
+ // Remove the partitioned voter
+ expectedVoterIds.remove(voterIdToRemove);
+ removeVoter(cluster, scheduler, voterIdToRemove, expectedVoterIds);
+ long highWatermark = cluster.maxHighWatermarkReached();
+ scheduler.runUntil(() -> cluster.allReachedHighWatermark(highWatermark + 10, expectedVoterIds));
+ }
+
+ @Property(tries = 100, afterFailure = AfterFailureMode.SAMPLE_ONLY)
+ void canRemoveAndAddBackVoter(
+ @ForAll int seed,
+ @ForAll @IntRange(min = 3, max = 5) int numVoters,
+ @ForAll @IntRange(min = 0, max = 2) int numObservers
+ ) {
+ Random random = new Random(seed);
+ Cluster cluster = new Cluster(numVoters, numObservers, random, true);
+ MessageRouter router = new MessageRouter(cluster);
+ EventScheduler scheduler = schedulerWithDefaultInvariants(cluster);
+ Set<Integer> expectedVoterIds = new HashSet<>(cluster.initialVoters.keySet());
+
+ initializeClusterAndStartAppending(cluster, router, scheduler, 10);
+
+ // Partition a random voter and remove it
+ int voterId = cluster.running.get(random.nextInt(numVoters)).id();
+ router.filter(voterId, new DropAllTraffic());
+ expectedVoterIds.remove(voterId);
+ removeVoter(cluster, scheduler, voterId, expectedVoterIds);
+
+ // Restore the network and add the voter back
+ router.filter(voterId, new PermitAllTraffic());
+ expectedVoterIds.add(voterId);
+ addVoter(cluster, scheduler, voterId, expectedVoterIds);
+
+ long highWatermark = cluster.maxHighWatermarkReached();
+ scheduler.runUntil(() -> cluster.allReachedHighWatermark(highWatermark + 10));
+ }
+
+ /**
+ * This method initializes the cluster and starts appending to the log.
+ * @param appendActionPeriodMs - the period at which the append action should be scheduled, in milliseconds
+ * This value is important for tests that add voters, since if the cluster is appending
+ * records too quickly, it may prevent the new voter from catching up to the leader's LEO
+ * so that it can be added to the voter set.
+ */
+ private void initializeClusterAndStartAppending(Cluster cluster, MessageRouter router, EventScheduler scheduler, int appendActionPeriodMs) {
+ cluster.startAll();
+ schedulePolling(scheduler, cluster, 3, 5);
+ scheduler.schedule(router::deliverAll, 0, 2, 5);
+ scheduler.schedule(new SequentialAppendAction(cluster), 0, appendActionPeriodMs, 3);
+ scheduler.runUntil(cluster::hasConsistentLeader);
+ scheduler.runUntil(() -> cluster.anyReachedHighWatermark(10));
+ }
+
+ private void addVoter(Cluster cluster, EventScheduler scheduler, int idToAdd, Set<Integer> expectedVoterIds) {
+ scheduler.scheduleUntil(
+ new AddVoterAction(cluster, cluster.running.get(idToAdd)),
+ () -> cluster.leaderHasCommittedVoterSet(expectedVoterIds),
+ 0,
+ 5,
+ 3
+ );
Review Comment:
Do you need `EventScheduler::scheduleUntil`? The scheduler is not a wall-clock or concurrent scheduler. It uses a mocked implementation of time. For example, did you consider performing the add voter action in the `runUntil` below?
You can run until the add voter RPC has been completed. If you look at `RaftMessage.Inbound`, there is a future that will be compleded when the `RaftMessage` has been handled.
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -645,39 +929,52 @@ private static class Cluster {
final AtomicInteger correlationIdCounter = new AtomicInteger();
final MockTime time = new MockTime();
final String clusterId = Uuid.randomUuid().toString();
- final Map<Integer, Node> voters = new HashMap<>();
- final Map<Integer, PersistentState> nodes = new HashMap<>();
+ final Map<Integer, Node> initialVoters = new HashMap<>();
+ final Map<Integer, PersistentState> persistentStates = new HashMap<>();
final Map<Integer, RaftNode> running = new HashMap<>();
+ final boolean withKip853;
- private Cluster(int numVoters, int numObservers, Random random) {
+ private Cluster(int numVoters, int numObservers, Random random, boolean withKip853) {
this.random = random;
+ this.withKip853 = withKip853;
for (int nodeId = 0; nodeId < numVoters; nodeId++) {
- voters.put(nodeId, nodeFromId(nodeId));
- nodes.put(nodeId, new PersistentState(nodeId));
+ initialVoters.put(nodeId, nodeFromId(nodeId));
+ persistentStates.put(nodeId, new PersistentState(nodeId));
}
for (int nodeIdDelta = 0; nodeIdDelta < numObservers; nodeIdDelta++) {
int nodeId = numVoters + nodeIdDelta;
- nodes.put(nodeId, new PersistentState(nodeId));
+ persistentStates.put(nodeId, new PersistentState(nodeId));
}
}
Set<InetSocketAddress> endpointsFromIds(Set<Integer> nodeIds) {
- return voters
+ return running
.values()
.stream()
- .filter(node -> nodeIds.contains(node.id()))
+ .filter(node -> nodeIds.contains(node.nodeId))
.map(Cluster::nodeAddress)
.collect(Collectors.toSet());
}
+ Optional<VoterSet> initialVoterSetFromIds() {
+ return Optional.of(VoterSet.fromMap(initialVoters
+ .values()
+ .stream()
+ .map(node -> new VoterSet.VoterNode(
+ ReplicaKey.of(node.id(), persistentStates.get(node.id()).nodeDirectoryId),
+ endpointsFromId(node.id(), MockNetworkChannel.LISTENER_NAME),
+ new SupportedVersionRange((short) 1)
+ )).collect(Collectors.toMap(voterNode -> voterNode.voterKey().id(), Function.identity()))));
+ }
+
Set<Integer> nodeIds() {
- return nodes.keySet();
+ return persistentStates.keySet();
}
- int majoritySize() {
- return voters.size() / 2 + 1;
+ int majoritySize(int numberOfVoters) {
+ return numberOfVoters / 2 + 1;
}
Review Comment:
This doesn't use any object fields. This can be a static method. Might as well move this to `RaftTestUtils`.
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -1326,42 +1745,47 @@ private void assertCommittedData(RaftNode node) {
return;
}
- AtomicLong startOffset = new AtomicLong(0);
- log.earliestSnapshotId().ifPresent(snapshotId -> {
- assertTrue(snapshotId.offset() <= highWatermark.getAsLong());
- startOffset.set(snapshotId.offset());
+ AtomicLong startOffset = new AtomicLong(log.startOffset());
- try (SnapshotReader<Integer> snapshot = RecordsSnapshotReader.of(
+ // We do not perform this check with KIP-853 enabled
+ // because we write the bootstrap snapshot to all initial voters
+ if (!cluster.withKip853) {
Review Comment:
This is not because of KIP-853. This is because the bootstrap checkpoint (`offset == 0`) doesn't contain any data batches. It only contains control batches.
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -1127,14 +1477,83 @@ private MajorityReachedHighWatermark(Cluster cluster) {
@Override
public void verify() {
- cluster.leaderHighWatermark().ifPresent(highWatermark -> {
- long numReachedHighWatermark = cluster.nodes.entrySet().stream()
- .filter(entry -> cluster.voters.containsKey(entry.getKey()))
- .filter(entry -> entry.getValue().log.endOffset().offset() >= highWatermark)
- .count();
- assertTrue(
- numReachedHighWatermark >= cluster.majoritySize(),
- "Insufficient nodes have reached current high watermark");
+ if (cluster.withKip853) {
+ /*
+ * For clusters running in KIP-853 mode, we check that a majority of at least one of the following
+ * voter sets has reached the high watermark:
+ * 1. the leader's voter set at the HWM (i.e. the last committed voter set)
+ * 2. the leader's lastVoterSet() (which may or may not be committed)
+ * Note that 1 and 2 can be the same set, but when they are not, lastVoterSet() is uncommitted,
+ * which follows from the AtMostOneUncommittedVoterSet invariant.
+ *
+ * A more elaborate check is necessary for this invariant because this method can get called after the
+ * leader has updated its lastVoterSet() with a new uncommitted voter set, but before the leader has
+ * updated its high watermark using the new voter set. In this case, we need to check that the majority
+ * of the last committed voter set has reached the current high watermark.
+ * */
+ cluster.leaderWithMaxEpoch().ifPresent(leaderNode -> {
+ leaderNode.client.highWatermark().ifPresent(highWatermark -> {
+ VoterSet voterSet = leaderNode.client.partitionState().lastVoterSet();
+ long numReachedHighWatermark = numReachedHighWatermark(highWatermark, voterSet.voterIds());
+ if (numReachedHighWatermark < cluster.majoritySize(voterSet.size())) {
+ leaderNode.client.partitionState().voterSetAtOffset(highWatermark - 1).ifPresent(otherVoterSet -> {
+ long nodesReachedHighWatermark = numReachedHighWatermark(highWatermark, otherVoterSet.voterIds());
+ assertTrue(
+ nodesReachedHighWatermark >= cluster.majoritySize(otherVoterSet.size()),
+ "Insufficient nodes have reached current high watermark. Expected at least " +
+ cluster.majoritySize(otherVoterSet.size()) + " but got " + nodesReachedHighWatermark);
+ });
+ return;
+ }
+ assertTrue(
+ numReachedHighWatermark >= cluster.majoritySize(voterSet.size()),
+ "Insufficient nodes have reached current high watermark. Expected at least " +
+ cluster.majoritySize(voterSet.size()) + " but got " + numReachedHighWatermark);
+ });
+ });
+ } else {
+ cluster.leaderHighWatermark().ifPresent(highWatermark -> {
+ long numReachedHighWatermark = numReachedHighWatermark(highWatermark, cluster.initialVoters.keySet());
+ assertTrue(
+ numReachedHighWatermark >= cluster.majoritySize(cluster.initialVoters.size()),
+ "Insufficient nodes have reached current high watermark");
+ });
+ }
+ }
+
+ private long numReachedHighWatermark(long highWatermark, Set<Integer> voterIds) {
+ return cluster.persistentStates.entrySet().stream()
+ .filter(entry -> voterIds.contains(entry.getKey()))
+ .filter(entry -> entry.getValue().log.endOffset().offset() >= highWatermark)
+ .count();
+ }
+ }
+
+ private static class AtMostOneUncommittedVotersRecord implements Invariant {
+ final Cluster cluster;
+
+ private AtMostOneUncommittedVotersRecord(Cluster cluster) {
+ this.cluster = cluster;
+ }
+
+ @Override
+ public void verify() {
+ cluster.leaderWithMaxEpoch().ifPresent(leaderNode -> {
+ leaderNode.log.readBatches(leaderNode.highWatermark(), OptionalLong.of(leaderNode.logEndOffset())).forEach(batch -> {
+ boolean seenUncommittedVotersRecord = false;
+ if (batch.isControlBatch) {
+ for (LogEntry entry : batch.entries) {
+ short typeId = ControlRecordType.parseTypeId(entry.record.key());
+ ControlRecordType type = ControlRecordType.fromTypeId(typeId);
+ if (type == ControlRecordType.KRAFT_VOTERS) {
+ if (seenUncommittedVotersRecord) {
+ fail("More than one uncommitted voters record found in the log");
+ }
+ seenUncommittedVotersRecord = true;
+ }
+ }
+ }
+ });
Review Comment:
This feels very implementation specific. The ideal invariant is that logs stay consistent irrespective of add/remove operations, leader election and network failures.
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -645,39 +837,56 @@ private static class Cluster {
final AtomicInteger correlationIdCounter = new AtomicInteger();
final MockTime time = new MockTime();
final String clusterId = Uuid.randomUuid().toString();
- final Map<Integer, Node> voters = new HashMap<>();
- final Map<Integer, PersistentState> nodes = new HashMap<>();
+ final Map<Integer, Node> initialVoters = new HashMap<>();
+ final Map<Integer, PersistentState> persistentStates = new HashMap<>();
final Map<Integer, RaftNode> running = new HashMap<>();
+ final boolean withKip853;
- private Cluster(int numVoters, int numObservers, Random random) {
+ private Cluster(int numVoters, int numObservers, Random random, boolean withKip853) {
this.random = random;
+ this.withKip853 = withKip853;
for (int nodeId = 0; nodeId < numVoters; nodeId++) {
- voters.put(nodeId, nodeFromId(nodeId));
- nodes.put(nodeId, new PersistentState(nodeId));
+ initialVoters.put(nodeId, nodeFromId(nodeId));
+ persistentStates.put(nodeId, new PersistentState(nodeId));
}
for (int nodeIdDelta = 0; nodeIdDelta < numObservers; nodeIdDelta++) {
int nodeId = numVoters + nodeIdDelta;
- nodes.put(nodeId, new PersistentState(nodeId));
+ persistentStates.put(nodeId, new PersistentState(nodeId));
}
}
Set<InetSocketAddress> endpointsFromIds(Set<Integer> nodeIds) {
- return voters
+ return running
.values()
.stream()
- .filter(node -> nodeIds.contains(node.id()))
+ .filter(node -> nodeIds.contains(node.nodeId))
.map(Cluster::nodeAddress)
.collect(Collectors.toSet());
}
+ Optional<VoterSet> initialVoterSetFromIds() {
Review Comment:
Did you address this comment by Alyssa?
The name doesn't seem to match the implementation. "From what ids?" Did you mean `initialVoterSet`?
##########
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java:
##########
@@ -3637,6 +3637,11 @@ QuorumState quorum() {
return quorum;
}
+ // Visible only for test
+ KRaftControlRecordStateMachine partitionState() {
Review Comment:
We should not expose this. This is very implementation specific. Did you consider adding the functionality you need to `MockLog` which is the replicated log implementation used by the simulation tests?
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -117,40 +126,83 @@ public class RaftEventSimulationTest {
void canElectInitialLeader(
@ForAll int seed,
@ForAll @IntRange(min = 1, max = 5) int numVoters,
- @ForAll @IntRange(min = 0, max = 5) int numObservers
+ @ForAll @IntRange(min = 0, max = 5) int numObservers,
+ @ForAll boolean withKip853
Review Comment:
Do you want to use `KRaftVersion` instead of a boolean?
This comment applies to a few of the tests.
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -546,6 +728,36 @@ void execute(EventScheduler scheduler) {
}
}
+ private static class ScheduledUntilConditionMetEvent extends Event {
+ final Random random;
+ final int periodMs;
+ final int jitterMs;
+ final Supplier<Boolean> exitCondition;
+
+ protected ScheduledUntilConditionMetEvent(Runnable action,
+ int eventId,
+ Random random,
+ long deadlineMs,
+ int periodMs,
+ int jitterMs,
+ Supplier<Boolean> exitCondition) {
Review Comment:
Please use this formatting:
```java
protected ScheduledUntilConditionMetEvent(
Runnable action,
int eventId,
Random random,
long deadlineMs,
int periodMs,
int jitterMs,
Supplier<Boolean> exitCondition
) {
```
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -975,7 +1297,35 @@ void poll() {
client.poll();
} while (client.isRunning() && !messageQueue.isEmpty());
} catch (Exception e) {
- throw new RuntimeException("Uncaught exception during poll of node " + nodeId, e);
+ throw new RuntimeException("Uncaught exception during poll of node " + nodeId + ": " + e, e);
+ }
+ }
+
+ /**
+ * Use this method to handle RPCs that KafkaRaftClient does not support but are necessary for testing,
+ * like API_VERSIONS when adding a voter
+ * @param request - the inbound request the RaftNode is handling
+ */
+ void handle(RaftRequest.Inbound request) {
+ ApiKeys apiKey = ApiKeys.forId(request.data().apiKey());
+ switch (apiKey) {
+ case API_VERSIONS:
+ ApiMessage apiVersionsResponse = new ApiVersionsResponse.Builder().
+ setSupportedFeatures(
+ Features.supportedFeatures(
+ Collections.singletonMap(
+ KRaftVersion.FEATURE_NAME,
+ new SupportedVersionRange(KRaftVersion.KRAFT_VERSION_1.featureLevel(), KRaftVersion.LATEST_PRODUCTION.featureLevel())
Review Comment:
```java
new SupportedVersionRange(
KRaftVersion.KRAFT_VERSION_0.featureLevel(),
KRaftVersion.LATEST_PRODUCTION.featureLevel()
)
```
Notice that the formatting change and that the min kraft version is 0 and not 1.
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -1127,14 +1331,75 @@ private MajorityReachedHighWatermark(Cluster cluster) {
@Override
public void verify() {
- cluster.leaderHighWatermark().ifPresent(highWatermark -> {
- long numReachedHighWatermark = cluster.nodes.entrySet().stream()
- .filter(entry -> cluster.voters.containsKey(entry.getKey()))
- .filter(entry -> entry.getValue().log.endOffset().offset() >= highWatermark)
- .count();
- assertTrue(
- numReachedHighWatermark >= cluster.majoritySize(),
- "Insufficient nodes have reached current high watermark");
+ if (cluster.withKip853) {
+ /*
+ * For clusters running in KIP-853 mode, we check that a majority of at least one of:
+ * 1. the leader's voter set at the HWM
+ * 2. the leader's lastVoterSet()
+ * has reached the HWM. We need to perform a more elaborate check here because in clusters where
+ * an Add/RemoveVoter request increases/decreases the majority of voters value by 1, the leader
+ * could have used either majority value to update its HWM value. This is because depending on
+ * whether the leader read the most recent VotersRecord prior to updating its HWM value, the number
+ * of nodes (the majority) used to calculate that HWM value is different. This matters for invariant
+ * checking because we perform this verification on every message delivery.
+ * */
+ cluster.leaderWithMaxEpoch().ifPresent(leaderNode -> {
+ leaderNode.client.highWatermark().ifPresent(highWatermark -> {
+ VoterSet voterSet = leaderNode.client.partitionState().lastVoterSet();
+ long numReachedHighWatermark = numReachedHighWatermark(highWatermark, voterSet.voterIds());
+ if (numReachedHighWatermark < cluster.majoritySize(voterSet.size())) {
+ leaderNode.client.partitionState().voterSetAtOffset(highWatermark - 1).ifPresent(otherVoterSet -> {
Review Comment:
The scenario that you are trying to handle is for the following log:
```
... | voterSet1 | ..... | high-watermark | ... | voterSet2 | ...
```
voterSet1 is the latest voter set before the high-watermark. This may be the static voter set.
voterSet2 is the oldest voter set after the high-watermark. It may or may not exist. It doesn't exist if dynamic voters is not supported or if the voter set has not been reconfigured.
KRaft is allowed to commit up to the HWM between voterSet1 or voterSet2, with either voter set.
Since voterSet1 and voterSet2 can only differ by at most one voter, we can define the minimal voter set as the smaller of the two voter sets. If voterSet2 doesn't exist, the minimal voter set is voterSet1.
With this definition, the high-watermark must have been replicated to the majority of the minimal voter set. What do think of this definition? This definition work with both static voters and dynamic voters. With static voters, voterSet2 never exists.
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests pt 1 [kafka]
Posted by "ahuang98 (via GitHub)" <gi...@apache.org>.
ahuang98 commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r1980500044
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -297,18 +309,83 @@ void canMakeProgressIfMajorityIsReachable(
scheduler.runUntil(() -> cluster.allReachedHighWatermark(2 * restoredLogEndOffset));
}
+ @Property(tries = 100, afterFailure = AfterFailureMode.SAMPLE_ONLY)
+ void canAddVotersIfMajorityIsReachable(
+ @ForAll int seed,
+ @ForAll @IntRange(min = 1, max = 3) int numObservers
+ ) {
+ int numVoters = 5;
+ Random random = new Random(seed);
+ Cluster cluster = new Cluster(numVoters, numObservers, random, true);
+ MessageRouter router = new MessageRouter(cluster);
+ EventScheduler scheduler = schedulerWithDefaultInvariants(cluster);
+ Set<Integer> expectedVoterIds = new HashSet<>(cluster.initialVoters.keySet());
+
+ initializeClusterAndStartAppending(cluster, router, scheduler, 10);
+
+ // Partition the nodes into two sets. Nodes are reachable within each set,
+ // but the two sets cannot communicate with each other. We should be able
+ // to make progress even if an election is needed in the larger set.
+ int firstObserverId = numVoters;
+ router.filter(
+ 0,
+ new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(2, 3, 4, firstObserverId)))
+ );
+ router.filter(
+ 1,
+ new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(2, 3, 4, firstObserverId)))
+ );
+ router.filter(2, new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(0, 1))));
+ router.filter(3, new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(0, 1))));
+ router.filter(4, new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(0, 1))));
+ router.filter(firstObserverId, new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(0, 1))));
+
+ long partitionLogEndOffset = cluster.maxLogEndOffset();
+ scheduler.runUntil(() -> cluster.anyReachedHighWatermark(2 * partitionLogEndOffset));
+
+ long minorityHighWatermark = cluster.maxHighWatermarkReached(Set.of(0, 1));
+ long majorityHighWatermark = cluster.maxHighWatermarkReached(Set.of(2, 3, 4));
+
+ assertTrue(
+ majorityHighWatermark > minorityHighWatermark,
+ String.format(
+ "majorityHighWatermark = %s, minorityHighWatermark = %s",
+ majorityHighWatermark,
+ minorityHighWatermark
+ )
+ );
+
+ // Verify we can add a voter, since a majority is still reachable
+ addVoter(cluster, scheduler, firstObserverId, expectedVoterIds);
+
+ // Now restore the partition and verify everyone catches up
+ router.filter(0, new PermitAllTraffic());
+ router.filter(1, new PermitAllTraffic());
+ router.filter(2, new PermitAllTraffic());
+ router.filter(3, new PermitAllTraffic());
+ router.filter(4, new PermitAllTraffic());
+ router.filter(firstObserverId, new PermitAllTraffic());
+
+ long restoredLogEndOffset = cluster.maxLogEndOffset();
+ scheduler.runUntil(() -> cluster.allReachedHighWatermark(2 * restoredLogEndOffset));
+ }
+
@Property(tries = 100, afterFailure = AfterFailureMode.SAMPLE_ONLY)
void leadershipAssignedOnlyOnceWithNetworkPartitionIfThereExistsMajority(
@ForAll int seed,
- @ForAll @IntRange(min = 0, max = 3) int numObservers
+ @ForAll @IntRange(min = 0, max = 3) int numObservers,
+ @ForAll boolean withKip853
) {
int numVoters = 5;
Random random = new Random(seed);
- Cluster cluster = new Cluster(numVoters, numObservers, random);
+ Cluster cluster = new Cluster(numVoters, numObservers, random, withKip853);
MessageRouter router = new MessageRouter(cluster);
EventScheduler scheduler = schedulerWithDefaultInvariants(cluster);
scheduler.addInvariant(new StableLeadership(cluster));
+ // Start cluster
+ cluster.startAll();
Review Comment:
I see, I saw that the call had been bumped up and my gut reaction was that it impacted the test - thanks for explaining
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests pt 1 [kafka]
Posted by "ahuang98 (via GitHub)" <gi...@apache.org>.
ahuang98 commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r1980496555
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -117,40 +126,58 @@ public class RaftEventSimulationTest {
void canElectInitialLeader(
@ForAll int seed,
@ForAll @IntRange(min = 1, max = 5) int numVoters,
- @ForAll @IntRange(min = 0, max = 5) int numObservers
+ @ForAll @IntRange(min = 0, max = 5) int numObservers,
+ @ForAll boolean withKip853
) {
Random random = new Random(seed);
- Cluster cluster = new Cluster(numVoters, numObservers, random);
+ Cluster cluster = new Cluster(numVoters, numObservers, random, withKip853);
MessageRouter router = new MessageRouter(cluster);
EventScheduler scheduler = schedulerWithDefaultInvariants(cluster);
- cluster.startAll();
- schedulePolling(scheduler, cluster, 3, 5);
- scheduler.schedule(router::deliverAll, 0, 2, 1);
- scheduler.schedule(new SequentialAppendAction(cluster), 0, 2, 3);
- scheduler.runUntil(cluster::hasConsistentLeader);
- scheduler.runUntil(() -> cluster.allReachedHighWatermark(10));
+ initializeClusterAndStartAppending(cluster, router, scheduler, 2);
+ }
+
+ @Property(tries = 100, afterFailure = AfterFailureMode.SAMPLE_ONLY)
+ void canAddAndRemoveVoters(
+ @ForAll int seed
+ ) {
+ Random random = new Random(seed);
+ int numVoters = 3;
+ int numObservers = 2;
+ Cluster cluster = new Cluster(numVoters, numObservers, random, true);
+ MessageRouter router = new MessageRouter(cluster);
+ EventScheduler scheduler = schedulerWithDefaultInvariants(cluster);
+ Set<Integer> expectedVoterIds = new HashSet<>(cluster.initialVoters.keySet());
+
+ initializeClusterAndStartAppending(cluster, router, scheduler, 10);
+
+ // Add all observers to voter set one by one
+ for (int voterIdToAdd = numVoters; voterIdToAdd < numVoters + numObservers; voterIdToAdd++) {
+ addVoter(cluster, scheduler, voterIdToAdd, expectedVoterIds);
+ }
+ runUntilNewHighWatermark(cluster, scheduler);
Review Comment:
can maybe get rid of the wrapper method then since it adds more obscurity w/o saving that much redundant code?
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #18987:
URL: https://github.com/apache/kafka/pull/18987#issuecomment-3157317076
This PR is being marked as stale since it has not had any activity in 90 days. If you
would like to keep this PR alive, please leave a comment asking for a review. If the PR has
merge conflicts, update it with the latest from the base branch.
<p>
If you are having difficulty finding a reviewer, please reach out on the
[mailing list](https://kafka.apache.org/contact).
<p>
If this PR is no longer valid or desired, please feel free to close it. If no activity
occurs in the next 30 days, it will be automatically closed.
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests [kafka]
Posted by "kevin-wu24 (via GitHub)" <gi...@apache.org>.
kevin-wu24 commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r2006079891
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -1127,14 +1477,83 @@ private MajorityReachedHighWatermark(Cluster cluster) {
@Override
public void verify() {
- cluster.leaderHighWatermark().ifPresent(highWatermark -> {
- long numReachedHighWatermark = cluster.nodes.entrySet().stream()
- .filter(entry -> cluster.voters.containsKey(entry.getKey()))
- .filter(entry -> entry.getValue().log.endOffset().offset() >= highWatermark)
- .count();
- assertTrue(
- numReachedHighWatermark >= cluster.majoritySize(),
- "Insufficient nodes have reached current high watermark");
+ if (cluster.withKip853) {
+ /*
+ * For clusters running in KIP-853 mode, we check that a majority of at least one of the following
+ * voter sets has reached the high watermark:
+ * 1. the leader's voter set at the HWM (i.e. the last committed voter set)
+ * 2. the leader's lastVoterSet() (which may or may not be committed)
+ * Note that 1 and 2 can be the same set, but when they are not, lastVoterSet() is uncommitted,
+ * which follows from the AtMostOneUncommittedVoterSet invariant.
+ *
+ * A more elaborate check is necessary for this invariant because this method can get called after the
+ * leader has updated its lastVoterSet() with a new uncommitted voter set, but before the leader has
+ * updated its high watermark using the new voter set. In this case, we need to check that the majority
+ * of the last committed voter set has reached the current high watermark.
+ * */
+ cluster.leaderWithMaxEpoch().ifPresent(leaderNode -> {
+ leaderNode.client.highWatermark().ifPresent(highWatermark -> {
+ VoterSet voterSet = leaderNode.client.partitionState().lastVoterSet();
+ long numReachedHighWatermark = numReachedHighWatermark(highWatermark, voterSet.voterIds());
+ if (numReachedHighWatermark < cluster.majoritySize(voterSet.size())) {
+ leaderNode.client.partitionState().voterSetAtOffset(highWatermark - 1).ifPresent(otherVoterSet -> {
+ long nodesReachedHighWatermark = numReachedHighWatermark(highWatermark, otherVoterSet.voterIds());
+ assertTrue(
+ nodesReachedHighWatermark >= cluster.majoritySize(otherVoterSet.size()),
+ "Insufficient nodes have reached current high watermark. Expected at least " +
+ cluster.majoritySize(otherVoterSet.size()) + " but got " + nodesReachedHighWatermark);
+ });
+ return;
+ }
+ assertTrue(
+ numReachedHighWatermark >= cluster.majoritySize(voterSet.size()),
+ "Insufficient nodes have reached current high watermark. Expected at least " +
+ cluster.majoritySize(voterSet.size()) + " but got " + numReachedHighWatermark);
+ });
+ });
+ } else {
+ cluster.leaderHighWatermark().ifPresent(highWatermark -> {
+ long numReachedHighWatermark = numReachedHighWatermark(highWatermark, cluster.initialVoters.keySet());
+ assertTrue(
+ numReachedHighWatermark >= cluster.majoritySize(cluster.initialVoters.size()),
+ "Insufficient nodes have reached current high watermark");
+ });
+ }
+ }
+
+ private long numReachedHighWatermark(long highWatermark, Set<Integer> voterIds) {
+ return cluster.persistentStates.entrySet().stream()
+ .filter(entry -> voterIds.contains(entry.getKey()))
+ .filter(entry -> entry.getValue().log.endOffset().offset() >= highWatermark)
+ .count();
+ }
+ }
+
+ private static class AtMostOneUncommittedVotersRecord implements Invariant {
+ final Cluster cluster;
+
+ private AtMostOneUncommittedVotersRecord(Cluster cluster) {
+ this.cluster = cluster;
+ }
+
+ @Override
+ public void verify() {
+ cluster.leaderWithMaxEpoch().ifPresent(leaderNode -> {
+ leaderNode.log.readBatches(leaderNode.highWatermark(), OptionalLong.of(leaderNode.logEndOffset())).forEach(batch -> {
+ boolean seenUncommittedVotersRecord = false;
+ if (batch.isControlBatch) {
+ for (LogEntry entry : batch.entries) {
+ short typeId = ControlRecordType.parseTypeId(entry.record.key());
+ ControlRecordType type = ControlRecordType.fromTypeId(typeId);
+ if (type == ControlRecordType.KRAFT_VOTERS) {
+ if (seenUncommittedVotersRecord) {
+ fail("More than one uncommitted voters record found in the log");
+ }
+ seenUncommittedVotersRecord = true;
+ }
+ }
+ }
+ });
Review Comment:
What do you mean by implementation-specific? Specific to KIP-853?
When we think of an invariant that encapsulates a "consistent log", isn't that really a combination of the other pre-defined invariants together (like monotonic HWM/Epoch)? From what I understand, this is adding an additional criteria for the log to be "consistent" while the RaftNodes are running.
We preform validation at the end of the test with the `ConsistentCommittedData` validation, but this only applies to committed data, and does not run while the RaftNodes are polling.
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests [kafka]
Posted by "kevin-wu24 (via GitHub)" <gi...@apache.org>.
kevin-wu24 commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r2006103261
##########
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java:
##########
@@ -3637,6 +3637,11 @@ QuorumState quorum() {
return quorum;
}
+ // Visible only for test
+ KRaftControlRecordStateMachine partitionState() {
Review Comment:
I think I understand what you're suggesting. Basically, we store the information we care about from the `partitionState` in `MockLog`, and update them accordingly when we append records or advance the HWM.
For my understanding, why is it a concern that the test is using this? I see that `RaftEventSimulationTest` is already using several `KafkaRaftClient` specific methods that aren't defined in the `RaftClient` interface (i.e. `quorum()`, `poll()`, etc.).
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests [kafka]
Posted by "kevin-wu24 (via GitHub)" <gi...@apache.org>.
kevin-wu24 commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r2006278691
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -1127,14 +1331,75 @@ private MajorityReachedHighWatermark(Cluster cluster) {
@Override
public void verify() {
- cluster.leaderHighWatermark().ifPresent(highWatermark -> {
- long numReachedHighWatermark = cluster.nodes.entrySet().stream()
- .filter(entry -> cluster.voters.containsKey(entry.getKey()))
- .filter(entry -> entry.getValue().log.endOffset().offset() >= highWatermark)
- .count();
- assertTrue(
- numReachedHighWatermark >= cluster.majoritySize(),
- "Insufficient nodes have reached current high watermark");
+ if (cluster.withKip853) {
+ /*
+ * For clusters running in KIP-853 mode, we check that a majority of at least one of:
+ * 1. the leader's voter set at the HWM
+ * 2. the leader's lastVoterSet()
+ * has reached the HWM. We need to perform a more elaborate check here because in clusters where
+ * an Add/RemoveVoter request increases/decreases the majority of voters value by 1, the leader
+ * could have used either majority value to update its HWM value. This is because depending on
+ * whether the leader read the most recent VotersRecord prior to updating its HWM value, the number
+ * of nodes (the majority) used to calculate that HWM value is different. This matters for invariant
+ * checking because we perform this verification on every message delivery.
+ * */
+ cluster.leaderWithMaxEpoch().ifPresent(leaderNode -> {
+ leaderNode.client.highWatermark().ifPresent(highWatermark -> {
+ VoterSet voterSet = leaderNode.client.partitionState().lastVoterSet();
+ long numReachedHighWatermark = numReachedHighWatermark(highWatermark, voterSet.voterIds());
+ if (numReachedHighWatermark < cluster.majoritySize(voterSet.size())) {
+ leaderNode.client.partitionState().voterSetAtOffset(highWatermark - 1).ifPresent(otherVoterSet -> {
Review Comment:
I agree that this minimal voter set check is valid, but I think it is too relaxed for a cluster with dynamic voters. Specifically, I have concerns about this statement:
> KRaft is allowed to commit up to the HWM between voterSet1 or voterSet2, with either voter set.
From my understanding of the code, once KRaft becomes aware of voterSet2, which happens at the following point in execution because `partitionState.lastVoterSet()` is set to voterSet2: `appendAsLeader -> partitionState.updateState()`, it should be using voterSet2 to calculate the HWM thereafter, which it does in the following code from `KafkaRaftClient#updateLeaderEndOffsetAndTimestamp`, which is called in `flushLeaderLog`:
```
if (state.updateLocalState(endOffsetMetadata, partitionState.lastVoterSet())) {
onUpdateLeaderHighWatermark(state, currentTimeMs);
}
```
The only point where KafkaRaftClient state is the following:
1. `partitionState.lastVoterSet() = votersSet2`
2. HWM value was calculated using voterSet1
is between `appendAsLeader -> partitionState.updateState()` and `flushLeaderLog` when voterSet2 is written to the log.
When checking the invariant in `verify()`, the RaftNode either has an uncommitted voter set in its log or it doesn't. If it doesn't, using a majority `lastVoterSet()` to verify the HWM value is always correct. If it does, using a majority of `lastVoterSet()` to verify the HWM is correct except in the case I mentioned above.
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #18987:
URL: https://github.com/apache/kafka/pull/18987#issuecomment-3260306877
This PR has been closed since it has not had any activity in 120 days. If you feel like this
was a mistake, or you would like to continue working on it, please feel free to re-open the
PR and ask for a review.
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests pt 1 [kafka]
Posted by "ahuang98 (via GitHub)" <gi...@apache.org>.
ahuang98 commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r1970761578
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -1127,14 +1331,75 @@ private MajorityReachedHighWatermark(Cluster cluster) {
@Override
public void verify() {
- cluster.leaderHighWatermark().ifPresent(highWatermark -> {
- long numReachedHighWatermark = cluster.nodes.entrySet().stream()
- .filter(entry -> cluster.voters.containsKey(entry.getKey()))
- .filter(entry -> entry.getValue().log.endOffset().offset() >= highWatermark)
- .count();
- assertTrue(
- numReachedHighWatermark >= cluster.majoritySize(),
- "Insufficient nodes have reached current high watermark");
+ if (cluster.withKip853) {
+ /*
+ * For clusters running in KIP-853 mode, we check that a majority of at least one of:
+ * 1. the leader's voter set at the HWM
+ * 2. the leader's lastVoterSet()
+ * has reached the HWM. We need to perform a more elaborate check here because in clusters where
+ * an Add/RemoveVoter request increases/decreases the majority of voters value by 1, the leader
+ * could have used either majority value to update its HWM value. This is because depending on
+ * whether the leader read the most recent VotersRecord prior to updating its HWM value, the number
Review Comment:
discussed offline
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests pt 1 [kafka]
Posted by "kevin-wu24 (via GitHub)" <gi...@apache.org>.
kevin-wu24 commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r1980466882
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -1127,14 +1457,83 @@ private MajorityReachedHighWatermark(Cluster cluster) {
@Override
public void verify() {
- cluster.leaderHighWatermark().ifPresent(highWatermark -> {
- long numReachedHighWatermark = cluster.nodes.entrySet().stream()
- .filter(entry -> cluster.voters.containsKey(entry.getKey()))
- .filter(entry -> entry.getValue().log.endOffset().offset() >= highWatermark)
- .count();
- assertTrue(
- numReachedHighWatermark >= cluster.majoritySize(),
- "Insufficient nodes have reached current high watermark");
+ if (cluster.withKip853) {
+ /*
+ * For clusters running in KIP-853 mode, we check that a majority of at least one of the following
+ * voter sets has reached the high watermark:
+ * 1. the leader's voter set at the HWM (i.e. the last committed voter set)
+ * 2. the leader's lastVoterSet() (which may or may not be committed)
+ * Note that 1 and 2 can be the same set, but when they are not, lastVoterSet() is uncommitted,
+ * which follows from the AtMostOneUncommittedVoterSet invariant.
+ *
+ * A more elaborate check is necessary for this invariant because this method can get called after the
+ * leader has updated its lastVoterSet() with a new uncommitted voter set, but before the leader has
+ * updated its high watermark using the new voter set. In this case, we need to check that the majority
+ * of the last committed voter set has reached the current high watermark.
+ * */
+ cluster.leaderWithMaxEpoch().ifPresent(leaderNode -> {
+ leaderNode.client.highWatermark().ifPresent(highWatermark -> {
+ VoterSet voterSet = leaderNode.client.partitionState().lastVoterSet();
+ long numReachedHighWatermark = numReachedHighWatermark(highWatermark, voterSet.voterIds());
+ if (numReachedHighWatermark < cluster.majoritySize(voterSet.size())) {
+ leaderNode.client.partitionState().voterSetAtOffset(highWatermark - 1).ifPresent(otherVoterSet -> {
+ long nodesReachedHighWatermark = numReachedHighWatermark(highWatermark, otherVoterSet.voterIds());
+ assertTrue(
+ nodesReachedHighWatermark >= cluster.majoritySize(otherVoterSet.size()),
+ "Insufficient nodes have reached current high watermark. Expected at least " +
+ cluster.majoritySize(otherVoterSet.size()) + " but got " + nodesReachedHighWatermark);
+ });
+ return;
+ }
+ assertTrue(
+ numReachedHighWatermark >= cluster.majoritySize(voterSet.size()),
+ "Insufficient nodes have reached current high watermark. Expected at least " +
+ cluster.majoritySize(voterSet.size()) + " but got " + numReachedHighWatermark);
+ });
+ });
+ } else {
+ cluster.leaderHighWatermark().ifPresent(highWatermark -> {
+ long numReachedHighWatermark = numReachedHighWatermark(highWatermark, cluster.initialVoters.keySet());
+ assertTrue(
+ numReachedHighWatermark >= cluster.majoritySize(cluster.initialVoters.size()),
+ "Insufficient nodes have reached current high watermark");
+ });
+ }
+ }
+
+ private long numReachedHighWatermark(long highWatermark, Set<Integer> voterIds) {
+ return cluster.persistentStates.entrySet().stream()
+ .filter(entry -> voterIds.contains(entry.getKey()))
+ .filter(entry -> entry.getValue().log.endOffset().offset() >= highWatermark)
+ .count();
+ }
+ }
+
+ private static class AtMostOneUncommittedVotersRecord implements Invariant {
+ final Cluster cluster;
+
+ private AtMostOneUncommittedVotersRecord(Cluster cluster) {
+ this.cluster = cluster;
+ }
+
+ @Override
+ public void verify() {
+ cluster.leaderWithMaxEpoch().ifPresent(leaderNode -> {
+ leaderNode.log.readBatches(leaderNode.highWatermark(), OptionalLong.of(leaderNode.logEndOffset())).forEach(batch -> {
+ Optional<LogEntry> uncommittedVotersEntry = Optional.empty();
Review Comment:
Yeah probably.
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -1127,14 +1457,83 @@ private MajorityReachedHighWatermark(Cluster cluster) {
@Override
public void verify() {
- cluster.leaderHighWatermark().ifPresent(highWatermark -> {
- long numReachedHighWatermark = cluster.nodes.entrySet().stream()
- .filter(entry -> cluster.voters.containsKey(entry.getKey()))
- .filter(entry -> entry.getValue().log.endOffset().offset() >= highWatermark)
- .count();
- assertTrue(
- numReachedHighWatermark >= cluster.majoritySize(),
- "Insufficient nodes have reached current high watermark");
+ if (cluster.withKip853) {
+ /*
+ * For clusters running in KIP-853 mode, we check that a majority of at least one of the following
+ * voter sets has reached the high watermark:
+ * 1. the leader's voter set at the HWM (i.e. the last committed voter set)
+ * 2. the leader's lastVoterSet() (which may or may not be committed)
+ * Note that 1 and 2 can be the same set, but when they are not, lastVoterSet() is uncommitted,
+ * which follows from the AtMostOneUncommittedVoterSet invariant.
+ *
+ * A more elaborate check is necessary for this invariant because this method can get called after the
+ * leader has updated its lastVoterSet() with a new uncommitted voter set, but before the leader has
+ * updated its high watermark using the new voter set. In this case, we need to check that the majority
+ * of the last committed voter set has reached the current high watermark.
+ * */
+ cluster.leaderWithMaxEpoch().ifPresent(leaderNode -> {
+ leaderNode.client.highWatermark().ifPresent(highWatermark -> {
+ VoterSet voterSet = leaderNode.client.partitionState().lastVoterSet();
+ long numReachedHighWatermark = numReachedHighWatermark(highWatermark, voterSet.voterIds());
+ if (numReachedHighWatermark < cluster.majoritySize(voterSet.size())) {
+ leaderNode.client.partitionState().voterSetAtOffset(highWatermark - 1).ifPresent(otherVoterSet -> {
+ long nodesReachedHighWatermark = numReachedHighWatermark(highWatermark, otherVoterSet.voterIds());
+ assertTrue(
+ nodesReachedHighWatermark >= cluster.majoritySize(otherVoterSet.size()),
+ "Insufficient nodes have reached current high watermark. Expected at least " +
+ cluster.majoritySize(otherVoterSet.size()) + " but got " + nodesReachedHighWatermark);
+ });
+ return;
+ }
+ assertTrue(
+ numReachedHighWatermark >= cluster.majoritySize(voterSet.size()),
+ "Insufficient nodes have reached current high watermark. Expected at least " +
+ cluster.majoritySize(voterSet.size()) + " but got " + numReachedHighWatermark);
+ });
+ });
+ } else {
+ cluster.leaderHighWatermark().ifPresent(highWatermark -> {
+ long numReachedHighWatermark = numReachedHighWatermark(highWatermark, cluster.initialVoters.keySet());
+ assertTrue(
+ numReachedHighWatermark >= cluster.majoritySize(cluster.initialVoters.size()),
+ "Insufficient nodes have reached current high watermark");
+ });
+ }
+ }
+
+ private long numReachedHighWatermark(long highWatermark, Set<Integer> voterIds) {
+ return cluster.persistentStates.entrySet().stream()
+ .filter(entry -> voterIds.contains(entry.getKey()))
+ .filter(entry -> entry.getValue().log.endOffset().offset() >= highWatermark)
+ .count();
+ }
+ }
+
+ private static class AtMostOneUncommittedVotersRecord implements Invariant {
+ final Cluster cluster;
+
+ private AtMostOneUncommittedVotersRecord(Cluster cluster) {
+ this.cluster = cluster;
+ }
+
+ @Override
+ public void verify() {
+ cluster.leaderWithMaxEpoch().ifPresent(leaderNode -> {
+ leaderNode.log.readBatches(leaderNode.highWatermark(), OptionalLong.of(leaderNode.logEndOffset())).forEach(batch -> {
+ Optional<LogEntry> uncommittedVotersEntry = Optional.empty();
Review Comment:
Yeah probably.
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests pt 1 [kafka]
Posted by "ahuang98 (via GitHub)" <gi...@apache.org>.
ahuang98 commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r1980497653
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -297,18 +309,83 @@ void canMakeProgressIfMajorityIsReachable(
scheduler.runUntil(() -> cluster.allReachedHighWatermark(2 * restoredLogEndOffset));
}
+ @Property(tries = 100, afterFailure = AfterFailureMode.SAMPLE_ONLY)
+ void canAddVotersIfMajorityIsReachable(
+ @ForAll int seed,
+ @ForAll @IntRange(min = 1, max = 3) int numObservers
+ ) {
+ int numVoters = 5;
+ Random random = new Random(seed);
+ Cluster cluster = new Cluster(numVoters, numObservers, random, true);
+ MessageRouter router = new MessageRouter(cluster);
+ EventScheduler scheduler = schedulerWithDefaultInvariants(cluster);
+ Set<Integer> expectedVoterIds = new HashSet<>(cluster.initialVoters.keySet());
+
+ initializeClusterAndStartAppending(cluster, router, scheduler, 10);
+
+ // Partition the nodes into two sets. Nodes are reachable within each set,
+ // but the two sets cannot communicate with each other. We should be able
+ // to make progress even if an election is needed in the larger set.
+ int firstObserverId = numVoters;
+ router.filter(
+ 0,
+ new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(2, 3, 4, firstObserverId)))
+ );
+ router.filter(
+ 1,
+ new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(2, 3, 4, firstObserverId)))
+ );
+ router.filter(2, new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(0, 1))));
+ router.filter(3, new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(0, 1))));
+ router.filter(4, new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(0, 1))));
+ router.filter(firstObserverId, new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(0, 1))));
+
+ long partitionLogEndOffset = cluster.maxLogEndOffset();
+ scheduler.runUntil(() -> cluster.anyReachedHighWatermark(2 * partitionLogEndOffset));
+
+ long minorityHighWatermark = cluster.maxHighWatermarkReached(Set.of(0, 1));
+ long majorityHighWatermark = cluster.maxHighWatermarkReached(Set.of(2, 3, 4));
+
+ assertTrue(
Review Comment:
got it, I would say if it's easy enough to make this one more exact (optionally the above test too) then it would be worth it
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests pt 1 [kafka]
Posted by "ahuang98 (via GitHub)" <gi...@apache.org>.
ahuang98 commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r1980498577
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -297,18 +309,83 @@ void canMakeProgressIfMajorityIsReachable(
scheduler.runUntil(() -> cluster.allReachedHighWatermark(2 * restoredLogEndOffset));
}
+ @Property(tries = 100, afterFailure = AfterFailureMode.SAMPLE_ONLY)
+ void canAddVotersIfMajorityIsReachable(
+ @ForAll int seed,
+ @ForAll @IntRange(min = 1, max = 3) int numObservers
+ ) {
+ int numVoters = 5;
+ Random random = new Random(seed);
+ Cluster cluster = new Cluster(numVoters, numObservers, random, true);
+ MessageRouter router = new MessageRouter(cluster);
+ EventScheduler scheduler = schedulerWithDefaultInvariants(cluster);
+ Set<Integer> expectedVoterIds = new HashSet<>(cluster.initialVoters.keySet());
+
+ initializeClusterAndStartAppending(cluster, router, scheduler, 10);
+
+ // Partition the nodes into two sets. Nodes are reachable within each set,
+ // but the two sets cannot communicate with each other. We should be able
+ // to make progress even if an election is needed in the larger set.
+ int firstObserverId = numVoters;
+ router.filter(
+ 0,
+ new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(2, 3, 4, firstObserverId)))
+ );
+ router.filter(
+ 1,
+ new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(2, 3, 4, firstObserverId)))
+ );
+ router.filter(2, new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(0, 1))));
+ router.filter(3, new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(0, 1))));
+ router.filter(4, new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(0, 1))));
+ router.filter(firstObserverId, new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(0, 1))));
+
+ long partitionLogEndOffset = cluster.maxLogEndOffset();
+ scheduler.runUntil(() -> cluster.anyReachedHighWatermark(2 * partitionLogEndOffset));
+
+ long minorityHighWatermark = cluster.maxHighWatermarkReached(Set.of(0, 1));
+ long majorityHighWatermark = cluster.maxHighWatermarkReached(Set.of(2, 3, 4));
+
+ assertTrue(
+ majorityHighWatermark > minorityHighWatermark,
+ String.format(
+ "majorityHighWatermark = %s, minorityHighWatermark = %s",
+ majorityHighWatermark,
+ minorityHighWatermark
+ )
+ );
+
+ // Verify we can add a voter, since a majority is still reachable
+ addVoter(cluster, scheduler, firstObserverId, expectedVoterIds);
Review Comment:
I see, either that or maybe rename the param to make it clear it's the original voter set?
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests pt 1 [kafka]
Posted by "kevin-wu24 (via GitHub)" <gi...@apache.org>.
kevin-wu24 commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r1980455784
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -297,18 +309,83 @@ void canMakeProgressIfMajorityIsReachable(
scheduler.runUntil(() -> cluster.allReachedHighWatermark(2 * restoredLogEndOffset));
}
+ @Property(tries = 100, afterFailure = AfterFailureMode.SAMPLE_ONLY)
+ void canAddVotersIfMajorityIsReachable(
+ @ForAll int seed,
+ @ForAll @IntRange(min = 1, max = 3) int numObservers
+ ) {
+ int numVoters = 5;
+ Random random = new Random(seed);
+ Cluster cluster = new Cluster(numVoters, numObservers, random, true);
+ MessageRouter router = new MessageRouter(cluster);
+ EventScheduler scheduler = schedulerWithDefaultInvariants(cluster);
+ Set<Integer> expectedVoterIds = new HashSet<>(cluster.initialVoters.keySet());
+
+ initializeClusterAndStartAppending(cluster, router, scheduler, 10);
+
+ // Partition the nodes into two sets. Nodes are reachable within each set,
+ // but the two sets cannot communicate with each other. We should be able
+ // to make progress even if an election is needed in the larger set.
+ int firstObserverId = numVoters;
+ router.filter(
+ 0,
+ new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(2, 3, 4, firstObserverId)))
+ );
+ router.filter(
+ 1,
+ new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(2, 3, 4, firstObserverId)))
+ );
+ router.filter(2, new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(0, 1))));
+ router.filter(3, new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(0, 1))));
+ router.filter(4, new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(0, 1))));
+ router.filter(firstObserverId, new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(0, 1))));
+
+ long partitionLogEndOffset = cluster.maxLogEndOffset();
+ scheduler.runUntil(() -> cluster.anyReachedHighWatermark(2 * partitionLogEndOffset));
+
+ long minorityHighWatermark = cluster.maxHighWatermarkReached(Set.of(0, 1));
+ long majorityHighWatermark = cluster.maxHighWatermarkReached(Set.of(2, 3, 4));
+
+ assertTrue(
+ majorityHighWatermark > minorityHighWatermark,
+ String.format(
+ "majorityHighWatermark = %s, minorityHighWatermark = %s",
+ majorityHighWatermark,
+ minorityHighWatermark
+ )
+ );
+
+ // Verify we can add a voter, since a majority is still reachable
+ addVoter(cluster, scheduler, firstObserverId, expectedVoterIds);
Review Comment:
It does. Modifications to `expectedVoterIds` are made within the `addVoter/removeVoter` method. The first thing these methods do is add/remove the id we pass in from `expectedVoterIds`. Maybe that line should be outside of the function for clarity?
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests pt 1 [kafka]
Posted by "ahuang98 (via GitHub)" <gi...@apache.org>.
ahuang98 commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r1982287669
##########
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java:
##########
@@ -3637,6 +3637,11 @@ QuorumState quorum() {
return quorum;
}
+ // Visible only for test
+ KRaftControlRecordStateMachine partitionState() {
Review Comment:
hm, I wonder if we could expose voterSetAtOffset and lastVoterSet instead of making the entire state machine public?
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests pt 1 [kafka]
Posted by "kevin-wu24 (via GitHub)" <gi...@apache.org>.
kevin-wu24 commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r1981774312
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -297,18 +309,83 @@ void canMakeProgressIfMajorityIsReachable(
scheduler.runUntil(() -> cluster.allReachedHighWatermark(2 * restoredLogEndOffset));
}
+ @Property(tries = 100, afterFailure = AfterFailureMode.SAMPLE_ONLY)
+ void canAddVotersIfMajorityIsReachable(
+ @ForAll int seed,
+ @ForAll @IntRange(min = 1, max = 3) int numObservers
+ ) {
+ int numVoters = 5;
+ Random random = new Random(seed);
+ Cluster cluster = new Cluster(numVoters, numObservers, random, true);
+ MessageRouter router = new MessageRouter(cluster);
+ EventScheduler scheduler = schedulerWithDefaultInvariants(cluster);
+ Set<Integer> expectedVoterIds = new HashSet<>(cluster.initialVoters.keySet());
+
+ initializeClusterAndStartAppending(cluster, router, scheduler, 10);
+
+ // Partition the nodes into two sets. Nodes are reachable within each set,
+ // but the two sets cannot communicate with each other. We should be able
+ // to make progress even if an election is needed in the larger set.
+ int firstObserverId = numVoters;
+ router.filter(
+ 0,
+ new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(2, 3, 4, firstObserverId)))
+ );
+ router.filter(
+ 1,
+ new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(2, 3, 4, firstObserverId)))
+ );
+ router.filter(2, new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(0, 1))));
+ router.filter(3, new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(0, 1))));
+ router.filter(4, new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(0, 1))));
+ router.filter(firstObserverId, new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(0, 1))));
+
+ long partitionLogEndOffset = cluster.maxLogEndOffset();
+ scheduler.runUntil(() -> cluster.anyReachedHighWatermark(2 * partitionLogEndOffset));
+
+ long minorityHighWatermark = cluster.maxHighWatermarkReached(Set.of(0, 1));
+ long majorityHighWatermark = cluster.maxHighWatermarkReached(Set.of(2, 3, 4));
+
+ assertTrue(
Review Comment:
> seems a bit inexact (i.e. you should know the exact max HW that the smaller partition has because it cannot have changed after the filters were applied?)
This is not exactly true, but it's mainly due to the framework. Basically, the HWM on the minority can still advance a little bit after we partition the network. This is because prior to partitioning, the network router could have delivered some fetch responses that would advance the HWM on the minority node, but that message is just sitting on the node's message queue and hasn't been polled yet to update internal state. When we call:
```
scheduler.runUntil(() -> cluster.anyReachedHighWatermark(2 * partitionLogEndOffset));
```
those minority nodes don't receive any new messages because of the filter, but their `poll` events can update their HWM value a little.
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests pt 1 [kafka]
Posted by "ahuang98 (via GitHub)" <gi...@apache.org>.
ahuang98 commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r1968254346
##########
raft/src/main/java/org/apache/kafka/raft/VoterSet.java:
##########
@@ -343,6 +343,10 @@ public ReplicaKey voterKey() {
return voterKey;
}
+ public int voterId() {
Review Comment:
can this be derived from voterKey().id() instead?
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -487,6 +555,12 @@ private EventScheduler schedulerWithDefaultInvariants(Cluster cluster) {
return scheduler;
}
+ private EventScheduler schedulerWithKip853Invariants(Cluster cluster) {
+ EventScheduler scheduler = schedulerWithDefaultInvariants(cluster);
+ scheduler.addInvariant(new AtMostOneUncommittedVoterSet(cluster));
Review Comment:
this invariant doesn't apply to all the tests? (if there is a reason this invariant can't be part of the default invariant set let's add a comment to explain why)
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -1127,14 +1331,75 @@ private MajorityReachedHighWatermark(Cluster cluster) {
@Override
public void verify() {
- cluster.leaderHighWatermark().ifPresent(highWatermark -> {
- long numReachedHighWatermark = cluster.nodes.entrySet().stream()
- .filter(entry -> cluster.voters.containsKey(entry.getKey()))
- .filter(entry -> entry.getValue().log.endOffset().offset() >= highWatermark)
- .count();
- assertTrue(
- numReachedHighWatermark >= cluster.majoritySize(),
- "Insufficient nodes have reached current high watermark");
+ if (cluster.withKip853) {
+ /*
+ * For clusters running in KIP-853 mode, we check that a majority of at least one of:
+ * 1. the leader's voter set at the HWM
+ * 2. the leader's lastVoterSet()
+ * has reached the HWM. We need to perform a more elaborate check here because in clusters where
+ * an Add/RemoveVoter request increases/decreases the majority of voters value by 1, the leader
+ * could have used either majority value to update its HWM value. This is because depending on
+ * whether the leader read the most recent VotersRecord prior to updating its HWM value, the number
+ * of nodes (the majority) used to calculate that HWM value is different. This matters for invariant
+ * checking because we perform this verification on every message delivery.
+ * */
+ cluster.leaderWithMaxEpoch().ifPresent(leaderNode -> {
+ leaderNode.client.highWatermark().ifPresent(highWatermark -> {
+ VoterSet voterSet = leaderNode.client.partitionState().lastVoterSet();
+ long numReachedHighWatermark = numReachedHighWatermark(highWatermark, voterSet.voterIds());
+ if (numReachedHighWatermark < cluster.majoritySize(voterSet.size())) {
+ leaderNode.client.partitionState().voterSetAtOffset(highWatermark - 1).ifPresent(otherVoterSet -> {
Review Comment:
I'm having a hard time understanding when this would be necessary - we check verify on every "event" (e.g. new record written to leader? what else qualifies) - and we're concerned that we might be checking majority has reached HWM before the HWM reflects the current voter set?
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -1127,14 +1331,75 @@ private MajorityReachedHighWatermark(Cluster cluster) {
@Override
public void verify() {
- cluster.leaderHighWatermark().ifPresent(highWatermark -> {
- long numReachedHighWatermark = cluster.nodes.entrySet().stream()
- .filter(entry -> cluster.voters.containsKey(entry.getKey()))
- .filter(entry -> entry.getValue().log.endOffset().offset() >= highWatermark)
- .count();
- assertTrue(
- numReachedHighWatermark >= cluster.majoritySize(),
- "Insufficient nodes have reached current high watermark");
+ if (cluster.withKip853) {
+ /*
+ * For clusters running in KIP-853 mode, we check that a majority of at least one of:
+ * 1. the leader's voter set at the HWM
+ * 2. the leader's lastVoterSet()
+ * has reached the HWM. We need to perform a more elaborate check here because in clusters where
+ * an Add/RemoveVoter request increases/decreases the majority of voters value by 1, the leader
+ * could have used either majority value to update its HWM value. This is because depending on
+ * whether the leader read the most recent VotersRecord prior to updating its HWM value, the number
Review Comment:
correct me if I'm wrong - wouldn't this always be the case? that the leader read its most recent votersRecord prior to updating HWM?
`appendAsLeader -> updateState -> maybeLoadLog (will catch the latest votersRecord)`
is called before
`flushLeaderLog -> maybeUpdateHighWatermark`
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -1127,14 +1331,75 @@ private MajorityReachedHighWatermark(Cluster cluster) {
@Override
public void verify() {
- cluster.leaderHighWatermark().ifPresent(highWatermark -> {
- long numReachedHighWatermark = cluster.nodes.entrySet().stream()
- .filter(entry -> cluster.voters.containsKey(entry.getKey()))
- .filter(entry -> entry.getValue().log.endOffset().offset() >= highWatermark)
- .count();
- assertTrue(
- numReachedHighWatermark >= cluster.majoritySize(),
- "Insufficient nodes have reached current high watermark");
+ if (cluster.withKip853) {
+ /*
+ * For clusters running in KIP-853 mode, we check that a majority of at least one of:
+ * 1. the leader's voter set at the HWM
Review Comment:
nit: wording is a bit unclear (e.g. reading "check that a majority of at least one of the leader's voter set at the HWM" doesn't make sense to me)
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests pt 1 [kafka]
Posted by "ahuang98 (via GitHub)" <gi...@apache.org>.
ahuang98 commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r1976004288
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -945,9 +987,13 @@ boolean allReachedHighWatermark(long offset) {
.allMatch(node -> node.highWatermark() >= offset);
}
- boolean allHaveLatestVoterSet(VoterSet latestVoterSet) {
- return running.values().stream()
- .allMatch(node -> node.client.partitionState().lastVoterSet().equals(latestVoterSet));
+ boolean leaderHasNextExpectedVoterSet(Set<Integer> voterIds) {
Review Comment:
nit: maybe `leaderHasVoterSet`
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests pt 1 [kafka]
Posted by "ahuang98 (via GitHub)" <gi...@apache.org>.
ahuang98 commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r1976005044
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -580,6 +576,30 @@ void canRecoverFromSingleNodeCommittedDataLoss(
scheduler.runUntil(() -> cluster.allReachedHighWatermark(highWatermarkBeforeRestart + 10));
}
+ @Property(tries = 100, afterFailure = AfterFailureMode.SAMPLE_ONLY)
+ void canSwapPartitionedVoterForObserver(
Review Comment:
`canRemovePartitionedVoter`?
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests pt 1 [kafka]
Posted by "ahuang98 (via GitHub)" <gi...@apache.org>.
ahuang98 commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r1970745986
##########
raft/src/main/java/org/apache/kafka/raft/VoterSet.java:
##########
@@ -343,6 +343,10 @@ public ReplicaKey voterKey() {
return voterKey;
}
+ public int voterId() {
Review Comment:
as in, can we remove this additional method and replace with calls to `voterKey().id()`
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests pt 1 [kafka]
Posted by "ahuang98 (via GitHub)" <gi...@apache.org>.
ahuang98 commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r1975971708
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -117,40 +125,101 @@ public class RaftEventSimulationTest {
void canElectInitialLeader(
@ForAll int seed,
@ForAll @IntRange(min = 1, max = 5) int numVoters,
- @ForAll @IntRange(min = 0, max = 5) int numObservers
+ @ForAll @IntRange(min = 0, max = 5) int numObservers,
+ @ForAll boolean withKip853
) {
Random random = new Random(seed);
- Cluster cluster = new Cluster(numVoters, numObservers, random);
+ Cluster cluster = new Cluster(numVoters, numObservers, random, withKip853);
MessageRouter router = new MessageRouter(cluster);
EventScheduler scheduler = schedulerWithDefaultInvariants(cluster);
- cluster.startAll();
- schedulePolling(scheduler, cluster, 3, 5);
- scheduler.schedule(router::deliverAll, 0, 2, 1);
- scheduler.schedule(new SequentialAppendAction(cluster), 0, 2, 3);
- scheduler.runUntil(cluster::hasConsistentLeader);
- scheduler.runUntil(() -> cluster.allReachedHighWatermark(10));
+ initializeClusterAndStartAppending(cluster, router, scheduler, 2);
+ }
+
+ @Property(tries = 100, afterFailure = AfterFailureMode.SAMPLE_ONLY)
+ void canElectInitialLeaderAndAddVoter(
+ @ForAll int seed,
+ @ForAll @IntRange(min = 1, max = 5) int numVoters,
+ @ForAll @IntRange(min = 1, max = 5) int numObservers
+ ) {
+ Random random = new Random(seed);
+ Cluster cluster = new Cluster(numVoters, numObservers, random, true);
+ MessageRouter router = new MessageRouter(cluster);
+ EventScheduler scheduler = schedulerWithKip853Invariants(cluster);
+
+ initializeClusterAndStartAppending(cluster, router, scheduler, 10);
+ int firstObserverId = numVoters;
+ scheduler.schedule(new AddVoterAction(cluster, cluster.running.get(firstObserverId)), 0, 5, 3);
+ scheduler.runUntil(() -> cluster.leaderWithMaxEpoch().get().client.partitionState().lastVoterSet().size() == numVoters + 1);
+ VoterSet latestVoterSet = cluster.leaderWithMaxEpoch().get().client.partitionState().lastVoterSet();
+ scheduler.runUntil(() -> cluster.allHaveLatestVoterSet(latestVoterSet));
Review Comment:
this is an interesting condition to have - can this just be replaced with allReachedHighWatermark?
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -487,6 +555,12 @@ private EventScheduler schedulerWithDefaultInvariants(Cluster cluster) {
return scheduler;
}
+ private EventScheduler schedulerWithKip853Invariants(Cluster cluster) {
+ EventScheduler scheduler = schedulerWithDefaultInvariants(cluster);
+ scheduler.addInvariant(new AtMostOneUncommittedVoterSet(cluster));
Review Comment:
I understand your points - but I do feel `initialVoterSetFromIds` is strange as it is now since it returns an empty set when withKip853=false. The name of the method doesn't really convey any reason why the behavior should be different with or without kip 853 (e.g. an initial voter set should exist regardless)
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -117,40 +125,101 @@ public class RaftEventSimulationTest {
void canElectInitialLeader(
@ForAll int seed,
@ForAll @IntRange(min = 1, max = 5) int numVoters,
- @ForAll @IntRange(min = 0, max = 5) int numObservers
+ @ForAll @IntRange(min = 0, max = 5) int numObservers,
+ @ForAll boolean withKip853
) {
Random random = new Random(seed);
- Cluster cluster = new Cluster(numVoters, numObservers, random);
+ Cluster cluster = new Cluster(numVoters, numObservers, random, withKip853);
MessageRouter router = new MessageRouter(cluster);
EventScheduler scheduler = schedulerWithDefaultInvariants(cluster);
- cluster.startAll();
- schedulePolling(scheduler, cluster, 3, 5);
- scheduler.schedule(router::deliverAll, 0, 2, 1);
- scheduler.schedule(new SequentialAppendAction(cluster), 0, 2, 3);
- scheduler.runUntil(cluster::hasConsistentLeader);
- scheduler.runUntil(() -> cluster.allReachedHighWatermark(10));
+ initializeClusterAndStartAppending(cluster, router, scheduler, 2);
+ }
+
+ @Property(tries = 100, afterFailure = AfterFailureMode.SAMPLE_ONLY)
+ void canElectInitialLeaderAndAddVoter(
+ @ForAll int seed,
+ @ForAll @IntRange(min = 1, max = 5) int numVoters,
+ @ForAll @IntRange(min = 1, max = 5) int numObservers
+ ) {
+ Random random = new Random(seed);
+ Cluster cluster = new Cluster(numVoters, numObservers, random, true);
+ MessageRouter router = new MessageRouter(cluster);
+ EventScheduler scheduler = schedulerWithKip853Invariants(cluster);
+
+ initializeClusterAndStartAppending(cluster, router, scheduler, 10);
+ int firstObserverId = numVoters;
+ scheduler.schedule(new AddVoterAction(cluster, cluster.running.get(firstObserverId)), 0, 5, 3);
+ scheduler.runUntil(() -> cluster.leaderWithMaxEpoch().get().client.partitionState().lastVoterSet().size() == numVoters + 1);
+ VoterSet latestVoterSet = cluster.leaderWithMaxEpoch().get().client.partitionState().lastVoterSet();
+ scheduler.runUntil(() -> cluster.allHaveLatestVoterSet(latestVoterSet));
+ }
+
+ @Property(tries = 100, afterFailure = AfterFailureMode.SAMPLE_ONLY)
+ void canAddAndRemoveVoters(
Review Comment:
so the first test checks if we can add a single voter - this one checks if we can add multiple and remove all but the first voter
would it be worth having a test which confirms we can remove _and then_ add back the voter? and/or a test for removing specifically the leader?
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -487,6 +555,12 @@ private EventScheduler schedulerWithDefaultInvariants(Cluster cluster) {
return scheduler;
}
+ private EventScheduler schedulerWithKip853Invariants(Cluster cluster) {
+ EventScheduler scheduler = schedulerWithDefaultInvariants(cluster);
+ scheduler.addInvariant(new AtMostOneUncommittedVoterSet(cluster));
Review Comment:
even if `AtMostOneUncommittedVoterSet` is a trivial check for tests which are not adding or removing voters, conceptually it should hold true - it might be worth making it work for all tests if it is not too difficult
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -117,40 +125,101 @@ public class RaftEventSimulationTest {
void canElectInitialLeader(
@ForAll int seed,
@ForAll @IntRange(min = 1, max = 5) int numVoters,
- @ForAll @IntRange(min = 0, max = 5) int numObservers
+ @ForAll @IntRange(min = 0, max = 5) int numObservers,
+ @ForAll boolean withKip853
) {
Random random = new Random(seed);
- Cluster cluster = new Cluster(numVoters, numObservers, random);
+ Cluster cluster = new Cluster(numVoters, numObservers, random, withKip853);
MessageRouter router = new MessageRouter(cluster);
EventScheduler scheduler = schedulerWithDefaultInvariants(cluster);
- cluster.startAll();
- schedulePolling(scheduler, cluster, 3, 5);
- scheduler.schedule(router::deliverAll, 0, 2, 1);
- scheduler.schedule(new SequentialAppendAction(cluster), 0, 2, 3);
- scheduler.runUntil(cluster::hasConsistentLeader);
- scheduler.runUntil(() -> cluster.allReachedHighWatermark(10));
+ initializeClusterAndStartAppending(cluster, router, scheduler, 2);
+ }
+
+ @Property(tries = 100, afterFailure = AfterFailureMode.SAMPLE_ONLY)
+ void canElectInitialLeaderAndAddVoter(
+ @ForAll int seed,
+ @ForAll @IntRange(min = 1, max = 5) int numVoters,
+ @ForAll @IntRange(min = 1, max = 5) int numObservers
+ ) {
+ Random random = new Random(seed);
+ Cluster cluster = new Cluster(numVoters, numObservers, random, true);
+ MessageRouter router = new MessageRouter(cluster);
+ EventScheduler scheduler = schedulerWithKip853Invariants(cluster);
+
+ initializeClusterAndStartAppending(cluster, router, scheduler, 10);
+ int firstObserverId = numVoters;
+ scheduler.schedule(new AddVoterAction(cluster, cluster.running.get(firstObserverId)), 0, 5, 3);
+ scheduler.runUntil(() -> cluster.leaderWithMaxEpoch().get().client.partitionState().lastVoterSet().size() == numVoters + 1);
+ VoterSet latestVoterSet = cluster.leaderWithMaxEpoch().get().client.partitionState().lastVoterSet();
+ scheduler.runUntil(() -> cluster.allHaveLatestVoterSet(latestVoterSet));
+ }
+
+ @Property(tries = 100, afterFailure = AfterFailureMode.SAMPLE_ONLY)
+ void canAddAndRemoveVoters(
+ @ForAll int seed,
+ @ForAll @IntRange(min = 1, max = 5) int numVoters,
+ @ForAll @IntRange(min = 1, max = 10) int numObservers
+ ) {
+ Random random = new Random(seed);
+ Cluster cluster = new Cluster(numVoters, numObservers, random, true);
+ MessageRouter router = new MessageRouter(cluster);
+ EventScheduler scheduler = schedulerWithKip853Invariants(cluster);
+
+ initializeClusterAndStartAppending(cluster, router, scheduler, 10);
+
+ // Schedule all the observers to be added as voters
+ for (int voterIdToAdd = numVoters; voterIdToAdd < numVoters + numObservers; voterIdToAdd++) {
+ int nextVoterSetSize = voterIdToAdd + 1;
+ scheduler.schedule(new AddVoterAction(cluster, cluster.running.get(voterIdToAdd)), 0, 5, 3);
+ scheduler.runUntil(() -> cluster.leaderWithMaxEpoch().get().client.partitionState().lastVoterSet().size() == nextVoterSetSize);
+ }
+ int voterSetSize = numVoters + numObservers;
+
+ // Schedule all voters besides the first to be removed
+ for (int voterIdToRemove = numVoters + numObservers - 1; voterIdToRemove > 0; voterIdToRemove--) {
+ int nextVoterSetSize = voterSetSize - 1;
+ scheduler.schedule(new RemoveVoterAction(cluster, cluster.running.get(voterIdToRemove)), 0, 5, 3);
+ if (cluster.leaderWithMaxEpoch().isEmpty()) {
+ scheduler.runUntil(cluster::hasConsistentLeader);
+ }
+ scheduler.runUntil(() -> cluster.leaderWithMaxEpoch().get().client.partitionState().lastVoterSet().size() == nextVoterSetSize);
+ }
+ }
+
+ @Property(tries = 100, afterFailure = AfterFailureMode.SAMPLE_ONLY)
+ void canElectInitialLeaderAndRemoveVoter(
Review Comment:
what does this test that isn't tested by canAddAndRemoveVoters
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -645,39 +837,56 @@ private static class Cluster {
final AtomicInteger correlationIdCounter = new AtomicInteger();
final MockTime time = new MockTime();
final String clusterId = Uuid.randomUuid().toString();
- final Map<Integer, Node> voters = new HashMap<>();
- final Map<Integer, PersistentState> nodes = new HashMap<>();
+ final Map<Integer, Node> initialVoters = new HashMap<>();
+ final Map<Integer, PersistentState> persistentStates = new HashMap<>();
final Map<Integer, RaftNode> running = new HashMap<>();
+ final boolean withKip853;
- private Cluster(int numVoters, int numObservers, Random random) {
+ private Cluster(int numVoters, int numObservers, Random random, boolean withKip853) {
this.random = random;
+ this.withKip853 = withKip853;
for (int nodeId = 0; nodeId < numVoters; nodeId++) {
- voters.put(nodeId, nodeFromId(nodeId));
- nodes.put(nodeId, new PersistentState(nodeId));
+ initialVoters.put(nodeId, nodeFromId(nodeId));
+ persistentStates.put(nodeId, new PersistentState(nodeId));
}
for (int nodeIdDelta = 0; nodeIdDelta < numObservers; nodeIdDelta++) {
int nodeId = numVoters + nodeIdDelta;
- nodes.put(nodeId, new PersistentState(nodeId));
+ persistentStates.put(nodeId, new PersistentState(nodeId));
}
}
Set<InetSocketAddress> endpointsFromIds(Set<Integer> nodeIds) {
- return voters
+ return running
.values()
.stream()
- .filter(node -> nodeIds.contains(node.id()))
+ .filter(node -> nodeIds.contains(node.nodeId))
.map(Cluster::nodeAddress)
.collect(Collectors.toSet());
}
+ Optional<VoterSet> initialVoterSetFromIds() {
Review Comment:
see previous comment for this method - feels weird this would rely on withKip853. can we just remove the withKip853 conditional and always return a voterset based on the initialVoters?
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests pt 1 [kafka]
Posted by "kevin-wu24 (via GitHub)" <gi...@apache.org>.
kevin-wu24 commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r1971722138
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -1127,14 +1331,75 @@ private MajorityReachedHighWatermark(Cluster cluster) {
@Override
public void verify() {
- cluster.leaderHighWatermark().ifPresent(highWatermark -> {
- long numReachedHighWatermark = cluster.nodes.entrySet().stream()
- .filter(entry -> cluster.voters.containsKey(entry.getKey()))
- .filter(entry -> entry.getValue().log.endOffset().offset() >= highWatermark)
- .count();
- assertTrue(
- numReachedHighWatermark >= cluster.majoritySize(),
- "Insufficient nodes have reached current high watermark");
+ if (cluster.withKip853) {
+ /*
+ * For clusters running in KIP-853 mode, we check that a majority of at least one of:
+ * 1. the leader's voter set at the HWM
Review Comment:
Sorry, upon further looking into this, I think the issue is just slightly different that what I've described thus far.
Essentially, when we fail the invariant check when only using `lastVoterSet()`, what's going on is that the caller of `verify()` is a different thread (lets say event scheduler thread) than the KRaft leader's raft-io thread which is continuously doing all of this internal state updating. The event scheduler thread is looking at a bunch of the leader's internal state (e.g. partitionState and highWatermark), which can definitely be in the following state: `partitionState` has been updated with a new voter set, but a new `highWatermark` value has not yet been calculated with this voter set yet, which could cause this invariant check to fail when only looking at `lastVoterSet()`. The leader may not be in this state for very long, since it's in this state in between the calls to `appendAsLeader -> updateState -> maybeLoadLog` and `flushLeaderLog -> maybeUpdateHighWatermark`, but it nevertheless looks like a valid state to me.
Invariants as a concept are predicates that should be true for the system at all times, no matter when you check them, so I think performing invariant verification how it's currently implemented is fine, as we could be checking the internal states of our raft nodes at any point in their execution. However, it just means we have to consider these "intermediary" states when performing verification.
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests [kafka]
Posted by "kevin-wu24 (via GitHub)" <gi...@apache.org>.
kevin-wu24 commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r2006278691
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -1127,14 +1331,75 @@ private MajorityReachedHighWatermark(Cluster cluster) {
@Override
public void verify() {
- cluster.leaderHighWatermark().ifPresent(highWatermark -> {
- long numReachedHighWatermark = cluster.nodes.entrySet().stream()
- .filter(entry -> cluster.voters.containsKey(entry.getKey()))
- .filter(entry -> entry.getValue().log.endOffset().offset() >= highWatermark)
- .count();
- assertTrue(
- numReachedHighWatermark >= cluster.majoritySize(),
- "Insufficient nodes have reached current high watermark");
+ if (cluster.withKip853) {
+ /*
+ * For clusters running in KIP-853 mode, we check that a majority of at least one of:
+ * 1. the leader's voter set at the HWM
+ * 2. the leader's lastVoterSet()
+ * has reached the HWM. We need to perform a more elaborate check here because in clusters where
+ * an Add/RemoveVoter request increases/decreases the majority of voters value by 1, the leader
+ * could have used either majority value to update its HWM value. This is because depending on
+ * whether the leader read the most recent VotersRecord prior to updating its HWM value, the number
+ * of nodes (the majority) used to calculate that HWM value is different. This matters for invariant
+ * checking because we perform this verification on every message delivery.
+ * */
+ cluster.leaderWithMaxEpoch().ifPresent(leaderNode -> {
+ leaderNode.client.highWatermark().ifPresent(highWatermark -> {
+ VoterSet voterSet = leaderNode.client.partitionState().lastVoterSet();
+ long numReachedHighWatermark = numReachedHighWatermark(highWatermark, voterSet.voterIds());
+ if (numReachedHighWatermark < cluster.majoritySize(voterSet.size())) {
+ leaderNode.client.partitionState().voterSetAtOffset(highWatermark - 1).ifPresent(otherVoterSet -> {
Review Comment:
I agree that this minimal voter set check is valid, but I think it is too relaxed for a cluster with dynamic voters. Specifically, I have concerns about this statement, but maybe I'm misunderstanding what is going on:
> KRaft is allowed to commit up to the HWM between voterSet1 or voterSet2, with either voter set.
From my understanding of the code, once KRaft becomes aware of voterSet2, which happens at the following point in execution because `partitionState.lastVoterSet()` is set to voterSet2: `appendAsLeader -> partitionState.updateState()`, it should be using voterSet2 to calculate the HWM thereafter when it handles `FetchRequests` in subsequent events.
For example, in the case of adding a voter, voterSet1 is the minimal voter set with majority size X, and voterSet2 is the uncommitted voter set with majority size X + 1. This means when verifying the HWM value, we would be checking that X nodes have replicated up to the HWM offset, not X + 1 nodes, which should only happen if the HWM value was actually calculated with voterSet1.
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests [kafka]
Posted by "kevin-wu24 (via GitHub)" <gi...@apache.org>.
kevin-wu24 commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r2007947443
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -1127,14 +1477,83 @@ private MajorityReachedHighWatermark(Cluster cluster) {
@Override
public void verify() {
- cluster.leaderHighWatermark().ifPresent(highWatermark -> {
- long numReachedHighWatermark = cluster.nodes.entrySet().stream()
- .filter(entry -> cluster.voters.containsKey(entry.getKey()))
- .filter(entry -> entry.getValue().log.endOffset().offset() >= highWatermark)
- .count();
- assertTrue(
- numReachedHighWatermark >= cluster.majoritySize(),
- "Insufficient nodes have reached current high watermark");
+ if (cluster.withKip853) {
+ /*
+ * For clusters running in KIP-853 mode, we check that a majority of at least one of the following
+ * voter sets has reached the high watermark:
+ * 1. the leader's voter set at the HWM (i.e. the last committed voter set)
+ * 2. the leader's lastVoterSet() (which may or may not be committed)
+ * Note that 1 and 2 can be the same set, but when they are not, lastVoterSet() is uncommitted,
+ * which follows from the AtMostOneUncommittedVoterSet invariant.
+ *
+ * A more elaborate check is necessary for this invariant because this method can get called after the
+ * leader has updated its lastVoterSet() with a new uncommitted voter set, but before the leader has
+ * updated its high watermark using the new voter set. In this case, we need to check that the majority
+ * of the last committed voter set has reached the current high watermark.
+ * */
+ cluster.leaderWithMaxEpoch().ifPresent(leaderNode -> {
+ leaderNode.client.highWatermark().ifPresent(highWatermark -> {
+ VoterSet voterSet = leaderNode.client.partitionState().lastVoterSet();
+ long numReachedHighWatermark = numReachedHighWatermark(highWatermark, voterSet.voterIds());
+ if (numReachedHighWatermark < cluster.majoritySize(voterSet.size())) {
+ leaderNode.client.partitionState().voterSetAtOffset(highWatermark - 1).ifPresent(otherVoterSet -> {
+ long nodesReachedHighWatermark = numReachedHighWatermark(highWatermark, otherVoterSet.voterIds());
+ assertTrue(
+ nodesReachedHighWatermark >= cluster.majoritySize(otherVoterSet.size()),
+ "Insufficient nodes have reached current high watermark. Expected at least " +
+ cluster.majoritySize(otherVoterSet.size()) + " but got " + nodesReachedHighWatermark);
+ });
+ return;
+ }
+ assertTrue(
+ numReachedHighWatermark >= cluster.majoritySize(voterSet.size()),
+ "Insufficient nodes have reached current high watermark. Expected at least " +
+ cluster.majoritySize(voterSet.size()) + " but got " + numReachedHighWatermark);
+ });
+ });
+ } else {
+ cluster.leaderHighWatermark().ifPresent(highWatermark -> {
+ long numReachedHighWatermark = numReachedHighWatermark(highWatermark, cluster.initialVoters.keySet());
+ assertTrue(
+ numReachedHighWatermark >= cluster.majoritySize(cluster.initialVoters.size()),
+ "Insufficient nodes have reached current high watermark");
+ });
+ }
+ }
+
+ private long numReachedHighWatermark(long highWatermark, Set<Integer> voterIds) {
+ return cluster.persistentStates.entrySet().stream()
+ .filter(entry -> voterIds.contains(entry.getKey()))
+ .filter(entry -> entry.getValue().log.endOffset().offset() >= highWatermark)
+ .count();
+ }
+ }
+
+ private static class AtMostOneUncommittedVotersRecord implements Invariant {
+ final Cluster cluster;
+
+ private AtMostOneUncommittedVotersRecord(Cluster cluster) {
+ this.cluster = cluster;
+ }
+
+ @Override
+ public void verify() {
+ cluster.leaderWithMaxEpoch().ifPresent(leaderNode -> {
+ leaderNode.log.readBatches(leaderNode.highWatermark(), OptionalLong.of(leaderNode.logEndOffset())).forEach(batch -> {
+ boolean seenUncommittedVotersRecord = false;
+ if (batch.isControlBatch) {
+ for (LogEntry entry : batch.entries) {
+ short typeId = ControlRecordType.parseTypeId(entry.record.key());
+ ControlRecordType type = ControlRecordType.fromTypeId(typeId);
+ if (type == ControlRecordType.KRAFT_VOTERS) {
+ if (seenUncommittedVotersRecord) {
+ fail("More than one uncommitted voters record found in the log");
+ }
+ seenUncommittedVotersRecord = true;
+ }
+ }
+ }
+ });
Review Comment:
Discussed offline. Will remove this invariant because it exposes something previously internal to the KRaft implementation of dynamic quorum reconfiguration.
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests pt 1 [kafka]
Posted by "ahuang98 (via GitHub)" <gi...@apache.org>.
ahuang98 commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r1980430607
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -297,18 +309,83 @@ void canMakeProgressIfMajorityIsReachable(
scheduler.runUntil(() -> cluster.allReachedHighWatermark(2 * restoredLogEndOffset));
}
+ @Property(tries = 100, afterFailure = AfterFailureMode.SAMPLE_ONLY)
+ void canAddVotersIfMajorityIsReachable(
+ @ForAll int seed,
+ @ForAll @IntRange(min = 1, max = 3) int numObservers
+ ) {
+ int numVoters = 5;
+ Random random = new Random(seed);
+ Cluster cluster = new Cluster(numVoters, numObservers, random, true);
+ MessageRouter router = new MessageRouter(cluster);
+ EventScheduler scheduler = schedulerWithDefaultInvariants(cluster);
+ Set<Integer> expectedVoterIds = new HashSet<>(cluster.initialVoters.keySet());
+
+ initializeClusterAndStartAppending(cluster, router, scheduler, 10);
+
+ // Partition the nodes into two sets. Nodes are reachable within each set,
+ // but the two sets cannot communicate with each other. We should be able
+ // to make progress even if an election is needed in the larger set.
+ int firstObserverId = numVoters;
+ router.filter(
+ 0,
+ new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(2, 3, 4, firstObserverId)))
+ );
+ router.filter(
+ 1,
+ new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(2, 3, 4, firstObserverId)))
+ );
+ router.filter(2, new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(0, 1))));
+ router.filter(3, new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(0, 1))));
+ router.filter(4, new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(0, 1))));
+ router.filter(firstObserverId, new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(0, 1))));
+
+ long partitionLogEndOffset = cluster.maxLogEndOffset();
+ scheduler.runUntil(() -> cluster.anyReachedHighWatermark(2 * partitionLogEndOffset));
+
+ long minorityHighWatermark = cluster.maxHighWatermarkReached(Set.of(0, 1));
+ long majorityHighWatermark = cluster.maxHighWatermarkReached(Set.of(2, 3, 4));
+
+ assertTrue(
+ majorityHighWatermark > minorityHighWatermark,
+ String.format(
+ "majorityHighWatermark = %s, minorityHighWatermark = %s",
+ majorityHighWatermark,
+ minorityHighWatermark
+ )
+ );
+
+ // Verify we can add a voter, since a majority is still reachable
+ addVoter(cluster, scheduler, firstObserverId, expectedVoterIds);
Review Comment:
should expectedVoterIds include the observer you're trying to add?
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -117,40 +126,58 @@ public class RaftEventSimulationTest {
void canElectInitialLeader(
@ForAll int seed,
@ForAll @IntRange(min = 1, max = 5) int numVoters,
- @ForAll @IntRange(min = 0, max = 5) int numObservers
+ @ForAll @IntRange(min = 0, max = 5) int numObservers,
+ @ForAll boolean withKip853
) {
Random random = new Random(seed);
- Cluster cluster = new Cluster(numVoters, numObservers, random);
+ Cluster cluster = new Cluster(numVoters, numObservers, random, withKip853);
MessageRouter router = new MessageRouter(cluster);
EventScheduler scheduler = schedulerWithDefaultInvariants(cluster);
- cluster.startAll();
- schedulePolling(scheduler, cluster, 3, 5);
- scheduler.schedule(router::deliverAll, 0, 2, 1);
- scheduler.schedule(new SequentialAppendAction(cluster), 0, 2, 3);
- scheduler.runUntil(cluster::hasConsistentLeader);
- scheduler.runUntil(() -> cluster.allReachedHighWatermark(10));
+ initializeClusterAndStartAppending(cluster, router, scheduler, 2);
+ }
+
+ @Property(tries = 100, afterFailure = AfterFailureMode.SAMPLE_ONLY)
+ void canAddAndRemoveVoters(
+ @ForAll int seed
+ ) {
+ Random random = new Random(seed);
+ int numVoters = 3;
+ int numObservers = 2;
+ Cluster cluster = new Cluster(numVoters, numObservers, random, true);
+ MessageRouter router = new MessageRouter(cluster);
+ EventScheduler scheduler = schedulerWithDefaultInvariants(cluster);
+ Set<Integer> expectedVoterIds = new HashSet<>(cluster.initialVoters.keySet());
+
+ initializeClusterAndStartAppending(cluster, router, scheduler, 10);
+
+ // Add all observers to voter set one by one
+ for (int voterIdToAdd = numVoters; voterIdToAdd < numVoters + numObservers; voterIdToAdd++) {
+ addVoter(cluster, scheduler, voterIdToAdd, expectedVoterIds);
+ }
+ runUntilNewHighWatermark(cluster, scheduler);
Review Comment:
what is the purpose of this check? highWatermark + 10 seems a bit arbitrary
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -297,18 +309,83 @@ void canMakeProgressIfMajorityIsReachable(
scheduler.runUntil(() -> cluster.allReachedHighWatermark(2 * restoredLogEndOffset));
}
+ @Property(tries = 100, afterFailure = AfterFailureMode.SAMPLE_ONLY)
+ void canAddVotersIfMajorityIsReachable(
+ @ForAll int seed,
+ @ForAll @IntRange(min = 1, max = 3) int numObservers
+ ) {
+ int numVoters = 5;
+ Random random = new Random(seed);
+ Cluster cluster = new Cluster(numVoters, numObservers, random, true);
+ MessageRouter router = new MessageRouter(cluster);
+ EventScheduler scheduler = schedulerWithDefaultInvariants(cluster);
+ Set<Integer> expectedVoterIds = new HashSet<>(cluster.initialVoters.keySet());
+
+ initializeClusterAndStartAppending(cluster, router, scheduler, 10);
+
+ // Partition the nodes into two sets. Nodes are reachable within each set,
+ // but the two sets cannot communicate with each other. We should be able
+ // to make progress even if an election is needed in the larger set.
+ int firstObserverId = numVoters;
+ router.filter(
+ 0,
+ new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(2, 3, 4, firstObserverId)))
+ );
+ router.filter(
+ 1,
+ new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(2, 3, 4, firstObserverId)))
+ );
+ router.filter(2, new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(0, 1))));
+ router.filter(3, new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(0, 1))));
+ router.filter(4, new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(0, 1))));
+ router.filter(firstObserverId, new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(0, 1))));
+
+ long partitionLogEndOffset = cluster.maxLogEndOffset();
+ scheduler.runUntil(() -> cluster.anyReachedHighWatermark(2 * partitionLogEndOffset));
+
+ long minorityHighWatermark = cluster.maxHighWatermarkReached(Set.of(0, 1));
+ long majorityHighWatermark = cluster.maxHighWatermarkReached(Set.of(2, 3, 4));
+
+ assertTrue(
Review Comment:
what's the point of the assertion? seems a bit inexact (i.e. you should know the exact max HW that the smaller partition has because it cannot have changed after the filters were applied?)
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -602,6 +858,14 @@ void schedule(Runnable action, int delayMs, int periodMs, int jitterMs) {
queue.offer(event);
}
+ void scheduleUntil(Runnable action, Supplier<Boolean> exitCondition, int delayMs, int periodMs, int jitterMs) {
+ long initialDeadlineMs = time.milliseconds() + delayMs;
+ int eventId = eventIdGenerator.incrementAndGet();
+ ScheduledUntilConditionMetEvent event = new ScheduledUntilConditionMetEvent(action, eventId, random, initialDeadlineMs, periodMs, jitterMs, exitCondition);
+ queue.offer(event);
+
Review Comment:
nit: extra new line
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -475,6 +537,104 @@ void canRecoverFromSingleNodeCommittedDataLoss(
scheduler.runUntil(() -> cluster.allReachedHighWatermark(highWatermarkBeforeRestart + 10));
}
+ @Property(tries = 100, afterFailure = AfterFailureMode.SAMPLE_ONLY)
+ void canAddObserverAndRemovePartitionedVoter(
+ @ForAll int seed,
+ @ForAll @IntRange(min = 3, max = 5) int numVoters
+ ) {
+ Random random = new Random(seed);
+ Cluster cluster = new Cluster(numVoters, 1, random, true);
+ MessageRouter router = new MessageRouter(cluster);
+ EventScheduler scheduler = schedulerWithDefaultInvariants(cluster);
+ Set<Integer> expectedVoterIds = new HashSet<>(cluster.initialVoters.keySet());
+
+ initializeClusterAndStartAppending(cluster, router, scheduler, 10);
+
+ // Partition a random voter, and add an observer to replace it
+ RaftNode voterToRemove = cluster.running.get(random.nextInt(numVoters));
+ router.filter(voterToRemove.id(), new DropAllTraffic());
+ addVoter(cluster, scheduler, numVoters, expectedVoterIds);
Review Comment:
same question here, does expectedVoterIds need to include the observer you are adding?
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -297,18 +309,83 @@ void canMakeProgressIfMajorityIsReachable(
scheduler.runUntil(() -> cluster.allReachedHighWatermark(2 * restoredLogEndOffset));
}
+ @Property(tries = 100, afterFailure = AfterFailureMode.SAMPLE_ONLY)
+ void canAddVotersIfMajorityIsReachable(
+ @ForAll int seed,
+ @ForAll @IntRange(min = 1, max = 3) int numObservers
+ ) {
+ int numVoters = 5;
+ Random random = new Random(seed);
+ Cluster cluster = new Cluster(numVoters, numObservers, random, true);
+ MessageRouter router = new MessageRouter(cluster);
+ EventScheduler scheduler = schedulerWithDefaultInvariants(cluster);
+ Set<Integer> expectedVoterIds = new HashSet<>(cluster.initialVoters.keySet());
+
+ initializeClusterAndStartAppending(cluster, router, scheduler, 10);
+
+ // Partition the nodes into two sets. Nodes are reachable within each set,
+ // but the two sets cannot communicate with each other. We should be able
+ // to make progress even if an election is needed in the larger set.
+ int firstObserverId = numVoters;
+ router.filter(
+ 0,
+ new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(2, 3, 4, firstObserverId)))
+ );
+ router.filter(
+ 1,
+ new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(2, 3, 4, firstObserverId)))
+ );
+ router.filter(2, new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(0, 1))));
+ router.filter(3, new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(0, 1))));
+ router.filter(4, new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(0, 1))));
+ router.filter(firstObserverId, new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(0, 1))));
+
+ long partitionLogEndOffset = cluster.maxLogEndOffset();
+ scheduler.runUntil(() -> cluster.anyReachedHighWatermark(2 * partitionLogEndOffset));
+
+ long minorityHighWatermark = cluster.maxHighWatermarkReached(Set.of(0, 1));
+ long majorityHighWatermark = cluster.maxHighWatermarkReached(Set.of(2, 3, 4));
+
+ assertTrue(
+ majorityHighWatermark > minorityHighWatermark,
+ String.format(
+ "majorityHighWatermark = %s, minorityHighWatermark = %s",
+ majorityHighWatermark,
+ minorityHighWatermark
+ )
+ );
+
+ // Verify we can add a voter, since a majority is still reachable
+ addVoter(cluster, scheduler, firstObserverId, expectedVoterIds);
+
+ // Now restore the partition and verify everyone catches up
+ router.filter(0, new PermitAllTraffic());
+ router.filter(1, new PermitAllTraffic());
+ router.filter(2, new PermitAllTraffic());
+ router.filter(3, new PermitAllTraffic());
+ router.filter(4, new PermitAllTraffic());
+ router.filter(firstObserverId, new PermitAllTraffic());
+
+ long restoredLogEndOffset = cluster.maxLogEndOffset();
+ scheduler.runUntil(() -> cluster.allReachedHighWatermark(2 * restoredLogEndOffset));
+ }
+
@Property(tries = 100, afterFailure = AfterFailureMode.SAMPLE_ONLY)
void leadershipAssignedOnlyOnceWithNetworkPartitionIfThereExistsMajority(
@ForAll int seed,
- @ForAll @IntRange(min = 0, max = 3) int numObservers
+ @ForAll @IntRange(min = 0, max = 3) int numObservers,
+ @ForAll boolean withKip853
) {
int numVoters = 5;
Random random = new Random(seed);
- Cluster cluster = new Cluster(numVoters, numObservers, random);
+ Cluster cluster = new Cluster(numVoters, numObservers, random, withKip853);
MessageRouter router = new MessageRouter(cluster);
EventScheduler scheduler = schedulerWithDefaultInvariants(cluster);
scheduler.addInvariant(new StableLeadership(cluster));
+ // Start cluster
+ cluster.startAll();
Review Comment:
you change the meaning of the test if you move this (this tests the scenario where network partition exists before cluster starts, the following test addresses a scenario where network partition starts after cluster is up for a while)
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -475,6 +537,104 @@ void canRecoverFromSingleNodeCommittedDataLoss(
scheduler.runUntil(() -> cluster.allReachedHighWatermark(highWatermarkBeforeRestart + 10));
}
+ @Property(tries = 100, afterFailure = AfterFailureMode.SAMPLE_ONLY)
+ void canAddObserverAndRemovePartitionedVoter(
+ @ForAll int seed,
+ @ForAll @IntRange(min = 3, max = 5) int numVoters
+ ) {
+ Random random = new Random(seed);
+ Cluster cluster = new Cluster(numVoters, 1, random, true);
+ MessageRouter router = new MessageRouter(cluster);
+ EventScheduler scheduler = schedulerWithDefaultInvariants(cluster);
+ Set<Integer> expectedVoterIds = new HashSet<>(cluster.initialVoters.keySet());
+
+ initializeClusterAndStartAppending(cluster, router, scheduler, 10);
+
+ // Partition a random voter, and add an observer to replace it
Review Comment:
```
// Partition a random voter. Add first observer as new voter
```
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -1127,14 +1457,83 @@ private MajorityReachedHighWatermark(Cluster cluster) {
@Override
public void verify() {
- cluster.leaderHighWatermark().ifPresent(highWatermark -> {
- long numReachedHighWatermark = cluster.nodes.entrySet().stream()
- .filter(entry -> cluster.voters.containsKey(entry.getKey()))
- .filter(entry -> entry.getValue().log.endOffset().offset() >= highWatermark)
- .count();
- assertTrue(
- numReachedHighWatermark >= cluster.majoritySize(),
- "Insufficient nodes have reached current high watermark");
+ if (cluster.withKip853) {
+ /*
+ * For clusters running in KIP-853 mode, we check that a majority of at least one of the following
+ * voter sets has reached the high watermark:
+ * 1. the leader's voter set at the HWM (i.e. the last committed voter set)
+ * 2. the leader's lastVoterSet() (which may or may not be committed)
+ * Note that 1 and 2 can be the same set, but when they are not, lastVoterSet() is uncommitted,
+ * which follows from the AtMostOneUncommittedVoterSet invariant.
+ *
+ * A more elaborate check is necessary for this invariant because this method can get called after the
+ * leader has updated its lastVoterSet() with a new uncommitted voter set, but before the leader has
+ * updated its high watermark using the new voter set. In this case, we need to check that the majority
+ * of the last committed voter set has reached the current high watermark.
+ * */
+ cluster.leaderWithMaxEpoch().ifPresent(leaderNode -> {
+ leaderNode.client.highWatermark().ifPresent(highWatermark -> {
+ VoterSet voterSet = leaderNode.client.partitionState().lastVoterSet();
+ long numReachedHighWatermark = numReachedHighWatermark(highWatermark, voterSet.voterIds());
+ if (numReachedHighWatermark < cluster.majoritySize(voterSet.size())) {
+ leaderNode.client.partitionState().voterSetAtOffset(highWatermark - 1).ifPresent(otherVoterSet -> {
+ long nodesReachedHighWatermark = numReachedHighWatermark(highWatermark, otherVoterSet.voterIds());
+ assertTrue(
+ nodesReachedHighWatermark >= cluster.majoritySize(otherVoterSet.size()),
+ "Insufficient nodes have reached current high watermark. Expected at least " +
+ cluster.majoritySize(otherVoterSet.size()) + " but got " + nodesReachedHighWatermark);
+ });
+ return;
+ }
+ assertTrue(
+ numReachedHighWatermark >= cluster.majoritySize(voterSet.size()),
+ "Insufficient nodes have reached current high watermark. Expected at least " +
+ cluster.majoritySize(voterSet.size()) + " but got " + numReachedHighWatermark);
+ });
+ });
+ } else {
+ cluster.leaderHighWatermark().ifPresent(highWatermark -> {
+ long numReachedHighWatermark = numReachedHighWatermark(highWatermark, cluster.initialVoters.keySet());
+ assertTrue(
+ numReachedHighWatermark >= cluster.majoritySize(cluster.initialVoters.size()),
+ "Insufficient nodes have reached current high watermark");
+ });
+ }
+ }
+
+ private long numReachedHighWatermark(long highWatermark, Set<Integer> voterIds) {
+ return cluster.persistentStates.entrySet().stream()
+ .filter(entry -> voterIds.contains(entry.getKey()))
+ .filter(entry -> entry.getValue().log.endOffset().offset() >= highWatermark)
+ .count();
+ }
+ }
+
+ private static class AtMostOneUncommittedVotersRecord implements Invariant {
+ final Cluster cluster;
+
+ private AtMostOneUncommittedVotersRecord(Cluster cluster) {
+ this.cluster = cluster;
+ }
+
+ @Override
+ public void verify() {
+ cluster.leaderWithMaxEpoch().ifPresent(leaderNode -> {
+ leaderNode.log.readBatches(leaderNode.highWatermark(), OptionalLong.of(leaderNode.logEndOffset())).forEach(batch -> {
+ Optional<LogEntry> uncommittedVotersEntry = Optional.empty();
Review Comment:
can this just be a bool
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests pt 1 [kafka]
Posted by "kevin-wu24 (via GitHub)" <gi...@apache.org>.
kevin-wu24 commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r1970766229
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -487,6 +555,12 @@ private EventScheduler schedulerWithDefaultInvariants(Cluster cluster) {
return scheduler;
}
+ private EventScheduler schedulerWithKip853Invariants(Cluster cluster) {
+ EventScheduler scheduler = schedulerWithDefaultInvariants(cluster);
+ scheduler.addInvariant(new AtMostOneUncommittedVoterSet(cluster));
Review Comment:
I will have to refactor the `voterSetFromIds()` method to actually return something a non-empty optional if we add this invariant for non-KIP-853 tests, as that would throw an exception during verification.
My main reason for not adding it initially is that it is trivially invariant for a cluster not running in KIP-853 mode, and that under the current implementation of the verification, we have to iterate over the offsets in the log between the HWM and the offset where the leader's `lastVoterSet()` is. Also, the current implementation is slightly confusing, so I'll refactor it.
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests [kafka]
Posted by "kevin-wu24 (via GitHub)" <gi...@apache.org>.
kevin-wu24 commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r2006031583
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -475,6 +563,99 @@ void canRecoverFromSingleNodeCommittedDataLoss(
scheduler.runUntil(() -> cluster.allReachedHighWatermark(highWatermarkBeforeRestart + 10));
}
+ @Property(tries = 100, afterFailure = AfterFailureMode.SAMPLE_ONLY)
+ void canAddObserverAndRemovePartitionedVoter(
+ @ForAll int seed,
+ @ForAll @IntRange(min = 3, max = 5) int numVoters
+ ) {
+ Random random = new Random(seed);
+ Cluster cluster = new Cluster(numVoters, 1, random, true);
+ MessageRouter router = new MessageRouter(cluster);
+ EventScheduler scheduler = schedulerWithDefaultInvariants(cluster);
+ Set<Integer> expectedVoterIds = new HashSet<>(cluster.initialVoters.keySet());
+
+ initializeClusterAndStartAppending(cluster, router, scheduler, 10);
+
+ // Partition a random voter. Add first observer as new voter
+ int voterIdToRemove = cluster.running.get(random.nextInt(numVoters)).id();
+ router.filter(voterIdToRemove, new DropAllTraffic());
+ expectedVoterIds.add(numVoters);
+ addVoter(cluster, scheduler, numVoters, expectedVoterIds);
+
+ // Remove the partitioned voter
+ expectedVoterIds.remove(voterIdToRemove);
+ removeVoter(cluster, scheduler, voterIdToRemove, expectedVoterIds);
+ long highWatermark = cluster.maxHighWatermarkReached();
+ scheduler.runUntil(() -> cluster.allReachedHighWatermark(highWatermark + 10, expectedVoterIds));
+ }
+
+ @Property(tries = 100, afterFailure = AfterFailureMode.SAMPLE_ONLY)
+ void canRemoveAndAddBackVoter(
+ @ForAll int seed,
+ @ForAll @IntRange(min = 3, max = 5) int numVoters,
+ @ForAll @IntRange(min = 0, max = 2) int numObservers
+ ) {
+ Random random = new Random(seed);
+ Cluster cluster = new Cluster(numVoters, numObservers, random, true);
+ MessageRouter router = new MessageRouter(cluster);
+ EventScheduler scheduler = schedulerWithDefaultInvariants(cluster);
+ Set<Integer> expectedVoterIds = new HashSet<>(cluster.initialVoters.keySet());
+
+ initializeClusterAndStartAppending(cluster, router, scheduler, 10);
+
+ // Partition a random voter and remove it
+ int voterId = cluster.running.get(random.nextInt(numVoters)).id();
+ router.filter(voterId, new DropAllTraffic());
+ expectedVoterIds.remove(voterId);
+ removeVoter(cluster, scheduler, voterId, expectedVoterIds);
+
+ // Restore the network and add the voter back
+ router.filter(voterId, new PermitAllTraffic());
+ expectedVoterIds.add(voterId);
+ addVoter(cluster, scheduler, voterId, expectedVoterIds);
+
+ long highWatermark = cluster.maxHighWatermarkReached();
+ scheduler.runUntil(() -> cluster.allReachedHighWatermark(highWatermark + 10));
+ }
+
+ /**
+ * This method initializes the cluster and starts appending to the log.
+ * @param appendActionPeriodMs - the period at which the append action should be scheduled, in milliseconds
+ * This value is important for tests that add voters, since if the cluster is appending
+ * records too quickly, it may prevent the new voter from catching up to the leader's LEO
+ * so that it can be added to the voter set.
+ */
+ private void initializeClusterAndStartAppending(Cluster cluster, MessageRouter router, EventScheduler scheduler, int appendActionPeriodMs) {
+ cluster.startAll();
+ schedulePolling(scheduler, cluster, 3, 5);
+ scheduler.schedule(router::deliverAll, 0, 2, 5);
+ scheduler.schedule(new SequentialAppendAction(cluster), 0, appendActionPeriodMs, 3);
+ scheduler.runUntil(cluster::hasConsistentLeader);
+ scheduler.runUntil(() -> cluster.anyReachedHighWatermark(10));
+ }
+
+ private void addVoter(Cluster cluster, EventScheduler scheduler, int idToAdd, Set<Integer> expectedVoterIds) {
+ scheduler.scheduleUntil(
+ new AddVoterAction(cluster, cluster.running.get(idToAdd)),
+ () -> cluster.leaderHasCommittedVoterSet(expectedVoterIds),
+ 0,
+ 5,
+ 3
+ );
Review Comment:
I remember running into a lot of issues when trying to use `runUntil` directly. Mainly because we need to do two things:
1. repeatedly schedule the operation that wraps an Add/RemoveVoter RPC until it completes successfully
2. stop scheduling that operation
It's not sufficient to use the existing `EventScheduler#schedule`, because that schedules the RPC from that point thereafter. What happens if I want to test a case like removing and adding back the same voter?
Maybe the naming `scheduleUntil` is not clear, but it means that we schedule the event until some condition is met, then we stop scheduling that event.
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests pt 1 [kafka]
Posted by "kevin-wu24 (via GitHub)" <gi...@apache.org>.
kevin-wu24 commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r1970056520
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -1127,14 +1331,75 @@ private MajorityReachedHighWatermark(Cluster cluster) {
@Override
public void verify() {
- cluster.leaderHighWatermark().ifPresent(highWatermark -> {
- long numReachedHighWatermark = cluster.nodes.entrySet().stream()
- .filter(entry -> cluster.voters.containsKey(entry.getKey()))
- .filter(entry -> entry.getValue().log.endOffset().offset() >= highWatermark)
- .count();
- assertTrue(
- numReachedHighWatermark >= cluster.majoritySize(),
- "Insufficient nodes have reached current high watermark");
+ if (cluster.withKip853) {
+ /*
+ * For clusters running in KIP-853 mode, we check that a majority of at least one of:
+ * 1. the leader's voter set at the HWM
+ * 2. the leader's lastVoterSet()
+ * has reached the HWM. We need to perform a more elaborate check here because in clusters where
+ * an Add/RemoveVoter request increases/decreases the majority of voters value by 1, the leader
+ * could have used either majority value to update its HWM value. This is because depending on
+ * whether the leader read the most recent VotersRecord prior to updating its HWM value, the number
+ * of nodes (the majority) used to calculate that HWM value is different. This matters for invariant
+ * checking because we perform this verification on every message delivery.
+ * */
+ cluster.leaderWithMaxEpoch().ifPresent(leaderNode -> {
+ leaderNode.client.highWatermark().ifPresent(highWatermark -> {
+ VoterSet voterSet = leaderNode.client.partitionState().lastVoterSet();
+ long numReachedHighWatermark = numReachedHighWatermark(highWatermark, voterSet.voterIds());
+ if (numReachedHighWatermark < cluster.majoritySize(voterSet.size())) {
+ leaderNode.client.partitionState().voterSetAtOffset(highWatermark - 1).ifPresent(otherVoterSet -> {
Review Comment:
An "event" from the perspective of the simulation tests is the execution of one `Runnable`. This detail is important because `SequentialAppendAction` will directly call `client.prepareAppend` and `client.schedulePreparedAppend` which can wakeup the message queue (this is where my implementation of `AddVoterAction` adds the voter-to-be-added's `AddVoterRequest` and `API_VERSIONS_RESPONSE`).
~~This check is incorrect, and we should rely on only the last voter set. More explained below~~
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests pt 1 [kafka]
Posted by "kevin-wu24 (via GitHub)" <gi...@apache.org>.
kevin-wu24 commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r1970766229
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -487,6 +555,12 @@ private EventScheduler schedulerWithDefaultInvariants(Cluster cluster) {
return scheduler;
}
+ private EventScheduler schedulerWithKip853Invariants(Cluster cluster) {
+ EventScheduler scheduler = schedulerWithDefaultInvariants(cluster);
+ scheduler.addInvariant(new AtMostOneUncommittedVoterSet(cluster));
Review Comment:
I will have to refactor the `voterSetFromIds()` method to actually return something a non-empty optional if we add this invariant for non-KIP-853 tests, as that would throw an exception during verification.
My main reason for not adding it initially is that it is trivially invariant for a cluster not running in KIP-853 mode, and that under the current implementation of the verification, we have to iterate over the offsets in the log between the HWM and the offset where the leader's `lastVoterSet()` is.
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests pt 1 [kafka]
Posted by "kevin-wu24 (via GitHub)" <gi...@apache.org>.
kevin-wu24 commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r1970773425
##########
raft/src/main/java/org/apache/kafka/raft/VoterSet.java:
##########
@@ -343,6 +343,10 @@ public ReplicaKey voterKey() {
return voterKey;
}
+ public int voterId() {
Review Comment:
I think I just have to refactor `initialVoterSetFromIds` to not be a one liner, and yes we can.
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests pt 1 [kafka]
Posted by "ahuang98 (via GitHub)" <gi...@apache.org>.
ahuang98 commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r1975951752
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -1127,14 +1381,83 @@ private MajorityReachedHighWatermark(Cluster cluster) {
@Override
public void verify() {
- cluster.leaderHighWatermark().ifPresent(highWatermark -> {
- long numReachedHighWatermark = cluster.nodes.entrySet().stream()
- .filter(entry -> cluster.voters.containsKey(entry.getKey()))
- .filter(entry -> entry.getValue().log.endOffset().offset() >= highWatermark)
- .count();
- assertTrue(
- numReachedHighWatermark >= cluster.majoritySize(),
- "Insufficient nodes have reached current high watermark");
+ if (cluster.withKip853) {
+ /*
+ * For clusters running in KIP-853 mode, we check that a majority of at least one of the following
+ * voter sets has reached the high watermark:
+ * 1. the leader's voter set at the HWM (i.e. the last committed voter set)
+ * 2. the leader's lastVoterSet() (which may or may not be committed)
+ * Note that 1 and 2 can be the same set, but when they are not, lastVoterSet() is uncommitted,
+ * which follows from the AtMostOneUncommittedVoterSet invariant.
Review Comment:
I think the below paragraph could simply be replaced with
```
We need to consider both options because it is possible verify is called after the leader's partitionState is updated but not committed yet.
```
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests pt 1 [kafka]
Posted by "kevin-wu24 (via GitHub)" <gi...@apache.org>.
kevin-wu24 commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r1976012893
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -580,6 +576,30 @@ void canRecoverFromSingleNodeCommittedDataLoss(
scheduler.runUntil(() -> cluster.allReachedHighWatermark(highWatermarkBeforeRestart + 10));
}
+ @Property(tries = 100, afterFailure = AfterFailureMode.SAMPLE_ONLY)
+ void canSwapPartitionedVoterForObserver(
Review Comment:
how about `canRemovePartitionedVoterAndAddObserver`?
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests [kafka]
Posted by "kevin-wu24 (via GitHub)" <gi...@apache.org>.
kevin-wu24 commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r2003724311
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -1127,14 +1331,75 @@ private MajorityReachedHighWatermark(Cluster cluster) {
@Override
public void verify() {
- cluster.leaderHighWatermark().ifPresent(highWatermark -> {
- long numReachedHighWatermark = cluster.nodes.entrySet().stream()
- .filter(entry -> cluster.voters.containsKey(entry.getKey()))
- .filter(entry -> entry.getValue().log.endOffset().offset() >= highWatermark)
- .count();
- assertTrue(
- numReachedHighWatermark >= cluster.majoritySize(),
- "Insufficient nodes have reached current high watermark");
+ if (cluster.withKip853) {
+ /*
+ * For clusters running in KIP-853 mode, we check that a majority of at least one of:
+ * 1. the leader's voter set at the HWM
+ * 2. the leader's lastVoterSet()
+ * has reached the HWM. We need to perform a more elaborate check here because in clusters where
+ * an Add/RemoveVoter request increases/decreases the majority of voters value by 1, the leader
+ * could have used either majority value to update its HWM value. This is because depending on
+ * whether the leader read the most recent VotersRecord prior to updating its HWM value, the number
Review Comment:
The summary of the discussion as to why we need to check the HWM in this way is contained here: https://github.com/apache/kafka/pull/18987#discussion_r1971722138.
TLDR: the simulation test can call `verify` and look at the RaftClient internal state in between those two points (i.e. `partitionState` has been updated with latest `votersRecord`, but the HWM is still the old value from the last committed `votersRecord`).
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests [kafka]
Posted by "kevin-wu24 (via GitHub)" <gi...@apache.org>.
kevin-wu24 commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r2006278691
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -1127,14 +1331,75 @@ private MajorityReachedHighWatermark(Cluster cluster) {
@Override
public void verify() {
- cluster.leaderHighWatermark().ifPresent(highWatermark -> {
- long numReachedHighWatermark = cluster.nodes.entrySet().stream()
- .filter(entry -> cluster.voters.containsKey(entry.getKey()))
- .filter(entry -> entry.getValue().log.endOffset().offset() >= highWatermark)
- .count();
- assertTrue(
- numReachedHighWatermark >= cluster.majoritySize(),
- "Insufficient nodes have reached current high watermark");
+ if (cluster.withKip853) {
+ /*
+ * For clusters running in KIP-853 mode, we check that a majority of at least one of:
+ * 1. the leader's voter set at the HWM
+ * 2. the leader's lastVoterSet()
+ * has reached the HWM. We need to perform a more elaborate check here because in clusters where
+ * an Add/RemoveVoter request increases/decreases the majority of voters value by 1, the leader
+ * could have used either majority value to update its HWM value. This is because depending on
+ * whether the leader read the most recent VotersRecord prior to updating its HWM value, the number
+ * of nodes (the majority) used to calculate that HWM value is different. This matters for invariant
+ * checking because we perform this verification on every message delivery.
+ * */
+ cluster.leaderWithMaxEpoch().ifPresent(leaderNode -> {
+ leaderNode.client.highWatermark().ifPresent(highWatermark -> {
+ VoterSet voterSet = leaderNode.client.partitionState().lastVoterSet();
+ long numReachedHighWatermark = numReachedHighWatermark(highWatermark, voterSet.voterIds());
+ if (numReachedHighWatermark < cluster.majoritySize(voterSet.size())) {
+ leaderNode.client.partitionState().voterSetAtOffset(highWatermark - 1).ifPresent(otherVoterSet -> {
Review Comment:
I agree that this minimal voter set check is valid, but I think it is too relaxed for a cluster with dynamic voters. Specifically, I have concerns about this statement, but maybe I'm misunderstanding what is going on:
> KRaft is allowed to commit up to the HWM between voterSet1 or voterSet2, with either voter set.
From my understanding of the code, once KRaft becomes aware of voterSet2, which happens at the following point in execution because `partitionState.lastVoterSet()` is set to voterSet2: `appendAsLeader -> partitionState.updateState()`, it should be using voterSet2 to calculate the HWM thereafter, which it does in the following code from `KafkaRaftClient#updateLeaderEndOffsetAndTimestamp`, which is called in `flushLeaderLog`:
```
if (state.updateLocalState(endOffsetMetadata, partitionState.lastVoterSet())) {
onUpdateLeaderHighWatermark(state, currentTimeMs);
}
```
The only point where KafkaRaftClient state is the following:
1. `partitionState.lastVoterSet() = votersSet2`
2. HWM value was calculated using voterSet1
is between `appendAsLeader -> partitionState.updateState()` and `flushLeaderLog` when voterSet2 is written to the log.
When checking the invariant in `verify()`, the RaftNode either has an uncommitted voter set in its log or it doesn't. If it doesn't, using a majority of `lastVoterSet()` to verify the HWM value is always correct. If it does, using a majority of `lastVoterSet()` to verify the HWM is correct except in the case I mentioned above.
For example, in the case of adding a voter, voterSet1 is the minimal voter set with majority size X, and voterSet2 is the uncommitted voter set with majority size X + 1. This means when verifying the HWM value, we would be checking that X nodes have replicated up to the HWM offset, not X + 1 nodes, which should only happen if the HWM value was actually calculated with voterSet1.
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests [kafka]
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r2060745648
##########
raft/src/test/java/org/apache/kafka/raft/MockLog.java:
##########
@@ -626,6 +676,14 @@ public String toString() {
);
}
+ public VoterSet lastVoterSet() {
+ return lastVoterSet;
+ }
+
+ public VoterSet lastCommittedVoterSet() {
+ return lastCommittedVoterSet;
+ }
Review Comment:
I don't see any test that show this functionality is correct. Please add test to MockLogTest.
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -734,12 +733,13 @@ private static class ScheduledUntilConditionMetEvent extends Event {
final Supplier<Boolean> exitCondition;
protected ScheduledUntilConditionMetEvent(Runnable action,
- int eventId,
- Random random,
- long deadlineMs,
- int periodMs,
- int jitterMs,
- Supplier<Boolean> exitCondition) {
+ int eventId,
Review Comment:
Please use this formatting:
```java
protected ScheduledUntilConditionMetEvent(
Runnable action,
int eventId,
...
```
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests pt 1 [kafka]
Posted by "kevin-wu24 (via GitHub)" <gi...@apache.org>.
kevin-wu24 commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r1980465079
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -297,18 +309,83 @@ void canMakeProgressIfMajorityIsReachable(
scheduler.runUntil(() -> cluster.allReachedHighWatermark(2 * restoredLogEndOffset));
}
+ @Property(tries = 100, afterFailure = AfterFailureMode.SAMPLE_ONLY)
+ void canAddVotersIfMajorityIsReachable(
+ @ForAll int seed,
+ @ForAll @IntRange(min = 1, max = 3) int numObservers
+ ) {
+ int numVoters = 5;
+ Random random = new Random(seed);
+ Cluster cluster = new Cluster(numVoters, numObservers, random, true);
+ MessageRouter router = new MessageRouter(cluster);
+ EventScheduler scheduler = schedulerWithDefaultInvariants(cluster);
+ Set<Integer> expectedVoterIds = new HashSet<>(cluster.initialVoters.keySet());
+
+ initializeClusterAndStartAppending(cluster, router, scheduler, 10);
+
+ // Partition the nodes into two sets. Nodes are reachable within each set,
+ // but the two sets cannot communicate with each other. We should be able
+ // to make progress even if an election is needed in the larger set.
+ int firstObserverId = numVoters;
+ router.filter(
+ 0,
+ new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(2, 3, 4, firstObserverId)))
+ );
+ router.filter(
+ 1,
+ new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(2, 3, 4, firstObserverId)))
+ );
+ router.filter(2, new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(0, 1))));
+ router.filter(3, new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(0, 1))));
+ router.filter(4, new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(0, 1))));
+ router.filter(firstObserverId, new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(0, 1))));
+
+ long partitionLogEndOffset = cluster.maxLogEndOffset();
+ scheduler.runUntil(() -> cluster.anyReachedHighWatermark(2 * partitionLogEndOffset));
+
+ long minorityHighWatermark = cluster.maxHighWatermarkReached(Set.of(0, 1));
+ long majorityHighWatermark = cluster.maxHighWatermarkReached(Set.of(2, 3, 4));
+
+ assertTrue(
+ majorityHighWatermark > minorityHighWatermark,
+ String.format(
+ "majorityHighWatermark = %s, minorityHighWatermark = %s",
+ majorityHighWatermark,
+ minorityHighWatermark
+ )
+ );
+
+ // Verify we can add a voter, since a majority is still reachable
+ addVoter(cluster, scheduler, firstObserverId, expectedVoterIds);
+
+ // Now restore the partition and verify everyone catches up
+ router.filter(0, new PermitAllTraffic());
+ router.filter(1, new PermitAllTraffic());
+ router.filter(2, new PermitAllTraffic());
+ router.filter(3, new PermitAllTraffic());
+ router.filter(4, new PermitAllTraffic());
+ router.filter(firstObserverId, new PermitAllTraffic());
+
+ long restoredLogEndOffset = cluster.maxLogEndOffset();
+ scheduler.runUntil(() -> cluster.allReachedHighWatermark(2 * restoredLogEndOffset));
+ }
+
@Property(tries = 100, afterFailure = AfterFailureMode.SAMPLE_ONLY)
void leadershipAssignedOnlyOnceWithNetworkPartitionIfThereExistsMajority(
@ForAll int seed,
- @ForAll @IntRange(min = 0, max = 3) int numObservers
+ @ForAll @IntRange(min = 0, max = 3) int numObservers,
+ @ForAll boolean withKip853
) {
int numVoters = 5;
Random random = new Random(seed);
- Cluster cluster = new Cluster(numVoters, numObservers, random);
+ Cluster cluster = new Cluster(numVoters, numObservers, random, withKip853);
MessageRouter router = new MessageRouter(cluster);
EventScheduler scheduler = schedulerWithDefaultInvariants(cluster);
scheduler.addInvariant(new StableLeadership(cluster));
+ // Start cluster
+ cluster.startAll();
Review Comment:
I guess what we mean by "cluster starting" is not super clear. I think the intention of this test is to have a network partition exist before the KRaft protocol starts (i.e. nodes sending messages to each other and performing the initial leader election). This is what it still does now, since the partition is defined before calls to the following:
```
schedulePolling(scheduler, cluster, 3, 5);
scheduler.schedule(router::deliverAll, 0, 2, 1);
scheduler.schedule(new SequentialAppendAction(cluster), 0, 2, 1);
scheduler.runUntil(cluster::hasConsistentLeader);
```
What `cluster.startAll()` does is just update the `Cluster`'s state and initialize all the stuff KRaft uses, but if no messages are being delivered and no polling is occurring, so I don't think we've actually "started" yet.
I think the function naming is misleading, it should be something like `initializeNodes()`. What "starts" KRaft from my perspective are the lines above.
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #18987: KAFKA-16580: Enable dynamic quorum reconfiguration for raft simulation tests
URL: https://github.com/apache/kafka/pull/18987
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests pt 1 [kafka]
Posted by "kevin-wu24 (via GitHub)" <gi...@apache.org>.
kevin-wu24 commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r1968672262
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -487,6 +555,12 @@ private EventScheduler schedulerWithDefaultInvariants(Cluster cluster) {
return scheduler;
}
+ private EventScheduler schedulerWithKip853Invariants(Cluster cluster) {
+ EventScheduler scheduler = schedulerWithDefaultInvariants(cluster);
+ scheduler.addInvariant(new AtMostOneUncommittedVoterSet(cluster));
Review Comment:
It can apply to all tests, and perhaps we should be just adding it to the default invariant set. I thought it would break the non-KIP-853 tests, so I haven't actually tried it yet.
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests pt 1 [kafka]
Posted by "ahuang98 (via GitHub)" <gi...@apache.org>.
ahuang98 commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r1970747972
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -487,6 +555,12 @@ private EventScheduler schedulerWithDefaultInvariants(Cluster cluster) {
return scheduler;
}
+ private EventScheduler schedulerWithKip853Invariants(Cluster cluster) {
+ EventScheduler scheduler = schedulerWithDefaultInvariants(cluster);
+ scheduler.addInvariant(new AtMostOneUncommittedVoterSet(cluster));
Review Comment:
seems like an issue if it breaks non-KIP-853 tests - maybe worth adding it to the default invariants to see if any tests break - technically this new version of the invariant is more lenient than the old version right?
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests [kafka]
Posted by "kevin-wu24 (via GitHub)" <gi...@apache.org>.
kevin-wu24 commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r2006461608
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -1127,14 +1331,75 @@ private MajorityReachedHighWatermark(Cluster cluster) {
@Override
public void verify() {
- cluster.leaderHighWatermark().ifPresent(highWatermark -> {
- long numReachedHighWatermark = cluster.nodes.entrySet().stream()
- .filter(entry -> cluster.voters.containsKey(entry.getKey()))
- .filter(entry -> entry.getValue().log.endOffset().offset() >= highWatermark)
- .count();
- assertTrue(
- numReachedHighWatermark >= cluster.majoritySize(),
- "Insufficient nodes have reached current high watermark");
+ if (cluster.withKip853) {
+ /*
+ * For clusters running in KIP-853 mode, we check that a majority of at least one of:
+ * 1. the leader's voter set at the HWM
+ * 2. the leader's lastVoterSet()
+ * has reached the HWM. We need to perform a more elaborate check here because in clusters where
+ * an Add/RemoveVoter request increases/decreases the majority of voters value by 1, the leader
+ * could have used either majority value to update its HWM value. This is because depending on
+ * whether the leader read the most recent VotersRecord prior to updating its HWM value, the number
+ * of nodes (the majority) used to calculate that HWM value is different. This matters for invariant
+ * checking because we perform this verification on every message delivery.
+ * */
+ cluster.leaderWithMaxEpoch().ifPresent(leaderNode -> {
+ leaderNode.client.highWatermark().ifPresent(highWatermark -> {
+ VoterSet voterSet = leaderNode.client.partitionState().lastVoterSet();
+ long numReachedHighWatermark = numReachedHighWatermark(highWatermark, voterSet.voterIds());
+ if (numReachedHighWatermark < cluster.majoritySize(voterSet.size())) {
+ leaderNode.client.partitionState().voterSetAtOffset(highWatermark - 1).ifPresent(otherVoterSet -> {
Review Comment:
If the main concern is to avoid doing a separate check when the cluster has KIP-853 enabled, I think I can just remove the if statement, using `lastVoterSet()` with a static voter set always works (honestly not sure why this was even there). This way we are not relying on the implementation detail that each voter set can only differ by 1 voter.
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests [kafka]
Posted by "kevin-wu24 (via GitHub)" <gi...@apache.org>.
kevin-wu24 commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r2006461608
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -1127,14 +1331,75 @@ private MajorityReachedHighWatermark(Cluster cluster) {
@Override
public void verify() {
- cluster.leaderHighWatermark().ifPresent(highWatermark -> {
- long numReachedHighWatermark = cluster.nodes.entrySet().stream()
- .filter(entry -> cluster.voters.containsKey(entry.getKey()))
- .filter(entry -> entry.getValue().log.endOffset().offset() >= highWatermark)
- .count();
- assertTrue(
- numReachedHighWatermark >= cluster.majoritySize(),
- "Insufficient nodes have reached current high watermark");
+ if (cluster.withKip853) {
+ /*
+ * For clusters running in KIP-853 mode, we check that a majority of at least one of:
+ * 1. the leader's voter set at the HWM
+ * 2. the leader's lastVoterSet()
+ * has reached the HWM. We need to perform a more elaborate check here because in clusters where
+ * an Add/RemoveVoter request increases/decreases the majority of voters value by 1, the leader
+ * could have used either majority value to update its HWM value. This is because depending on
+ * whether the leader read the most recent VotersRecord prior to updating its HWM value, the number
+ * of nodes (the majority) used to calculate that HWM value is different. This matters for invariant
+ * checking because we perform this verification on every message delivery.
+ * */
+ cluster.leaderWithMaxEpoch().ifPresent(leaderNode -> {
+ leaderNode.client.highWatermark().ifPresent(highWatermark -> {
+ VoterSet voterSet = leaderNode.client.partitionState().lastVoterSet();
+ long numReachedHighWatermark = numReachedHighWatermark(highWatermark, voterSet.voterIds());
+ if (numReachedHighWatermark < cluster.majoritySize(voterSet.size())) {
+ leaderNode.client.partitionState().voterSetAtOffset(highWatermark - 1).ifPresent(otherVoterSet -> {
Review Comment:
If the main concern is to avoid doing a separate check when the cluster has KIP-853 enabled, I think I can just remove the if statement, using `lastVoterSet()` with a static voter set always works (honestly not sure why this was even there). This way we are not relying on the implementation detail that each voter set can only differ by 1 voter.
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests pt 1 [kafka]
Posted by "kevin-wu24 (via GitHub)" <gi...@apache.org>.
kevin-wu24 commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r1970773425
##########
raft/src/main/java/org/apache/kafka/raft/VoterSet.java:
##########
@@ -343,6 +343,10 @@ public ReplicaKey voterKey() {
return voterKey;
}
+ public int voterId() {
Review Comment:
I think I just have to refactor `initialVoterSetFromIds` to not be a one liner, and yes we can.
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests pt 1 [kafka]
Posted by "kevin-wu24 (via GitHub)" <gi...@apache.org>.
kevin-wu24 commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r1971722138
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -1127,14 +1331,75 @@ private MajorityReachedHighWatermark(Cluster cluster) {
@Override
public void verify() {
- cluster.leaderHighWatermark().ifPresent(highWatermark -> {
- long numReachedHighWatermark = cluster.nodes.entrySet().stream()
- .filter(entry -> cluster.voters.containsKey(entry.getKey()))
- .filter(entry -> entry.getValue().log.endOffset().offset() >= highWatermark)
- .count();
- assertTrue(
- numReachedHighWatermark >= cluster.majoritySize(),
- "Insufficient nodes have reached current high watermark");
+ if (cluster.withKip853) {
+ /*
+ * For clusters running in KIP-853 mode, we check that a majority of at least one of:
+ * 1. the leader's voter set at the HWM
Review Comment:
Sorry, upon further looking into this, I think the issue is just slightly different that what I've described thus far.
Essentially, when we fail the invariant check when only using `lastVoterSet()`, what's going on is that the caller of `verify()` is a different thread (lets say the main event scheduler thread) than the calls to `KafkaRaftClient#poll`, and `KafkaRaftClient#prepareAppend/schedulePreparedAppend` which are what run on the threads created by`Event#execute` to continuously update KRaft internal state. You can double check by looking at `schedulePolling` and `SequentialAppendAction`.
The main event scheduler thread is looking at a bunch of the leader's internal state (e.g. partitionState and highWatermark), which can definitely be in the following state: `partitionState` has been updated with a new voter set, but a new `highWatermark` value has not yet been calculated with this voter set yet, which could cause this invariant check to fail when only looking at `lastVoterSet()`. The leader may not be in this state for very long, since it's in this state in between the calls to `appendAsLeader -> updateState -> maybeLoadLog` and `flushLeaderLog -> maybeUpdateHighWatermark`, but it nevertheless looks like a valid state to me.
Invariants as a concept are predicates that should be true for the system at all times, no matter when you check them, so I think performing invariant verification how it's currently implemented is fine, as we could be checking the internal states of our raft nodes at any point in their execution. However, it just means we have to consider these "intermediary" states when performing verification.
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests pt 1 [kafka]
Posted by "kevin-wu24 (via GitHub)" <gi...@apache.org>.
kevin-wu24 commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r1976010924
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -117,40 +125,101 @@ public class RaftEventSimulationTest {
void canElectInitialLeader(
@ForAll int seed,
@ForAll @IntRange(min = 1, max = 5) int numVoters,
- @ForAll @IntRange(min = 0, max = 5) int numObservers
+ @ForAll @IntRange(min = 0, max = 5) int numObservers,
+ @ForAll boolean withKip853
) {
Random random = new Random(seed);
- Cluster cluster = new Cluster(numVoters, numObservers, random);
+ Cluster cluster = new Cluster(numVoters, numObservers, random, withKip853);
MessageRouter router = new MessageRouter(cluster);
EventScheduler scheduler = schedulerWithDefaultInvariants(cluster);
- cluster.startAll();
- schedulePolling(scheduler, cluster, 3, 5);
- scheduler.schedule(router::deliverAll, 0, 2, 1);
- scheduler.schedule(new SequentialAppendAction(cluster), 0, 2, 3);
- scheduler.runUntil(cluster::hasConsistentLeader);
- scheduler.runUntil(() -> cluster.allReachedHighWatermark(10));
+ initializeClusterAndStartAppending(cluster, router, scheduler, 2);
+ }
+
+ @Property(tries = 100, afterFailure = AfterFailureMode.SAMPLE_ONLY)
+ void canElectInitialLeaderAndAddVoter(
+ @ForAll int seed,
+ @ForAll @IntRange(min = 1, max = 5) int numVoters,
+ @ForAll @IntRange(min = 1, max = 5) int numObservers
+ ) {
+ Random random = new Random(seed);
+ Cluster cluster = new Cluster(numVoters, numObservers, random, true);
+ MessageRouter router = new MessageRouter(cluster);
+ EventScheduler scheduler = schedulerWithKip853Invariants(cluster);
+
+ initializeClusterAndStartAppending(cluster, router, scheduler, 10);
+ int firstObserverId = numVoters;
+ scheduler.schedule(new AddVoterAction(cluster, cluster.running.get(firstObserverId)), 0, 5, 3);
+ scheduler.runUntil(() -> cluster.leaderWithMaxEpoch().get().client.partitionState().lastVoterSet().size() == numVoters + 1);
+ VoterSet latestVoterSet = cluster.leaderWithMaxEpoch().get().client.partitionState().lastVoterSet();
+ scheduler.runUntil(() -> cluster.allHaveLatestVoterSet(latestVoterSet));
+ }
+
+ @Property(tries = 100, afterFailure = AfterFailureMode.SAMPLE_ONLY)
+ void canAddAndRemoveVoters(
Review Comment:
I actually added a similar test in the most recent commit, which is to remove a voter that we partition and then add an observer. It's a similar case but I can add one where we partition a voter -> remove it -> restore networking -> add it back.
For removing specifically the leader, IIUC, in tests where I remove nodes, I remove a random node so that the `tries` property means we will remove a leader with probability = 1/numVoters on each try.
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests pt 1 [kafka]
Posted by "kevin-wu24 (via GitHub)" <gi...@apache.org>.
kevin-wu24 commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r1980465079
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -297,18 +309,83 @@ void canMakeProgressIfMajorityIsReachable(
scheduler.runUntil(() -> cluster.allReachedHighWatermark(2 * restoredLogEndOffset));
}
+ @Property(tries = 100, afterFailure = AfterFailureMode.SAMPLE_ONLY)
+ void canAddVotersIfMajorityIsReachable(
+ @ForAll int seed,
+ @ForAll @IntRange(min = 1, max = 3) int numObservers
+ ) {
+ int numVoters = 5;
+ Random random = new Random(seed);
+ Cluster cluster = new Cluster(numVoters, numObservers, random, true);
+ MessageRouter router = new MessageRouter(cluster);
+ EventScheduler scheduler = schedulerWithDefaultInvariants(cluster);
+ Set<Integer> expectedVoterIds = new HashSet<>(cluster.initialVoters.keySet());
+
+ initializeClusterAndStartAppending(cluster, router, scheduler, 10);
+
+ // Partition the nodes into two sets. Nodes are reachable within each set,
+ // but the two sets cannot communicate with each other. We should be able
+ // to make progress even if an election is needed in the larger set.
+ int firstObserverId = numVoters;
+ router.filter(
+ 0,
+ new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(2, 3, 4, firstObserverId)))
+ );
+ router.filter(
+ 1,
+ new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(2, 3, 4, firstObserverId)))
+ );
+ router.filter(2, new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(0, 1))));
+ router.filter(3, new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(0, 1))));
+ router.filter(4, new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(0, 1))));
+ router.filter(firstObserverId, new DropOutboundRequestsTo(cluster.endpointsFromIds(Set.of(0, 1))));
+
+ long partitionLogEndOffset = cluster.maxLogEndOffset();
+ scheduler.runUntil(() -> cluster.anyReachedHighWatermark(2 * partitionLogEndOffset));
+
+ long minorityHighWatermark = cluster.maxHighWatermarkReached(Set.of(0, 1));
+ long majorityHighWatermark = cluster.maxHighWatermarkReached(Set.of(2, 3, 4));
+
+ assertTrue(
+ majorityHighWatermark > minorityHighWatermark,
+ String.format(
+ "majorityHighWatermark = %s, minorityHighWatermark = %s",
+ majorityHighWatermark,
+ minorityHighWatermark
+ )
+ );
+
+ // Verify we can add a voter, since a majority is still reachable
+ addVoter(cluster, scheduler, firstObserverId, expectedVoterIds);
+
+ // Now restore the partition and verify everyone catches up
+ router.filter(0, new PermitAllTraffic());
+ router.filter(1, new PermitAllTraffic());
+ router.filter(2, new PermitAllTraffic());
+ router.filter(3, new PermitAllTraffic());
+ router.filter(4, new PermitAllTraffic());
+ router.filter(firstObserverId, new PermitAllTraffic());
+
+ long restoredLogEndOffset = cluster.maxLogEndOffset();
+ scheduler.runUntil(() -> cluster.allReachedHighWatermark(2 * restoredLogEndOffset));
+ }
+
@Property(tries = 100, afterFailure = AfterFailureMode.SAMPLE_ONLY)
void leadershipAssignedOnlyOnceWithNetworkPartitionIfThereExistsMajority(
@ForAll int seed,
- @ForAll @IntRange(min = 0, max = 3) int numObservers
+ @ForAll @IntRange(min = 0, max = 3) int numObservers,
+ @ForAll boolean withKip853
) {
int numVoters = 5;
Random random = new Random(seed);
- Cluster cluster = new Cluster(numVoters, numObservers, random);
+ Cluster cluster = new Cluster(numVoters, numObservers, random, withKip853);
MessageRouter router = new MessageRouter(cluster);
EventScheduler scheduler = schedulerWithDefaultInvariants(cluster);
scheduler.addInvariant(new StableLeadership(cluster));
+ // Start cluster
+ cluster.startAll();
Review Comment:
I guess what we mean by "cluster starting" is not super clear. I think the intention of this test is to have a network partition exist before the KRaft protocol starts (i.e. nodes sending messages to each other and performing the initial leader election). This is what it still does now, since the partition is defined before calls to the following:
```
schedulePolling(scheduler, cluster, 3, 5);
scheduler.schedule(router::deliverAll, 0, 2, 1);
scheduler.schedule(new SequentialAppendAction(cluster), 0, 2, 1);
scheduler.runUntil(cluster::hasConsistentLeader);
```
What `cluster.startAll()` does is just update the `Cluster`'s state and initialize all the stuff KRaft uses, but if no messages are being delivered and no polling is occurring, I don't think we've actually "started" yet.
I think the function naming is misleading, it should be something like `initializeNodes()`. What "starts" KRaft from my perspective are the lines above.
--
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-16580: Enable dynamic quorum reconfiguration for raft simulation tests pt 1 [kafka]
Posted by "kevin-wu24 (via GitHub)" <gi...@apache.org>.
kevin-wu24 commented on code in PR #18987:
URL: https://github.com/apache/kafka/pull/18987#discussion_r1980459283
##########
raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java:
##########
@@ -117,40 +126,58 @@ public class RaftEventSimulationTest {
void canElectInitialLeader(
@ForAll int seed,
@ForAll @IntRange(min = 1, max = 5) int numVoters,
- @ForAll @IntRange(min = 0, max = 5) int numObservers
+ @ForAll @IntRange(min = 0, max = 5) int numObservers,
+ @ForAll boolean withKip853
) {
Random random = new Random(seed);
- Cluster cluster = new Cluster(numVoters, numObservers, random);
+ Cluster cluster = new Cluster(numVoters, numObservers, random, withKip853);
MessageRouter router = new MessageRouter(cluster);
EventScheduler scheduler = schedulerWithDefaultInvariants(cluster);
- cluster.startAll();
- schedulePolling(scheduler, cluster, 3, 5);
- scheduler.schedule(router::deliverAll, 0, 2, 1);
- scheduler.schedule(new SequentialAppendAction(cluster), 0, 2, 3);
- scheduler.runUntil(cluster::hasConsistentLeader);
- scheduler.runUntil(() -> cluster.allReachedHighWatermark(10));
+ initializeClusterAndStartAppending(cluster, router, scheduler, 2);
+ }
+
+ @Property(tries = 100, afterFailure = AfterFailureMode.SAMPLE_ONLY)
+ void canAddAndRemoveVoters(
+ @ForAll int seed
+ ) {
+ Random random = new Random(seed);
+ int numVoters = 3;
+ int numObservers = 2;
+ Cluster cluster = new Cluster(numVoters, numObservers, random, true);
+ MessageRouter router = new MessageRouter(cluster);
+ EventScheduler scheduler = schedulerWithDefaultInvariants(cluster);
+ Set<Integer> expectedVoterIds = new HashSet<>(cluster.initialVoters.keySet());
+
+ initializeClusterAndStartAppending(cluster, router, scheduler, 10);
+
+ // Add all observers to voter set one by one
+ for (int voterIdToAdd = numVoters; voterIdToAdd < numVoters + numObservers; voterIdToAdd++) {
+ addVoter(cluster, scheduler, voterIdToAdd, expectedVoterIds);
+ }
+ runUntilNewHighWatermark(cluster, scheduler);
Review Comment:
I agree it is arbitrary. However, other tests do end with lines like
```
scheduler.runUntil(() -> cluster.allReachedHighWatermark(highWatermark + 10));
```
It's mainly just to actually continue delivering messages and polling a bit on the raft nodes after we commit the VotersRecord from calling `addVoter/removeVoter`. Otherwise, this test and others like it would just end after we commit the VotersRecord from the last call to `addVoter`.
--
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