You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "ahuang98 (via GitHub)" <gi...@apache.org> on 2025/05/07 22:57:01 UTC
[PR] [KAFKA-18345] Prevent deadlocked elections [kafka]
ahuang98 opened a new pull request, #19658:
URL: https://github.com/apache/kafka/pull/19658
After a small number of retries, it will become statistically likely that the exponential backoff returned by binaryExponentialElectionBackoffMs gets stuck at quorumConfig.electionBackoffMaxMs which has no jitter in it.
This change lowers `RETRY_BACKOFF_BASE_MS` to 50ms from 100ms and uses this as jitter for the configurable electionBackoffMaxMs. Basically, with retries=1, the returned backoff is 50ms, but after that the returned backoff has increasing randomness and likelihood to be larger until the first calculation exceeds the configured electionBackoffMaxMs + jitter.
--
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-18345] Prevent deadlocked elections [kafka]
Posted by "ahuang98 (via GitHub)" <gi...@apache.org>.
ahuang98 commented on code in PR #19658:
URL: https://github.com/apache/kafka/pull/19658#discussion_r2082481731
##########
raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientTest.java:
##########
@@ -4569,6 +4574,55 @@ public void testObserverFetchWithNoLocalId(boolean withKip853Rpc) throws Excepti
assertEquals(3, context.log.lastFetchedEpoch());
}
+ @Test
+ public void testExponentialElectionBackoffMs() {
+ Random mockedRandom = Mockito.mock(Random.class);
+ int electionBackoffMaxMs = 1000;
+
+ // test the bound of the first call to random.nextInt
+ for (int retries = 1; retries < 11; retries++) {
+ binaryExponentialElectionBackoffMs(electionBackoffMaxMs, retries, mockedRandom);
+ ArgumentCaptor<Integer> nextIntCaptor = ArgumentCaptor.forClass(Integer.class);
+ Mockito.verify(mockedRandom).nextInt(Mockito.eq(1), nextIntCaptor.capture());
+ int actualBound = nextIntCaptor.getValue();
+ int expectedBound = (int) (2 * Math.pow(2, retries - 1));
+ assertEquals(expectedBound, actualBound, "Incorrect bound for retries=" + retries);
+ Mockito.clearInvocations(mockedRandom);
+ }
+ // after the 10th retry, the bound of the first call to random.nextInt will remain capped to
+ // (RETRY_BACKOFF_BASE_MS * 2 << 10)=2048 to prevent overflow
+ int firstNextIntMaxBound = 2048;
+ for (int retries = 11; retries < 13; retries++) {
+ binaryExponentialElectionBackoffMs(electionBackoffMaxMs, retries, mockedRandom);
+ ArgumentCaptor<Integer> nextIntCaptor = ArgumentCaptor.forClass(Integer.class);
+ Mockito.verify(mockedRandom).nextInt(Mockito.eq(1), nextIntCaptor.capture());
+ int actualBound = nextIntCaptor.getValue();
+ assertEquals(firstNextIntMaxBound, actualBound);
+ Mockito.clearInvocations(mockedRandom);
+ }
+
+ // test that the return value of the method is capped to QUORUM_ELECTION_BACKOFF_MAX_MS_CONFIG + jitter
+ int jitterMs = 50;
+ // any bound >= (1000 + jitter)/(RETRY_BACKOFF_BASE_MS)=21 will result in returning this cap
+ for (int firstNextInt : Arrays.asList(21, 1000, firstNextIntMaxBound)) {
+ Mockito.when(mockedRandom.nextInt(1, firstNextIntMaxBound)).thenReturn(firstNextInt);
+ Mockito.when(mockedRandom.nextInt(RETRY_BACKOFF_BASE_MS)).thenReturn(jitterMs);
+
+ int returnedBackoffMs = binaryExponentialElectionBackoffMs(electionBackoffMaxMs, 11, mockedRandom);
+
+ // verify nextInt was called on both expected bounds
+ ArgumentCaptor<Integer> nextIntCaptor = ArgumentCaptor.forClass(Integer.class);
+ Mockito.verify(mockedRandom).nextInt(Mockito.eq(1), nextIntCaptor.capture());
+ Mockito.verify(mockedRandom).nextInt(nextIntCaptor.capture());
+ List<Integer> allCapturedBounds = nextIntCaptor.getAllValues();
+ assertEquals(firstNextIntMaxBound, allCapturedBounds.get(0));
+ assertEquals(RETRY_BACKOFF_BASE_MS, allCapturedBounds.get(1));
+
+ // finally verify the backoff returned is capped to electionBackoffMaxMs + jitterMs
+ assertEquals(electionBackoffMaxMs + jitterMs, returnedBackoffMs);
+ }
Review Comment:
Ah, I see, I had changed this into a loop in an earlier commit. Thanks for pointing this out!
--
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-18345] Prevent deadlocked elections [kafka]
Posted by "Yunyung (via GitHub)" <gi...@apache.org>.
Yunyung commented on code in PR #19658:
URL: https://github.com/apache/kafka/pull/19658#discussion_r2081198851
##########
raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientTest.java:
##########
@@ -4569,6 +4574,55 @@ public void testObserverFetchWithNoLocalId(boolean withKip853Rpc) throws Excepti
assertEquals(3, context.log.lastFetchedEpoch());
}
+ @Test
+ public void testExponentialElectionBackoffMs() {
+ Random mockedRandom = Mockito.mock(Random.class);
+ int electionBackoffMaxMs = 1000;
+
+ // test the bound of the first call to random.nextInt
+ for (int retries = 1; retries < 11; retries++) {
+ binaryExponentialElectionBackoffMs(electionBackoffMaxMs, retries, mockedRandom);
+ ArgumentCaptor<Integer> nextIntCaptor = ArgumentCaptor.forClass(Integer.class);
+ Mockito.verify(mockedRandom).nextInt(Mockito.eq(1), nextIntCaptor.capture());
+ int actualBound = nextIntCaptor.getValue();
+ int expectedBound = (int) (2 * Math.pow(2, retries - 1));
+ assertEquals(expectedBound, actualBound, "Incorrect bound for retries=" + retries);
+ Mockito.clearInvocations(mockedRandom);
+ }
+ // after the 10th retry, the bound of the first call to random.nextInt will remain capped to
+ // (RETRY_BACKOFF_BASE_MS * 2 << 10)=2048 to prevent overflow
+ int firstNextIntMaxBound = 2048;
+ for (int retries = 11; retries < 13; retries++) {
+ binaryExponentialElectionBackoffMs(electionBackoffMaxMs, retries, mockedRandom);
+ ArgumentCaptor<Integer> nextIntCaptor = ArgumentCaptor.forClass(Integer.class);
+ Mockito.verify(mockedRandom).nextInt(Mockito.eq(1), nextIntCaptor.capture());
+ int actualBound = nextIntCaptor.getValue();
+ assertEquals(firstNextIntMaxBound, actualBound);
+ Mockito.clearInvocations(mockedRandom);
+ }
+
+ // test that the return value of the method is capped to QUORUM_ELECTION_BACKOFF_MAX_MS_CONFIG + jitter
+ int jitterMs = 50;
+ // any bound >= (1000 + jitter)/(RETRY_BACKOFF_BASE_MS)=21 will result in returning this cap
+ for (int firstNextInt : Arrays.asList(21, 1000, firstNextIntMaxBound)) {
+ Mockito.when(mockedRandom.nextInt(1, firstNextIntMaxBound)).thenReturn(firstNextInt);
+ Mockito.when(mockedRandom.nextInt(RETRY_BACKOFF_BASE_MS)).thenReturn(jitterMs);
+
+ int returnedBackoffMs = binaryExponentialElectionBackoffMs(electionBackoffMaxMs, 11, mockedRandom);
+
+ // verify nextInt was called on both expected bounds
+ ArgumentCaptor<Integer> nextIntCaptor = ArgumentCaptor.forClass(Integer.class);
+ Mockito.verify(mockedRandom).nextInt(Mockito.eq(1), nextIntCaptor.capture());
+ Mockito.verify(mockedRandom).nextInt(nextIntCaptor.capture());
+ List<Integer> allCapturedBounds = nextIntCaptor.getAllValues();
+ assertEquals(firstNextIntMaxBound, allCapturedBounds.get(0));
+ assertEquals(RETRY_BACKOFF_BASE_MS, allCapturedBounds.get(1));
+
+ // finally verify the backoff returned is capped to electionBackoffMaxMs + jitterMs
+ assertEquals(electionBackoffMaxMs + jitterMs, returnedBackoffMs);
+ }
Review Comment:
Add Mockito.clearInvocations(mockedRandom);
--
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-18345] Prevent deadlocked elections [kafka]
Posted by "TaiJuWu (via GitHub)" <gi...@apache.org>.
TaiJuWu commented on code in PR #19658:
URL: https://github.com/apache/kafka/pull/19658#discussion_r2082277116
##########
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java:
##########
@@ -1048,14 +1053,18 @@ private void maybeHandleElectionLoss(NomineeState state, long currentTimeMs) {
}
}
- private int binaryExponentialElectionBackoffMs(int retries) {
+ // visible for testing
+ static int binaryExponentialElectionBackoffMs(int backoffMaxMs, int retries, Random random) {
Review Comment:
Thanks for your explanation!
--
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-18345] Prevent deadlocked elections [kafka]
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on code in PR #19658:
URL: https://github.com/apache/kafka/pull/19658#discussion_r2082461965
##########
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java:
##########
@@ -1048,14 +1053,18 @@ private void maybeHandleElectionLoss(NomineeState state, long currentTimeMs) {
}
}
- private int binaryExponentialElectionBackoffMs(int retries) {
+ // visible for testing
+ static int binaryExponentialElectionBackoffMs(int backoffMaxMs, int retries, Random random) {
if (retries <= 0) {
throw new IllegalArgumentException("Retries " + retries + " should be larger than zero");
}
- // upper limit exponential co-efficients at 20 to avoid overflow
+ // Takes minimum of the following:
+ // 1. exponential backoff calculation (maxes out at 102.4 seconds)
+ // 2. configurable electionBackoffMaxMs + jitter
+ // The jitter is added to prevent deadlock of elections.
Review Comment:
I don't think deadlock is the correct term. The actors/replicas are not dead or blocked. I think that livelock is a more accurate description of the issue.
The replicas are changing (i.e. holding election) but they are changing in a way that they can't make progress (i.e. elect a 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-18345] Prevent deadlocked elections [kafka]
Posted by "ahuang98 (via GitHub)" <gi...@apache.org>.
ahuang98 commented on code in PR #19658:
URL: https://github.com/apache/kafka/pull/19658#discussion_r2082550000
##########
raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientTest.java:
##########
@@ -4569,6 +4574,55 @@ public void testObserverFetchWithNoLocalId(boolean withKip853Rpc) throws Excepti
assertEquals(3, context.log.lastFetchedEpoch());
}
+ @Test
+ public void testExponentialElectionBackoffMs() {
Review Comment:
I chose not to thinking it would overcomplicate the if else conditions based on the number of retries. Let me split this up into two tests actually.
--
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-18345] Prevent deadlocked elections [kafka]
Posted by "TaiJuWu (via GitHub)" <gi...@apache.org>.
TaiJuWu commented on code in PR #19658:
URL: https://github.com/apache/kafka/pull/19658#discussion_r2080885590
##########
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java:
##########
@@ -1048,14 +1053,18 @@ private void maybeHandleElectionLoss(NomineeState state, long currentTimeMs) {
}
}
- private int binaryExponentialElectionBackoffMs(int retries) {
+ // visible for testing
+ static int binaryExponentialElectionBackoffMs(int backoffMaxMs, int retries, Random random) {
Review Comment:
Could we use https://github.com/apache/kafka/blob/e1b2adea07cebaac6d1138ec18778c575aef57f4/clients/src/main/java/org/apache/kafka/common/utils/ExponentialBackoff.java#L34 in the module?
--
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-18345] Prevent deadlocked elections [kafka]
Posted by "ahuang98 (via GitHub)" <gi...@apache.org>.
ahuang98 commented on code in PR #19658:
URL: https://github.com/apache/kafka/pull/19658#discussion_r2082035897
##########
raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientTest.java:
##########
@@ -4569,6 +4574,55 @@ public void testObserverFetchWithNoLocalId(boolean withKip853Rpc) throws Excepti
assertEquals(3, context.log.lastFetchedEpoch());
}
+ @Test
+ public void testExponentialElectionBackoffMs() {
+ Random mockedRandom = Mockito.mock(Random.class);
+ int electionBackoffMaxMs = 1000;
+
+ // test the bound of the first call to random.nextInt
+ for (int retries = 1; retries < 11; retries++) {
+ binaryExponentialElectionBackoffMs(electionBackoffMaxMs, retries, mockedRandom);
+ ArgumentCaptor<Integer> nextIntCaptor = ArgumentCaptor.forClass(Integer.class);
+ Mockito.verify(mockedRandom).nextInt(Mockito.eq(1), nextIntCaptor.capture());
+ int actualBound = nextIntCaptor.getValue();
+ int expectedBound = (int) (2 * Math.pow(2, retries - 1));
+ assertEquals(expectedBound, actualBound, "Incorrect bound for retries=" + retries);
+ Mockito.clearInvocations(mockedRandom);
+ }
+ // after the 10th retry, the bound of the first call to random.nextInt will remain capped to
+ // (RETRY_BACKOFF_BASE_MS * 2 << 10)=2048 to prevent overflow
+ int firstNextIntMaxBound = 2048;
+ for (int retries = 11; retries < 13; retries++) {
+ binaryExponentialElectionBackoffMs(electionBackoffMaxMs, retries, mockedRandom);
+ ArgumentCaptor<Integer> nextIntCaptor = ArgumentCaptor.forClass(Integer.class);
+ Mockito.verify(mockedRandom).nextInt(Mockito.eq(1), nextIntCaptor.capture());
+ int actualBound = nextIntCaptor.getValue();
+ assertEquals(firstNextIntMaxBound, actualBound);
+ Mockito.clearInvocations(mockedRandom);
+ }
+
+ // test that the return value of the method is capped to QUORUM_ELECTION_BACKOFF_MAX_MS_CONFIG + jitter
+ int jitterMs = 50;
+ // any bound >= (1000 + jitter)/(RETRY_BACKOFF_BASE_MS)=21 will result in returning this cap
+ for (int firstNextInt : Arrays.asList(21, 1000, firstNextIntMaxBound)) {
+ Mockito.when(mockedRandom.nextInt(1, firstNextIntMaxBound)).thenReturn(firstNextInt);
+ Mockito.when(mockedRandom.nextInt(RETRY_BACKOFF_BASE_MS)).thenReturn(jitterMs);
+
+ int returnedBackoffMs = binaryExponentialElectionBackoffMs(electionBackoffMaxMs, 11, mockedRandom);
+
+ // verify nextInt was called on both expected bounds
+ ArgumentCaptor<Integer> nextIntCaptor = ArgumentCaptor.forClass(Integer.class);
+ Mockito.verify(mockedRandom).nextInt(Mockito.eq(1), nextIntCaptor.capture());
+ Mockito.verify(mockedRandom).nextInt(nextIntCaptor.capture());
+ List<Integer> allCapturedBounds = nextIntCaptor.getAllValues();
+ assertEquals(firstNextIntMaxBound, allCapturedBounds.get(0));
+ assertEquals(RETRY_BACKOFF_BASE_MS, allCapturedBounds.get(1));
+
+ // finally verify the backoff returned is capped to electionBackoffMaxMs + jitterMs
+ assertEquals(electionBackoffMaxMs + jitterMs, returnedBackoffMs);
+ }
Review Comment:
this doesn't seem to be necessary because mockedRandom is declared within the test and this is the end of the 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-18345] Prevent deadlocked elections [kafka]
Posted by "ahuang98 (via GitHub)" <gi...@apache.org>.
ahuang98 commented on code in PR #19658:
URL: https://github.com/apache/kafka/pull/19658#discussion_r2082083752
##########
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java:
##########
@@ -1048,14 +1053,18 @@ private void maybeHandleElectionLoss(NomineeState state, long currentTimeMs) {
}
}
- private int binaryExponentialElectionBackoffMs(int retries) {
+ // visible for testing
+ static int binaryExponentialElectionBackoffMs(int backoffMaxMs, int retries, Random random) {
Review Comment:
nice find! looks like we could use that to accomplish something very similar. unfortunately it looks like that class as the same issue where at some point it becomes statistically likely that the backoff returned is capped to a static value. this is probably fine for most cases / a more widespread change than I'm comfortable making in this PR.
btw, this issue should only affect the pre-KIP-996 world. after this is merged and cherry-picked to older releases, I will open a new PR to remove this exponential backoff completely
##########
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java:
##########
@@ -1048,14 +1053,18 @@ private void maybeHandleElectionLoss(NomineeState state, long currentTimeMs) {
}
}
- private int binaryExponentialElectionBackoffMs(int retries) {
+ // visible for testing
+ static int binaryExponentialElectionBackoffMs(int backoffMaxMs, int retries, Random random) {
Review Comment:
nice find! looks like we could use that to accomplish something very similar. unfortunately it looks like that class has the same issue where at some point it becomes statistically likely that the backoff returned is capped to a static value. this is probably fine for most cases / a more widespread change than I'm comfortable making in this PR.
btw, this issue should only affect the pre-KIP-996 world. after this is merged and cherry-picked to older releases, I will open a new PR to remove this exponential backoff completely
--
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-18345] Prevent deadlocked elections [kafka]
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on code in PR #19658:
URL: https://github.com/apache/kafka/pull/19658#discussion_r2082461965
##########
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java:
##########
@@ -1048,14 +1053,18 @@ private void maybeHandleElectionLoss(NomineeState state, long currentTimeMs) {
}
}
- private int binaryExponentialElectionBackoffMs(int retries) {
+ // visible for testing
+ static int binaryExponentialElectionBackoffMs(int backoffMaxMs, int retries, Random random) {
if (retries <= 0) {
throw new IllegalArgumentException("Retries " + retries + " should be larger than zero");
}
- // upper limit exponential co-efficients at 20 to avoid overflow
+ // Takes minimum of the following:
+ // 1. exponential backoff calculation (maxes out at 102.4 seconds)
+ // 2. configurable electionBackoffMaxMs + jitter
+ // The jitter is added to prevent deadlock of elections.
Review Comment:
I don't think deadlock is the correct term. The actors/replicas are not dead or blocked. I think that livelock is a more accurate description of the issue.
The replicas are changing (i.e. holding election) but they are changing in a way that they can't make process(i.e. elect a leader).
##########
raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientTest.java:
##########
@@ -4569,6 +4574,55 @@ public void testObserverFetchWithNoLocalId(boolean withKip853Rpc) throws Excepti
assertEquals(3, context.log.lastFetchedEpoch());
}
+ @Test
+ public void testExponentialElectionBackoffMs() {
Review Comment:
Did you consider having 3 parametrized tests? On each test invocation you can create a new mocked `Random`. This should allow you to remove `clearInvocations` and be sure that your mock objects are behaving correctly.
##########
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java:
##########
@@ -1048,14 +1053,18 @@ private void maybeHandleElectionLoss(NomineeState state, long currentTimeMs) {
}
}
- private int binaryExponentialElectionBackoffMs(int retries) {
+ // visible for testing
+ static int binaryExponentialElectionBackoffMs(int backoffMaxMs, int retries, Random random) {
Review Comment:
How about moving this static method to `RaftUtil` and make it public? If you do this you can also move the test to `RaftUtilTest`.
--
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-18345] Prevent deadlocked elections [kafka]
Posted by "ahuang98 (via GitHub)" <gi...@apache.org>.
ahuang98 commented on code in PR #19658:
URL: https://github.com/apache/kafka/pull/19658#discussion_r2082083752
##########
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java:
##########
@@ -1048,14 +1053,18 @@ private void maybeHandleElectionLoss(NomineeState state, long currentTimeMs) {
}
}
- private int binaryExponentialElectionBackoffMs(int retries) {
+ // visible for testing
+ static int binaryExponentialElectionBackoffMs(int backoffMaxMs, int retries, Random random) {
Review Comment:
nice find! looks like we could use that to accomplish something very similar. unfortunately it looks like that class has the same issue where at some point it becomes statistically likely that the backoff returned is capped to a static value. this is probably fine for most cases and fixing this would be a more widespread change that I'm not comfortable making in this PR.
btw, this issue should only affect the pre-KIP-996 world. after this is merged and cherry-picked to older releases, I will open a new PR to remove this exponential backoff completely
--
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-18345] Prevent deadlocked elections [kafka]
Posted by "Yunyung (via GitHub)" <gi...@apache.org>.
Yunyung commented on code in PR #19658:
URL: https://github.com/apache/kafka/pull/19658#discussion_r2082456060
##########
raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientTest.java:
##########
@@ -4569,6 +4574,55 @@ public void testObserverFetchWithNoLocalId(boolean withKip853Rpc) throws Excepti
assertEquals(3, context.log.lastFetchedEpoch());
}
+ @Test
+ public void testExponentialElectionBackoffMs() {
+ Random mockedRandom = Mockito.mock(Random.class);
+ int electionBackoffMaxMs = 1000;
+
+ // test the bound of the first call to random.nextInt
+ for (int retries = 1; retries < 11; retries++) {
+ binaryExponentialElectionBackoffMs(electionBackoffMaxMs, retries, mockedRandom);
+ ArgumentCaptor<Integer> nextIntCaptor = ArgumentCaptor.forClass(Integer.class);
+ Mockito.verify(mockedRandom).nextInt(Mockito.eq(1), nextIntCaptor.capture());
+ int actualBound = nextIntCaptor.getValue();
+ int expectedBound = (int) (2 * Math.pow(2, retries - 1));
+ assertEquals(expectedBound, actualBound, "Incorrect bound for retries=" + retries);
+ Mockito.clearInvocations(mockedRandom);
+ }
+ // after the 10th retry, the bound of the first call to random.nextInt will remain capped to
+ // (RETRY_BACKOFF_BASE_MS * 2 << 10)=2048 to prevent overflow
+ int firstNextIntMaxBound = 2048;
+ for (int retries = 11; retries < 13; retries++) {
+ binaryExponentialElectionBackoffMs(electionBackoffMaxMs, retries, mockedRandom);
+ ArgumentCaptor<Integer> nextIntCaptor = ArgumentCaptor.forClass(Integer.class);
+ Mockito.verify(mockedRandom).nextInt(Mockito.eq(1), nextIntCaptor.capture());
+ int actualBound = nextIntCaptor.getValue();
+ assertEquals(firstNextIntMaxBound, actualBound);
+ Mockito.clearInvocations(mockedRandom);
+ }
+
+ // test that the return value of the method is capped to QUORUM_ELECTION_BACKOFF_MAX_MS_CONFIG + jitter
+ int jitterMs = 50;
+ // any bound >= (1000 + jitter)/(RETRY_BACKOFF_BASE_MS)=21 will result in returning this cap
+ for (int firstNextInt : Arrays.asList(21, 1000, firstNextIntMaxBound)) {
+ Mockito.when(mockedRandom.nextInt(1, firstNextIntMaxBound)).thenReturn(firstNextInt);
+ Mockito.when(mockedRandom.nextInt(RETRY_BACKOFF_BASE_MS)).thenReturn(jitterMs);
+
+ int returnedBackoffMs = binaryExponentialElectionBackoffMs(electionBackoffMaxMs, 11, mockedRandom);
+
+ // verify nextInt was called on both expected bounds
+ ArgumentCaptor<Integer> nextIntCaptor = ArgumentCaptor.forClass(Integer.class);
+ Mockito.verify(mockedRandom).nextInt(Mockito.eq(1), nextIntCaptor.capture());
+ Mockito.verify(mockedRandom).nextInt(nextIntCaptor.capture());
+ List<Integer> allCapturedBounds = nextIntCaptor.getAllValues();
+ assertEquals(firstNextIntMaxBound, allCapturedBounds.get(0));
+ assertEquals(RETRY_BACKOFF_BASE_MS, allCapturedBounds.get(1));
+
+ // finally verify the backoff returned is capped to electionBackoffMaxMs + jitterMs
+ assertEquals(electionBackoffMaxMs + jitterMs, returnedBackoffMs);
+ }
Review Comment:
Pardon me, is this correct? The test fails on my side because the mock invocations accumulate inside the loop.
--
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-18345] Prevent deadlocked elections [kafka]
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on code in PR #19658:
URL: https://github.com/apache/kafka/pull/19658#discussion_r2085061929
##########
raft/src/main/java/org/apache/kafka/raft/RaftUtil.java:
##########
@@ -755,4 +756,18 @@ static boolean hasValidTopicPartition(DescribeQuorumRequestData data, TopicParti
data.topics().get(0).partitions().size() == 1 &&
data.topics().get(0).partitions().get(0).partitionIndex() == topicPartition.partition();
}
+
+ static int binaryExponentialElectionBackoffMs(int backoffMaxMs, int backoffBaseMs, int retries, Random random) {
+ if (retries <= 0) {
+ throw new IllegalArgumentException("Retries " + retries + " should be larger than zero");
+ }
+ // Takes minimum of the following:
+ // 1. exponential backoff calculation (maxes out at 102.4 seconds)
+ // 2. configurable electionBackoffMaxMs + jitter
+ // The jitter is added to prevent deadlock of elections.
Review Comment:
I mentioned this in another comment. This is not a deadlock.
--
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-18345; Prevent livelocked elections [kafka]
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio merged PR #19658:
URL: https://github.com/apache/kafka/pull/19658
--
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