You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "showuon (via GitHub)" <gi...@apache.org> on 2023/09/23 09:58:06 UTC
[GitHub] [kafka] showuon opened a new pull request, #14428: KAFKA-15489: resign leadership when no fetch from majority voters
showuon opened a new pull request, #14428:
URL: https://github.com/apache/kafka/pull/14428
In [KIP-595](https://cwiki.apache.org/confluence/display/KAFKA/KIP-595%3A+A+Raft+Protocol+for+the+Metadata+Quorum#KIP595:ARaftProtocolfortheMetadataQuorum-LeaderProgressTimeout), we expect to piggy-back on the `quorum.fetch.timeout.ms` config, and if the leader did not receive Fetch requests from a majority of the quorum for that amount of time, it would begin a new election, to resolve the network partition in the quorum. But we missed this implementation in current KRaft. Fixed it in this PR.
This PR did:
1. Added a fetchTimer with fetchTimeout in `LeaderState`, and check if expired each time when leader poll. If expired, resigning the leadership and start a new election.
2. Added `fetchedVoters` in `LeaderState`, and update the value each time received a fetchRequest, and clear it and reset fetchTimer if the majority of fetchRequest received.
3. Added 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-15489: resign leadership when no fetch from majority voters [kafka]
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on code in PR #14428:
URL: https://github.com/apache/kafka/pull/14428#discussion_r1406806920
##########
raft/src/main/java/org/apache/kafka/raft/LeaderState.java:
##########
@@ -79,6 +86,39 @@ protected LeaderState(
this.grantingVoters = Collections.unmodifiableSet(new HashSet<>(grantingVoters));
this.log = logContext.logger(LeaderState.class);
this.accumulator = Objects.requireNonNull(accumulator, "accumulator must be non-null");
+ // use the 1.5x fetch timeout to tolerate some network transition time or other IO time.
+ this.fetchTimeoutMs = (int) (fetchTimeoutMs * 1.5);
+ this.fetchTimer = time.timer(fetchTimeoutMs);
+ }
+
+ // Check if the fetchTimer is expired because we didn't receive a valid fetch/fetchSnapshot request from the majority of
+ // the voters within fetch timeout.
+ public boolean hasMajorityFollowerFetchExpired(long currentTimeMs) {
+ fetchTimer.update(currentTimeMs);
+ boolean isExpired = fetchTimer.isExpired();
+ if (isExpired) {
+ log.info("Did not receive fetch request from the majority of the voters within {}ms. Current fetched voters are {}.",
+ fetchTimeoutMs, fetchedVoters);
+ }
+ return isExpired;
+ }
+
+ // Reset the fetch timer if we've received fetch/fetchSnapshot request from the majority of the voter
+ public void maybeResetMajorityFollowerFetchTimer(int id, long currentTimeMs) {
+ updateFetchedVoters(id);
+ // The majority number of the voters excluding the leader. Ex: 3 voters, the value will be 1
+ int majority = voterStates.size() / 2;
+ if (fetchedVoters.size() >= majority) {
+ fetchedVoters.clear();
+ fetchTimer.update(currentTimeMs);
+ fetchTimer.reset(fetchTimeoutMs);
+ }
+ }
+
+ private void updateFetchedVoters(int id) {
+ if (isVoter(id)) {
Review Comment:
We should be defensive against this getting called with the local replica. Let's throw an `IllegalArgumentException`, if `id` is equal to the `localId`.
--
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-15489: resign leadership when no fetch from majority voters [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #14428:
URL: https://github.com/apache/kafka/pull/14428#discussion_r1343475329
##########
raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientTest.java:
##########
@@ -485,6 +485,49 @@ public void testHandleBeginQuorumEpochAfterUserInitiatedResign() throws Exceptio
context.listener.currentLeaderAndEpoch());
}
+ @Test
+ public void testLeaderShouldResignLeadershipIfNotGetFetchRequestFromMajorityVoters() throws Exception {
Review Comment:
Added a test in `KafkaRaftClientSnapshotTest`.
> Can we also tests the opposite. That the leader doesn't resign if the majority of the replicas (including the leader) have fetch in the last fetchTimeoutMs?
I didn't follow you. I've verified:
```
- 1/2 fetch time
leadership not get reassigned
- fetch from one voter
--- timer reset ---
- 1/2 fetch time
leadership not get reassigned
- fetch from another voter
--- timer reset ---
- 1/2 fetch time
leadership not get reassigned
- fetch from the observer
- 1/2 fetch time
--- expired ---
leadership should get reassigned
```
I think I've verified what you want. Let me know if I need to add other things. Thanks.
--
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-15489: resign leadership when no fetch from majority voters [kafka]
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on PR #14428:
URL: https://github.com/apache/kafka/pull/14428#issuecomment-1828499581
Excuse the delays @showuon . I'll review this today and this week!
--
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
[GitHub] [kafka] satishd commented on pull request #14428: KAFKA-15489: resign leadership when no fetch from majority voters
Posted by "satishd (via GitHub)" <gi...@apache.org>.
satishd commented on PR #14428:
URL: https://github.com/apache/kafka/pull/14428#issuecomment-1732284491
cc @mumrah
--
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-15489: resign leadership when no fetch from majority voters [kafka]
Posted by "ahuang98 (via GitHub)" <gi...@apache.org>.
ahuang98 commented on code in PR #14428:
URL: https://github.com/apache/kafka/pull/14428#discussion_r1350814062
##########
raft/src/main/java/org/apache/kafka/raft/LeaderState.java:
##########
@@ -79,6 +86,38 @@ protected LeaderState(
this.grantingVoters = Collections.unmodifiableSet(new HashSet<>(grantingVoters));
this.log = logContext.logger(LeaderState.class);
this.accumulator = Objects.requireNonNull(accumulator, "accumulator must be non-null");
+ this.fetchTimeoutMs = fetchTimeoutMs;
+ this.fetchTimer = time.timer(fetchTimeoutMs);
+ }
+
+ // Check if the fetchTimer is expired because we didn't receive fetch/fetchSnapshot request from the majority of
Review Comment:
nit:
```suggestion
// Check if the fetchTimer is expired because we didn't receive a valid fetch/fetchSnapshot request from the majority of
```
same for the next method's description
--
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-15489: resign leadership when no fetch from majority voters [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on PR #14428:
URL: https://github.com/apache/kafka/pull/14428#issuecomment-1829267766
> Hi @showuon ,
>
> Thanks for the changes. They look good to me in general. One potential issue with this implementation is that the leader doesn't check that the fetching voters are making progress.
>
> Just because the leader returned a successful response to FETCH and FETCH_SNAPSHOT doesn't mean that the followers were able to handle the response correctly.
>
> For example, imagine the case where the log end offset (LEO) is at 1000 and all of the followers are continuously fetching at offset 0 without ever increasing their fetch offset. This can happen if the followers encounter an error when processing the FETCH or FETCH_SNAPSHOT response.
>
> In this scenario the leader will never be able to increase the HWM. I think that this scenario is specific to KRaft and doesn't exists in Raft because KRaft is pull vs Raft which is push.
>
> What do you think? Do you agree? If so should we address this issue in this PR or create an issue for this and fix it in a future PR?
Good catch! Yes, that's indeed a potential problem. This PR has been pending for a long time, let's focus on the current issue in this PR first. I've filed: [KAFKA-15911](https://issues.apache.org/jira/browse/KAFKA-15911) for the potential issue.
--
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-15489: resign leadership when no fetch from majority voters [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #14428:
URL: https://github.com/apache/kafka/pull/14428#discussion_r1407336829
##########
raft/src/test/java/org/apache/kafka/raft/LeaderStateTest.java:
##########
@@ -447,6 +452,38 @@ public void testDescribeQuorumWithObservers() {
observerState);
}
+ @Test
+ public void testMajorityFollowerFetchTimeoutExpiration() {
+ int node1 = 1;
+ int node2 = 2;
+ int node3 = 3;
+ int node4 = 4;
+ int observer5 = 5;
+ LeaderState<?> state = newLeaderState(mkSet(localId, node1, node2, node3, node4), 0L);
+ assertFalse(state.hasMajorityFollowerFetchExpired(time.milliseconds()));
+ int resignLeadershipTimeout = (int) (fetchTimeoutMs * 1.5);
Review Comment:
Agree! If we need to update the `1.5` factor, we can change in one place. Updated.
--
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-15489: resign leadership when no fetch from majority voters [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #14428:
URL: https://github.com/apache/kafka/pull/14428#discussion_r1407334210
##########
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java:
##########
@@ -1340,6 +1341,9 @@ private FetchSnapshotResponseData handleFetchSnapshotRequest(
UnalignedRecords records = snapshot.slice(partitionSnapshot.position(), Math.min(data.maxBytes(), maxSnapshotSize));
+ Optional<LeaderState<T>> state = quorum.maybeLeaderState();
+ state.ifPresent(s -> s.maybeResetMajorityFollowerFetchTimer(data.replicaId(), currentTimeMs));
+
Review Comment:
Good point. Updated.
--
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-15489: resign leadership when no fetch from majority voters [kafka]
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on code in PR #14428:
URL: https://github.com/apache/kafka/pull/14428#discussion_r1406995366
##########
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java:
##########
@@ -1990,7 +1995,7 @@ private long pollLeader(long currentTimeMs) {
LeaderState<T> state = quorum.leaderStateOrThrow();
maybeFireLeaderChange(state);
- if (shutdown.get() != null || state.isResignRequested()) {
+ if (shutdown.get() != null || state.isResignRequested() || state.hasMajorityFollowerFetchExpired(currentTimeMs)) {
Review Comment:
In line [2014](https://github.com/apache/kafka/pull/14428/files#diff-1da15c51e641ea46ea5c86201ab8f21cfee9e7c575102a39c7bae0d5ffd7de39R2014) KRaft calculates the next time it should poll the leader.
```java
return Math.min(timeUntilFlush, timeUntilSend);
```
If you extend this to something like below, it should wake up the leader when the timer has expired.
```java
return Math.min(timeUntilFlush, timeUntilSend, state.timeUntilCheckQuorumExpires(currentTimeMs));
```
The other option is to extend `hasMajorityFollowerFetchExpired` to include this information by using `Timer::remainingMs` instead of `isExpired`.
--
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-15489: resign leadership when no fetch from majority voters [kafka]
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on code in PR #14428:
URL: https://github.com/apache/kafka/pull/14428#discussion_r1408235310
##########
raft/src/main/java/org/apache/kafka/raft/LeaderState.java:
##########
@@ -79,6 +88,55 @@ protected LeaderState(
this.grantingVoters = Collections.unmodifiableSet(new HashSet<>(grantingVoters));
this.log = logContext.logger(LeaderState.class);
this.accumulator = Objects.requireNonNull(accumulator, "accumulator must be non-null");
+ // use the 1.5x of fetch timeout to tolerate some network transition time or other IO time.
+ this.checkQuorumTimeoutMs = (int) (fetchTimeoutMs * CHECK_QUORUM_TIMEOUT_FACTOR);
+ this.checkQuorumTimer = time.timer(checkQuorumTimeoutMs);
+ }
+
+ /**
+ * Get the remaining time in milliseconds until the checkQuorumTimer expires.
+ * This will happen if we didn't receive a valid fetch/fetchSnapshot request from the majority of the voters within checkQuorumTimeoutMs.
+ *
+ * @param currentTimeMs the current timestamp in millisecond
+ * @return the remainingMs before the checkQuorumTimer expired
+ */
+ public long timeUntilCheckQuorumExpires(long currentTimeMs) {
+ checkQuorumTimer.update(currentTimeMs);
+ long remainingMs = checkQuorumTimer.remainingMs();
+ if (remainingMs == 0) {
+ log.info(
+ "Did not receive fetch request from the majority of the voters within {}ms. Current fetched voters are {}.",
+ checkQuorumTimeoutMs,
+ fetchedVoters);
Review Comment:
How about this indentation:
```java
log.info(
"Did not receive fetch request from the majority of the voters within {}ms. Current fetched voters are {}.",
checkQuorumTimeoutMs,
fetchedVoters
);
```
--
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
[GitHub] [kafka] jsancio commented on a diff in pull request #14428: KAFKA-15489: resign leadership when no fetch from majority voters
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on code in PR #14428:
URL: https://github.com/apache/kafka/pull/14428#discussion_r1341771602
##########
raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientTest.java:
##########
@@ -485,6 +485,49 @@ public void testHandleBeginQuorumEpochAfterUserInitiatedResign() throws Exceptio
context.listener.currentLeaderAndEpoch());
}
+ @Test
+ public void testLeaderShouldResignLeadershipIfNotGetFetchRequestFromMajorityVoters() throws Exception {
Review Comment:
Can we also tests the opposite. That the leader doesn't resign if the majority of the replicas (including the leader) have fetch in the last `fetchTimeoutMs`?
Can we also add test(s) under `KafkaRaftClientSnapshotTest` that show that the leader also considers `FETCH_SNAPSHOT` requests for determining network connectivity?
##########
raft/src/main/java/org/apache/kafka/raft/LeaderState.java:
##########
@@ -55,6 +57,11 @@ public class LeaderState<T> implements EpochState {
private final Map<Integer, ReplicaState> observerStates = new HashMap<>();
private final Logger log;
private final BatchAccumulator<T> accumulator;
+ private final Set<Integer> fetchedVoters = new HashSet<>();
+ private final Timer fetchTimer;
+ private final int fetchTimeoutMs;
+ // The majority number of the voters excluding the leader. Ex: 3 voters, the value will be 1
+ private final int majority;
Review Comment:
Let's not cache this value. Let's always compute this from the size of `voterStates`. We can add a function for this.
I am hoping to add support for dynamic voters soon and this is one less thing that I would have to change :smile: .
##########
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java:
##########
@@ -965,6 +965,10 @@ private CompletableFuture<FetchResponseData> handleFetchRequest(
}
int replicaId = FetchRequest.replicaId(request);
+
+ Optional<LeaderState<T>> state = quorum.maybeLeaderState();
+ state.ifPresent(s -> s.maybeResetMajorityFollowerFetchTimeout(replicaId, currentTimeMs));
Review Comment:
I am thinking that the leader should only update the timer on successful FETCH and FETCH_SNAPSHOT responses. If the majority of the replicas are not successfully replicating the log, the leader will not be able commit write operations (`RaftClient::scheduleAppend`) from the state machine.
##########
raft/src/main/java/org/apache/kafka/raft/LeaderState.java:
##########
@@ -66,7 +73,9 @@ protected LeaderState(
Set<Integer> voters,
Set<Integer> grantingVoters,
BatchAccumulator<T> accumulator,
- LogContext logContext
+ LogContext logContext,
+ Time time,
+ int fetchTimeoutMs
Review Comment:
Let's re-order this so that it better matches `CandidateState` and `FollowerState`. How about this order for the parameters:
```java
protected LeaderState(
Time time,
int localId,
int epoch,
long epochStartOffset,
Set<Integer> voters,
Set<Integer> grantingVoters,
BatchAccumulator<T> accumulator,
int fetchTimeoutMs,
LogContext logContext
)
```
##########
raft/src/main/java/org/apache/kafka/raft/LeaderState.java:
##########
@@ -76,9 +85,37 @@ protected LeaderState(
boolean hasAcknowledgedLeader = voterId == localId;
this.voterStates.put(voterId, new ReplicaState(voterId, hasAcknowledgedLeader));
}
+ this.majority = voters.size() / 2;
this.grantingVoters = Collections.unmodifiableSet(new HashSet<>(grantingVoters));
this.log = logContext.logger(LeaderState.class);
this.accumulator = Objects.requireNonNull(accumulator, "accumulator must be non-null");
+ this.fetchTimeoutMs = fetchTimeoutMs;
+ this.fetchTimer = time.timer(fetchTimeoutMs);
+ }
+
+ public boolean hasMajorityFollowerFetchTimeoutExpired(long currentTimeMs) {
+ fetchTimer.update(currentTimeMs);
+ boolean isExpired = fetchTimer.isExpired();
+ if (isExpired) {
+ log.info("Did not receive fetch request from the majority of the voters within {}ms. Current fetched voters are {}.",
+ fetchTimeoutMs, fetchedVoters);
+ }
+ return isExpired;
+ }
+
+ public void maybeResetMajorityFollowerFetchTimeout(int id, long currentTimeMs) {
+ updateFetchedVoters(id);
+ if (fetchedVoters.size() >= majority) {
+ fetchedVoters.clear();
+ fetchTimer.update(currentTimeMs);
+ fetchTimer.reset(fetchTimeoutMs);
+ }
+ }
+
+ private void updateFetchedVoters(int id) {
+ if (isVoter(id)) {
+ fetchedVoters.add(id);
+ }
Review Comment:
I see.
I think that the invariant that we need to hold is that for any time span of `fetchTimeoutMs` the majority of the replicas have performed a successful `FETCH` and `FETCH_SNAPSHOT`. Note that `ReplicaState` already contains the `lastFetchTimestamp`.
The part that is not clear to me is when or how to wake up the leader for a `poll`. We need to update `KafkaRaftClient::pollLeader` so that the replicas' last fetch time is taken into account when blocking on the `messageQueue.poll`.
What do you think?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15489: resign leadership when no fetch from majority voters [kafka]
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on code in PR #14428:
URL: https://github.com/apache/kafka/pull/14428#discussion_r1406803592
##########
raft/src/main/java/org/apache/kafka/raft/LeaderState.java:
##########
@@ -79,6 +86,39 @@ protected LeaderState(
this.grantingVoters = Collections.unmodifiableSet(new HashSet<>(grantingVoters));
this.log = logContext.logger(LeaderState.class);
this.accumulator = Objects.requireNonNull(accumulator, "accumulator must be non-null");
+ // use the 1.5x fetch timeout to tolerate some network transition time or other IO time.
+ this.fetchTimeoutMs = (int) (fetchTimeoutMs * 1.5);
+ this.fetchTimer = time.timer(fetchTimeoutMs);
+ }
+
+ // Check if the fetchTimer is expired because we didn't receive a valid fetch/fetchSnapshot request from the majority of
+ // the voters within fetch timeout.
+ public boolean hasMajorityFollowerFetchExpired(long currentTimeMs) {
Review Comment:
See my other comment about check quorum and Alyssa's pre-vote KIP. If you agree, maybe we should call this `hasCheckQuorumFailed`.
--
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-15489: resign leadership when no fetch from majority voters [kafka]
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on code in PR #14428:
URL: https://github.com/apache/kafka/pull/14428#discussion_r1406995366
##########
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java:
##########
@@ -1990,7 +1995,7 @@ private long pollLeader(long currentTimeMs) {
LeaderState<T> state = quorum.leaderStateOrThrow();
maybeFireLeaderChange(state);
- if (shutdown.get() != null || state.isResignRequested()) {
+ if (shutdown.get() != null || state.isResignRequested() || state.hasMajorityFollowerFetchExpired(currentTimeMs)) {
Review Comment:
In line [2014](https://github.com/apache/kafka/pull/14428/files#diff-1da15c51e641ea46ea5c86201ab8f21cfee9e7c575102a39c7bae0d5ffd7de39R2014) KRaft calculates the next time it should poll the leader.
```java
return Math.min(timeUntilFlush, timeUntilSend);
```
If you extend this to something like below, it should wake up the leader when the timer has expired.
```java
return Math.min(timeUntilFlush, timeUntilSend, state.checkQuorumExpiry());
```
--
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-15489: resign leadership when no fetch from majority voters [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on PR #14428:
URL: https://github.com/apache/kafka/pull/14428#issuecomment-1798236911
@jsancio @ahuang98 , do you have any other comments? I'd like to include this fix into v3.5.2 if possible. Thanks.
--
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-15489: resign leadership when no fetch from majority voters [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #14428:
URL: https://github.com/apache/kafka/pull/14428#discussion_r1343475329
##########
raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientTest.java:
##########
@@ -485,6 +485,49 @@ public void testHandleBeginQuorumEpochAfterUserInitiatedResign() throws Exceptio
context.listener.currentLeaderAndEpoch());
}
+ @Test
+ public void testLeaderShouldResignLeadershipIfNotGetFetchRequestFromMajorityVoters() throws Exception {
Review Comment:
Added a test in `KafkaRaftClientSnapshotTest`.
> Can we also tests the opposite. That the leader doesn't resign if the majority of the replicas (including the leader) have fetch in the last fetchTimeoutMs?
I didn't follow you. I've verified:
```
A 3 controller cluster
- 1/2 fetch time
leadership not get reassigned
- fetch from one voter
--- timer reset ---
- 1/2 fetch time
leadership not get reassigned
- fetch from another voter
--- timer reset ---
- 1/2 fetch time
leadership not get reassigned
- fetch from the observer
- 1/2 fetch time
--- expired ---
leadership should get reassigned
```
I think I've verified what you want. Let me know if I need to add other things. Thanks.
--
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-15489: resign leadership when no fetch from majority voters [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #14428:
URL: https://github.com/apache/kafka/pull/14428#discussion_r1343475329
##########
raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientTest.java:
##########
@@ -485,6 +485,49 @@ public void testHandleBeginQuorumEpochAfterUserInitiatedResign() throws Exceptio
context.listener.currentLeaderAndEpoch());
}
+ @Test
+ public void testLeaderShouldResignLeadershipIfNotGetFetchRequestFromMajorityVoters() throws Exception {
Review Comment:
Added a test in `KafkaRaftClientSnapshotTest`.
> Can we also tests the opposite. That the leader doesn't resign if the majority of the replicas (including the leader) have fetch in the last fetchTimeoutMs?
I didn't follow you. I've verified:
```
3 controller cluster
- 1/2 fetch time
leadership not get reassigned
- fetch from one voter
--- timer reset ---
- 1/2 fetch time
leadership not get reassigned
- fetch from another voter
--- timer reset ---
- 1/2 fetch time
leadership not get reassigned
- fetch from the observer
- 1/2 fetch time
--- expired ---
leadership should get reassigned
```
I think I've verified what you want. Let me know if I need to add other things. Thanks.
--
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
[GitHub] [kafka] ahuang98 commented on a diff in pull request #14428: KAFKA-15489: resign leadership when no fetch from majority voters
Posted by "ahuang98 (via GitHub)" <gi...@apache.org>.
ahuang98 commented on code in PR #14428:
URL: https://github.com/apache/kafka/pull/14428#discussion_r1337653712
##########
raft/src/main/java/org/apache/kafka/raft/LeaderState.java:
##########
@@ -76,9 +85,37 @@ protected LeaderState(
boolean hasAcknowledgedLeader = voterId == localId;
this.voterStates.put(voterId, new ReplicaState(voterId, hasAcknowledgedLeader));
}
+ this.majority = voters.size() / 2;
this.grantingVoters = Collections.unmodifiableSet(new HashSet<>(grantingVoters));
this.log = logContext.logger(LeaderState.class);
this.accumulator = Objects.requireNonNull(accumulator, "accumulator must be non-null");
+ this.fetchTimeoutMs = fetchTimeoutMs;
+ this.fetchTimer = time.timer(fetchTimeoutMs);
+ }
+
+ public boolean hasMajorityFollowerFetchTimeoutExpired(long currentTimeMs) {
Review Comment:
nit: these are pretty lengthy and arguably unintuitive method names, could we add a short comment on what each method does?
--
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-15489: resign leadership when no fetch from majority voters [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #14428:
URL: https://github.com/apache/kafka/pull/14428#discussion_r1407333526
##########
raft/src/main/java/org/apache/kafka/raft/LeaderState.java:
##########
@@ -79,6 +86,39 @@ protected LeaderState(
this.grantingVoters = Collections.unmodifiableSet(new HashSet<>(grantingVoters));
this.log = logContext.logger(LeaderState.class);
this.accumulator = Objects.requireNonNull(accumulator, "accumulator must be non-null");
+ // use the 1.5x fetch timeout to tolerate some network transition time or other IO time.
+ this.fetchTimeoutMs = (int) (fetchTimeoutMs * 1.5);
+ this.fetchTimer = time.timer(fetchTimeoutMs);
Review Comment:
Good suggestion. Updated.
--
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-15489: resign leadership when no fetch from majority voters [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #14428:
URL: https://github.com/apache/kafka/pull/14428#discussion_r1342665261
##########
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java:
##########
@@ -965,6 +965,10 @@ private CompletableFuture<FetchResponseData> handleFetchRequest(
}
int replicaId = FetchRequest.replicaId(request);
+
+ Optional<LeaderState<T>> state = quorum.maybeLeaderState();
+ state.ifPresent(s -> s.maybeResetMajorityFollowerFetchTimeout(replicaId, currentTimeMs));
Review Comment:
Agree! Now, we update the timer when we update the lastFetchTimestamp in replicaState.
--
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-15489: resign leadership when no fetch from majority voters [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #14428:
URL: https://github.com/apache/kafka/pull/14428#discussion_r1407337495
##########
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java:
##########
@@ -1990,7 +1995,7 @@ private long pollLeader(long currentTimeMs) {
LeaderState<T> state = quorum.leaderStateOrThrow();
maybeFireLeaderChange(state);
- if (shutdown.get() != null || state.isResignRequested()) {
+ if (shutdown.get() != null || state.isResignRequested() || state.hasMajorityFollowerFetchExpired(currentTimeMs)) {
Review Comment:
Ah, good catch! Updated the `hasMajorityFollowerFetchExpired` to `timeUntilCheckQuorumExpires` 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-15489: resign leadership when no fetch from majority voters [kafka]
Posted by "ahuang98 (via GitHub)" <gi...@apache.org>.
ahuang98 commented on code in PR #14428:
URL: https://github.com/apache/kafka/pull/14428#discussion_r1343169152
##########
raft/src/main/java/org/apache/kafka/raft/LeaderState.java:
##########
@@ -76,9 +85,37 @@ protected LeaderState(
boolean hasAcknowledgedLeader = voterId == localId;
this.voterStates.put(voterId, new ReplicaState(voterId, hasAcknowledgedLeader));
}
+ this.majority = voters.size() / 2;
this.grantingVoters = Collections.unmodifiableSet(new HashSet<>(grantingVoters));
this.log = logContext.logger(LeaderState.class);
this.accumulator = Objects.requireNonNull(accumulator, "accumulator must be non-null");
+ this.fetchTimeoutMs = fetchTimeoutMs;
+ this.fetchTimer = time.timer(fetchTimeoutMs);
+ }
+
+ public boolean hasMajorityFollowerFetchTimeoutExpired(long currentTimeMs) {
Review Comment:
I was more suggesting that this method might benefit from a comment which describes behavior. I guess the info log explains it well enough
--
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-15489: resign leadership when no fetch from majority voters [kafka]
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio merged PR #14428:
URL: https://github.com/apache/kafka/pull/14428
--
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-15489: resign leadership when no fetch from majority voters [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on PR #14428:
URL: https://github.com/apache/kafka/pull/14428#issuecomment-1759306603
> I'm wondering if we may start causing leaders to resign when followers are slow/backlogged and make the situation worse? E.g. if we have multiple followers that need to catch up via a large fetch snapshot, they are unable to fetch again prior to the timeout expiring, and cause the current leader to resign. I don't believe this would be very disruptive but wanted to check folks had considered this/similar situation.
Yes, with 1.5x of timeout, this issue should be resolved. Also, if one follower is slow due to whatever reason, and doesn't fetch again within fetch timeout, it'll also start a new election. That's already the current implementation.
> I think we can also modify QUORUM_FETCH_TIMEOUT_MS_DOC to be slightly more explicit too (i.e. Maximum time a leader can go without receiving valid fetch or fetchsnapshot request from a majority of the quorum before resigning or something slightly different if we choose to use 1.5x)
Doc updated. I don't think we need to mention anything about 1.5x because that's the implementation detail.
--
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-15489: resign leadership when no fetch from majority voters [kafka]
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on code in PR #14428:
URL: https://github.com/apache/kafka/pull/14428#discussion_r1406768273
##########
raft/src/main/java/org/apache/kafka/raft/LeaderState.java:
##########
@@ -79,6 +86,39 @@ protected LeaderState(
this.grantingVoters = Collections.unmodifiableSet(new HashSet<>(grantingVoters));
this.log = logContext.logger(LeaderState.class);
this.accumulator = Objects.requireNonNull(accumulator, "accumulator must be non-null");
+ // use the 1.5x fetch timeout to tolerate some network transition time or other IO time.
+ this.fetchTimeoutMs = (int) (fetchTimeoutMs * 1.5);
+ this.fetchTimer = time.timer(fetchTimeoutMs);
Review Comment:
Since this is not set to the fetch timeout maybe we can call this `checkQuorumTimeoutMs` and `checkQuorumTimer`. I am suggesting these names because @ahuang98 uses "check quorum" in the pre-vote KIP: https://cwiki.apache.org/confluence/display/KAFKA/KIP-996%3A+Pre-Vote
##########
raft/src/main/java/org/apache/kafka/raft/LeaderState.java:
##########
@@ -79,6 +86,39 @@ protected LeaderState(
this.grantingVoters = Collections.unmodifiableSet(new HashSet<>(grantingVoters));
this.log = logContext.logger(LeaderState.class);
this.accumulator = Objects.requireNonNull(accumulator, "accumulator must be non-null");
+ // use the 1.5x fetch timeout to tolerate some network transition time or other IO time.
+ this.fetchTimeoutMs = (int) (fetchTimeoutMs * 1.5);
+ this.fetchTimer = time.timer(fetchTimeoutMs);
+ }
+
+ // Check if the fetchTimer is expired because we didn't receive a valid fetch/fetchSnapshot request from the majority of
+ // the voters within fetch timeout.
Review Comment:
Can we make this a Java doc comments. E.g.:
```java
/**
* Check if the fetchTimer is expired because we didn't receive a valid fetch/fetchSnapshot request from
* the majority of the voters within fetch timeout.
*
* @param ...
* @return ...
*/
```
##########
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java:
##########
@@ -1340,6 +1341,9 @@ private FetchSnapshotResponseData handleFetchSnapshotRequest(
UnalignedRecords records = snapshot.slice(partitionSnapshot.position(), Math.min(data.maxBytes(), maxSnapshotSize));
+ Optional<LeaderState<T>> state = quorum.maybeLeaderState();
+ state.ifPresent(s -> s.maybeResetMajorityFollowerFetchTimer(data.replicaId(), currentTimeMs));
+
Review Comment:
Since the check above `leaderValidation.isPresent()` is false, it means that this replica is guarantee to be the leader at this point in time. I prefer if we use `leaderStateOrThrow` instead of `maybeLeaderState` to make this clear to future readers.
##########
raft/src/main/java/org/apache/kafka/raft/LeaderState.java:
##########
@@ -79,6 +86,39 @@ protected LeaderState(
this.grantingVoters = Collections.unmodifiableSet(new HashSet<>(grantingVoters));
this.log = logContext.logger(LeaderState.class);
this.accumulator = Objects.requireNonNull(accumulator, "accumulator must be non-null");
+ // use the 1.5x fetch timeout to tolerate some network transition time or other IO time.
+ this.fetchTimeoutMs = (int) (fetchTimeoutMs * 1.5);
+ this.fetchTimer = time.timer(fetchTimeoutMs);
+ }
+
+ // Check if the fetchTimer is expired because we didn't receive a valid fetch/fetchSnapshot request from the majority of
+ // the voters within fetch timeout.
+ public boolean hasMajorityFollowerFetchExpired(long currentTimeMs) {
+ fetchTimer.update(currentTimeMs);
+ boolean isExpired = fetchTimer.isExpired();
+ if (isExpired) {
+ log.info("Did not receive fetch request from the majority of the voters within {}ms. Current fetched voters are {}.",
+ fetchTimeoutMs, fetchedVoters);
+ }
+ return isExpired;
+ }
+
+ // Reset the fetch timer if we've received fetch/fetchSnapshot request from the majority of the voter
+ public void maybeResetMajorityFollowerFetchTimer(int id, long currentTimeMs) {
+ updateFetchedVoters(id);
+ // The majority number of the voters excluding the leader. Ex: 3 voters, the value will be 1
+ int majority = voterStates.size() / 2;
+ if (fetchedVoters.size() >= majority) {
+ fetchedVoters.clear();
+ fetchTimer.update(currentTimeMs);
+ fetchTimer.reset(fetchTimeoutMs);
+ }
+ }
+
+ private void updateFetchedVoters(int id) {
+ if (isVoter(id)) {
Review Comment:
We should be defensive against this getting called with the local replica. Let's throw an `IllegalArgumentException` if `id` is equal to the `localId`.
##########
raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientTest.java:
##########
@@ -485,6 +485,49 @@ public void testHandleBeginQuorumEpochAfterUserInitiatedResign() throws Exceptio
context.listener.currentLeaderAndEpoch());
}
+ @Test
+ public void testLeaderShouldResignLeadershipIfNotGetFetchRequestFromMajorityVoters() throws Exception {
Review Comment:
Got it. This test covers all of the cases I was thinking about.
##########
raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientSnapshotTest.java:
##########
@@ -701,6 +702,97 @@ public void testFetchSnapshotRequestAsLeader() throws Exception {
assertEquals(memoryRecords.buffer(), ((UnalignedMemoryRecords) response.unalignedRecords()).buffer());
}
+ @Test
+ public void testLeaderShouldResignLeadershipIfNotGetFetchSnapshotRequestFromMajorityVoters() throws Exception {
+ int localId = 0;
+ int voter1 = 1;
+ int voter2 = 2;
+ int observerId3 = 3;
+ Set<Integer> voters = Utils.mkSet(localId, voter1, voter2);
+ OffsetAndEpoch snapshotId = new OffsetAndEpoch(1, 1);
+ List<String> records = Arrays.asList("foo", "bar");
+
+ RaftClientTestContext context = new RaftClientTestContext.Builder(localId, voters)
+ .appendToLog(snapshotId.epoch(), Arrays.asList("a"))
+ .build();
+
+ int resignLeadershipTimeout = (int) (context.fetchTimeoutMs * 1.5);
+ context.becomeLeader();
+ int epoch = context.currentEpoch();
+
+ FetchSnapshotRequestData voter1FetchSnapshotRequest = fetchSnapshotRequest(
+ context.clusterId.toString(),
+ voter1,
+ context.metadataPartition,
+ epoch,
+ snapshotId,
+ Integer.MAX_VALUE,
+ 0
+ );
+
+ FetchSnapshotRequestData voter2FetchSnapshotRequest = fetchSnapshotRequest(
+ context.clusterId.toString(),
+ voter2,
+ context.metadataPartition,
+ epoch,
+ snapshotId,
+ Integer.MAX_VALUE,
+ 0
+ );
+
+ FetchSnapshotRequestData observerFetchSnapshotRequest = fetchSnapshotRequest(
+ context.clusterId.toString(),
+ observerId3,
+ context.metadataPartition,
+ epoch,
+ snapshotId,
+ Integer.MAX_VALUE,
+ 0
+ );
+
+ context.advanceLocalLeaderHighWatermarkToLogEndOffset();
+ try (SnapshotWriter<String> snapshot = context.client.createSnapshot(snapshotId, 0).get()) {
+ assertEquals(snapshotId, snapshot.snapshotId());
+ snapshot.append(records);
+ snapshot.freeze();
+ }
+
+ // fetch timeout is not expired, the leader should not get resigned
+ context.time.sleep(resignLeadershipTimeout / 2);
+ context.client.poll();
+ assertFalse(context.client.quorum().isResigned());
+
+ // voter1 sends fetchSnapshotRequest, the fetch timer should be reset
+ context.deliverRequest(voter1FetchSnapshotRequest);
+ context.client.poll();
+ context.assertSentFetchSnapshotResponse(context.metadataPartition);
+
+ // Since the fetch timer is reset, the leader should not get resigned
+ context.time.sleep(resignLeadershipTimeout / 2);
+ context.client.poll();
+ assertFalse(context.client.quorum().isResigned());
+
+ // voter2 sends fetchSnapshotRequest, the fetch timer should be reset
+ context.deliverRequest(voter2FetchSnapshotRequest);
+ context.client.poll();
+ context.assertSentFetchSnapshotResponse(context.metadataPartition);
+
+ // Since the fetch timer is reset, the leader should not get resigned
+ context.time.sleep(resignLeadershipTimeout / 2);
+ context.client.poll();
+ assertFalse(context.client.quorum().isResigned());
+
+ // An observer sends fetchSnapshotRequest, but the fetch timer should not be reset.
+ context.deliverRequest(observerFetchSnapshotRequest);
+ context.client.poll();
+ context.assertSentFetchSnapshotResponse(context.metadataPartition);
+
+ // After this sleep, the fetch timeout should expire since we don't receive fetch request from the majority voters within fetchTimeoutMs
+ context.time.sleep(resignLeadershipTimeout / 2);
+ context.client.poll();
+ assertTrue(context.client.quorum().isResigned());
Review Comment:
Great test. Thank you!
##########
raft/src/main/java/org/apache/kafka/raft/LeaderState.java:
##########
@@ -79,6 +86,39 @@ protected LeaderState(
this.grantingVoters = Collections.unmodifiableSet(new HashSet<>(grantingVoters));
this.log = logContext.logger(LeaderState.class);
this.accumulator = Objects.requireNonNull(accumulator, "accumulator must be non-null");
+ // use the 1.5x fetch timeout to tolerate some network transition time or other IO time.
+ this.fetchTimeoutMs = (int) (fetchTimeoutMs * 1.5);
+ this.fetchTimer = time.timer(fetchTimeoutMs);
+ }
+
+ // Check if the fetchTimer is expired because we didn't receive a valid fetch/fetchSnapshot request from the majority of
+ // the voters within fetch timeout.
+ public boolean hasMajorityFollowerFetchExpired(long currentTimeMs) {
+ fetchTimer.update(currentTimeMs);
+ boolean isExpired = fetchTimer.isExpired();
+ if (isExpired) {
+ log.info("Did not receive fetch request from the majority of the voters within {}ms. Current fetched voters are {}.",
+ fetchTimeoutMs, fetchedVoters);
+ }
+ return isExpired;
+ }
+
+ // Reset the fetch timer if we've received fetch/fetchSnapshot request from the majority of the voter
Review Comment:
Same comment here. Can we make this a Java doc comment?
##########
raft/src/main/java/org/apache/kafka/raft/LeaderState.java:
##########
@@ -79,6 +86,39 @@ protected LeaderState(
this.grantingVoters = Collections.unmodifiableSet(new HashSet<>(grantingVoters));
this.log = logContext.logger(LeaderState.class);
this.accumulator = Objects.requireNonNull(accumulator, "accumulator must be non-null");
+ // use the 1.5x fetch timeout to tolerate some network transition time or other IO time.
+ this.fetchTimeoutMs = (int) (fetchTimeoutMs * 1.5);
+ this.fetchTimer = time.timer(fetchTimeoutMs);
+ }
+
+ // Check if the fetchTimer is expired because we didn't receive a valid fetch/fetchSnapshot request from the majority of
+ // the voters within fetch timeout.
+ public boolean hasMajorityFollowerFetchExpired(long currentTimeMs) {
+ fetchTimer.update(currentTimeMs);
+ boolean isExpired = fetchTimer.isExpired();
+ if (isExpired) {
+ log.info("Did not receive fetch request from the majority of the voters within {}ms. Current fetched voters are {}.",
+ fetchTimeoutMs, fetchedVoters);
Review Comment:
We use the following indentations in the `raft` module:
```java
log.info(
"Did not receive fetch request from the majority of the voters within {}ms. Current fetched voters are {}.",
fetchTimeoutMs,
fetchedVoters
);
```
##########
raft/src/main/java/org/apache/kafka/raft/LeaderState.java:
##########
@@ -55,17 +57,22 @@ public class LeaderState<T> implements EpochState {
private final Map<Integer, ReplicaState> observerStates = new HashMap<>();
private final Logger log;
private final BatchAccumulator<T> accumulator;
+ private final Set<Integer> fetchedVoters = new HashSet<>();
Review Comment:
What do you think about writing a comment explaining this field. For example, this set includes all of the voter followers that `FETCH` or `FETCH_SNAPSHOT` during the current `fetchTimer` interval.
##########
raft/src/main/java/org/apache/kafka/raft/LeaderState.java:
##########
@@ -79,6 +86,39 @@ protected LeaderState(
this.grantingVoters = Collections.unmodifiableSet(new HashSet<>(grantingVoters));
this.log = logContext.logger(LeaderState.class);
this.accumulator = Objects.requireNonNull(accumulator, "accumulator must be non-null");
+ // use the 1.5x fetch timeout to tolerate some network transition time or other IO time.
+ this.fetchTimeoutMs = (int) (fetchTimeoutMs * 1.5);
+ this.fetchTimer = time.timer(fetchTimeoutMs);
+ }
+
+ // Check if the fetchTimer is expired because we didn't receive a valid fetch/fetchSnapshot request from the majority of
+ // the voters within fetch timeout.
+ public boolean hasMajorityFollowerFetchExpired(long currentTimeMs) {
Review Comment:
See my other comment about check quorum and Alyssa pre-vote KIP. If you agree, maybe we should call this `hasCheckQuorumFailed`.
##########
clients/src/main/java/org/apache/kafka/common/requests/FetchSnapshotRequest.java:
##########
@@ -70,6 +71,7 @@ public static FetchSnapshotRequestData singleton(
return new FetchSnapshotRequestData()
.setClusterId(clusterId)
+ .setReplicaId(replicaId)
Review Comment:
Good catch. Thanks for fixing this!
I think this is the best we can do here. This means that FETCH_SNAPSHOT won't count towards check quorum until the voters have been upgraded to this version.
##########
raft/src/test/java/org/apache/kafka/raft/LeaderStateTest.java:
##########
@@ -447,6 +452,38 @@ public void testDescribeQuorumWithObservers() {
observerState);
}
+ @Test
+ public void testMajorityFollowerFetchTimeoutExpiration() {
+ int node1 = 1;
+ int node2 = 2;
+ int node3 = 3;
+ int node4 = 4;
+ int observer5 = 5;
+ LeaderState<?> state = newLeaderState(mkSet(localId, node1, node2, node3, node4), 0L);
+ assertFalse(state.hasMajorityFollowerFetchExpired(time.milliseconds()));
+ int resignLeadershipTimeout = (int) (fetchTimeoutMs * 1.5);
Review Comment:
This applies to the other tests but I am thinking that we should either 1) make `1.5` a constant and use that constant in all of the tests or 2) add a method that returns the "check quorum" timeout ms and use that in all of the tests.
##########
raft/src/main/java/org/apache/kafka/raft/LeaderState.java:
##########
@@ -79,6 +86,39 @@ protected LeaderState(
this.grantingVoters = Collections.unmodifiableSet(new HashSet<>(grantingVoters));
this.log = logContext.logger(LeaderState.class);
this.accumulator = Objects.requireNonNull(accumulator, "accumulator must be non-null");
+ // use the 1.5x fetch timeout to tolerate some network transition time or other IO time.
+ this.fetchTimeoutMs = (int) (fetchTimeoutMs * 1.5);
+ this.fetchTimer = time.timer(fetchTimeoutMs);
+ }
+
+ // Check if the fetchTimer is expired because we didn't receive a valid fetch/fetchSnapshot request from the majority of
+ // the voters within fetch timeout.
+ public boolean hasMajorityFollowerFetchExpired(long currentTimeMs) {
+ fetchTimer.update(currentTimeMs);
+ boolean isExpired = fetchTimer.isExpired();
+ if (isExpired) {
+ log.info("Did not receive fetch request from the majority of the voters within {}ms. Current fetched voters are {}.",
+ fetchTimeoutMs, fetchedVoters);
+ }
+ return isExpired;
+ }
+
+ // Reset the fetch timer if we've received fetch/fetchSnapshot request from the majority of the voter
+ public void maybeResetMajorityFollowerFetchTimer(int id, long currentTimeMs) {
Review Comment:
How about `updateCheckQuorumForFollowingVoter(int id, long currentTimeMs)`?
--
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-15489: resign leadership when no fetch from majority voters [kafka]
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on code in PR #14428:
URL: https://github.com/apache/kafka/pull/14428#discussion_r1406995366
##########
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java:
##########
@@ -1990,7 +1995,7 @@ private long pollLeader(long currentTimeMs) {
LeaderState<T> state = quorum.leaderStateOrThrow();
maybeFireLeaderChange(state);
- if (shutdown.get() != null || state.isResignRequested()) {
+ if (shutdown.get() != null || state.isResignRequested() || state.hasMajorityFollowerFetchExpired(currentTimeMs)) {
Review Comment:
In line [2014](https://github.com/apache/kafka/pull/14428/files#diff-1da15c51e641ea46ea5c86201ab8f21cfee9e7c575102a39c7bae0d5ffd7de39R2014) KRaft calculates the next time it should poll the leader.
```java
return Math.min(timeUntilFlush, timeUntilSend);
```
If you extend this to something like below, it should wake up the leader when the timer has expired.
```java
return Math.min(timeUntilFlush, timeUntilSend, state.timeUntilCheckQuorumExpires(currentTimeMs));
```
--
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-15489: resign leadership when no fetch from majority voters [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #14428:
URL: https://github.com/apache/kafka/pull/14428#discussion_r1408609808
##########
raft/src/main/java/org/apache/kafka/raft/LeaderState.java:
##########
@@ -79,6 +88,55 @@ protected LeaderState(
this.grantingVoters = Collections.unmodifiableSet(new HashSet<>(grantingVoters));
this.log = logContext.logger(LeaderState.class);
this.accumulator = Objects.requireNonNull(accumulator, "accumulator must be non-null");
+ // use the 1.5x of fetch timeout to tolerate some network transition time or other IO time.
+ this.checkQuorumTimeoutMs = (int) (fetchTimeoutMs * CHECK_QUORUM_TIMEOUT_FACTOR);
+ this.checkQuorumTimer = time.timer(checkQuorumTimeoutMs);
+ }
+
+ /**
+ * Get the remaining time in milliseconds until the checkQuorumTimer expires.
+ * This will happen if we didn't receive a valid fetch/fetchSnapshot request from the majority of the voters within checkQuorumTimeoutMs.
+ *
+ * @param currentTimeMs the current timestamp in millisecond
+ * @return the remainingMs before the checkQuorumTimer expired
+ */
+ public long timeUntilCheckQuorumExpires(long currentTimeMs) {
+ checkQuorumTimer.update(currentTimeMs);
+ long remainingMs = checkQuorumTimer.remainingMs();
+ if (remainingMs == 0) {
+ log.info(
+ "Did not receive fetch request from the majority of the voters within {}ms. Current fetched voters are {}.",
+ checkQuorumTimeoutMs,
+ fetchedVoters);
Review Comment:
Updated.
--
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-15489: resign leadership when no fetch from majority voters [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on PR #14428:
URL: https://github.com/apache/kafka/pull/14428#issuecomment-1831052748
> I took at look at the build and there seem to be a lot of failures. Can you confirm that they are not related to this change?
No, they doesn't look related. Let's check the latest build results later.
--
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-15489: resign leadership when no fetch from majority voters [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #14428:
URL: https://github.com/apache/kafka/pull/14428#discussion_r1342669657
##########
raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientTest.java:
##########
@@ -485,6 +485,49 @@ public void testHandleBeginQuorumEpochAfterUserInitiatedResign() throws Exceptio
context.listener.currentLeaderAndEpoch());
}
+ @Test
+ public void testLeaderShouldResignLeadershipIfNotGetFetchRequestFromMajorityVoters() throws Exception {
Review Comment:
Will add tests later.
--
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-15489: resign leadership when no fetch from majority voters [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #14428:
URL: https://github.com/apache/kafka/pull/14428#discussion_r1343475329
##########
raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientTest.java:
##########
@@ -485,6 +485,49 @@ public void testHandleBeginQuorumEpochAfterUserInitiatedResign() throws Exceptio
context.listener.currentLeaderAndEpoch());
}
+ @Test
+ public void testLeaderShouldResignLeadershipIfNotGetFetchRequestFromMajorityVoters() throws Exception {
Review Comment:
Added a test in `KafkaRaftClientSnapshotTest`.
> Can we also tests the opposite. That the leader doesn't resign if the majority of the replicas (including the leader) have fetch in the last fetchTimeoutMs?
I didn't follow you. I've verified:
```
- 1/2 fetch time
leadership not get reassigned
- fetch from one voter
- 1/2 fetch time
leadership not get reassigned
- fetch from another voter
- 1/2 fetch time
leadership not get reassigned
- fetch from the observer
- 1/2 fetch time
leadership should get reassigned
```
I think I've verified what you want. Let me know if I need to add other things. Thanks.
--
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-15489: resign leadership when no fetch from majority voters [kafka]
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on code in PR #14428:
URL: https://github.com/apache/kafka/pull/14428#discussion_r1406771117
##########
raft/src/main/java/org/apache/kafka/raft/LeaderState.java:
##########
@@ -79,6 +86,39 @@ protected LeaderState(
this.grantingVoters = Collections.unmodifiableSet(new HashSet<>(grantingVoters));
this.log = logContext.logger(LeaderState.class);
this.accumulator = Objects.requireNonNull(accumulator, "accumulator must be non-null");
+ // use the 1.5x fetch timeout to tolerate some network transition time or other IO time.
+ this.fetchTimeoutMs = (int) (fetchTimeoutMs * 1.5);
+ this.fetchTimer = time.timer(fetchTimeoutMs);
+ }
+
+ // Check if the fetchTimer is expired because we didn't receive a valid fetch/fetchSnapshot request from the majority of
+ // the voters within fetch timeout.
Review Comment:
Can we make this a Java doc comment. E.g.:
```java
/**
* Check if the fetchTimer is expired because we didn't receive a valid fetch/fetchSnapshot request from
* the majority of the voters within fetch timeout.
*
* @param ...
* @return ...
*/
```
--
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-15489: resign leadership when no fetch from majority voters [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on PR #14428:
URL: https://github.com/apache/kafka/pull/14428#issuecomment-1835574772
Yes, I agree. Thanks for helping merge 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
[GitHub] [kafka] showuon commented on pull request #14428: KAFKA-15489: resign leadership when no fetch from majority voters
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on PR #14428:
URL: https://github.com/apache/kafka/pull/14428#issuecomment-1732271440
@hachikuji @cmccabe @jsancio , call for review. Thanks.
--
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-15489: resign leadership when no fetch from majority voters [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #14428:
URL: https://github.com/apache/kafka/pull/14428#discussion_r1343472387
##########
raft/src/main/java/org/apache/kafka/raft/LeaderState.java:
##########
@@ -76,9 +85,37 @@ protected LeaderState(
boolean hasAcknowledgedLeader = voterId == localId;
this.voterStates.put(voterId, new ReplicaState(voterId, hasAcknowledgedLeader));
}
+ this.majority = voters.size() / 2;
this.grantingVoters = Collections.unmodifiableSet(new HashSet<>(grantingVoters));
this.log = logContext.logger(LeaderState.class);
this.accumulator = Objects.requireNonNull(accumulator, "accumulator must be non-null");
+ this.fetchTimeoutMs = fetchTimeoutMs;
+ this.fetchTimer = time.timer(fetchTimeoutMs);
+ }
+
+ public boolean hasMajorityFollowerFetchTimeoutExpired(long currentTimeMs) {
Review Comment:
Fair enough. Added comments on the methods. Thanks.
--
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-15489: resign leadership when no fetch from majority voters [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on PR #14428:
URL: https://github.com/apache/kafka/pull/14428#issuecomment-1744833245
@jsancio @ahuang98 , I've addressed the comments. Please take a look again. Thanks.
--
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-15489: resign leadership when no fetch from majority voters [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #14428:
URL: https://github.com/apache/kafka/pull/14428#discussion_r1342676499
##########
raft/src/main/java/org/apache/kafka/raft/LeaderState.java:
##########
@@ -76,9 +85,37 @@ protected LeaderState(
boolean hasAcknowledgedLeader = voterId == localId;
this.voterStates.put(voterId, new ReplicaState(voterId, hasAcknowledgedLeader));
}
+ this.majority = voters.size() / 2;
this.grantingVoters = Collections.unmodifiableSet(new HashSet<>(grantingVoters));
this.log = logContext.logger(LeaderState.class);
this.accumulator = Objects.requireNonNull(accumulator, "accumulator must be non-null");
+ this.fetchTimeoutMs = fetchTimeoutMs;
+ this.fetchTimer = time.timer(fetchTimeoutMs);
+ }
+
+ public boolean hasMajorityFollowerFetchTimeoutExpired(long currentTimeMs) {
+ fetchTimer.update(currentTimeMs);
+ boolean isExpired = fetchTimer.isExpired();
+ if (isExpired) {
+ log.info("Did not receive fetch request from the majority of the voters within {}ms. Current fetched voters are {}.",
+ fetchTimeoutMs, fetchedVoters);
+ }
+ return isExpired;
+ }
+
+ public void maybeResetMajorityFollowerFetchTimeout(int id, long currentTimeMs) {
+ updateFetchedVoters(id);
+ if (fetchedVoters.size() >= majority) {
+ fetchedVoters.clear();
+ fetchTimer.update(currentTimeMs);
+ fetchTimer.reset(fetchTimeoutMs);
+ }
+ }
+
+ private void updateFetchedVoters(int id) {
+ if (isVoter(id)) {
+ fetchedVoters.add(id);
+ }
Review Comment:
> Note that ReplicaState already contains the lastFetchTimestamp.
I'm trying to re-use the `lastFetchTimestamp` in ReplicaState today, but found it won't work as expected since the default value for it is `-1`, which means, when a note becomes a leader, all the `lastFetchTimestamp` of follower nodes are `-1`. Using current `timer` way is more readable IMO.
> The part that is not clear to me is when or how to wake up the leader for a poll. We need to update KafkaRaftClient::pollLeader so that the replicas' last fetch time is taken into account when blocking on the messageQueue.poll
Good question. My thought is, we add some buffer to tolerate the operation time. Like when [checking shrinkISR](https://github.com/apache/kafka/blob/trunk/core/src/main/scala/kafka/server/ReplicaManager.scala#L373-L375), we give a 1.5x of the timeout to make things easier, instead of calculating the accurate timestamp. So, I'm thinking we use `fetchTimeout * 1.5`. WDYT?
--
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-15489: resign leadership when no fetch from majority voters [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #14428:
URL: https://github.com/apache/kafka/pull/14428#discussion_r1342668812
##########
raft/src/main/java/org/apache/kafka/raft/LeaderState.java:
##########
@@ -76,9 +85,37 @@ protected LeaderState(
boolean hasAcknowledgedLeader = voterId == localId;
this.voterStates.put(voterId, new ReplicaState(voterId, hasAcknowledgedLeader));
}
+ this.majority = voters.size() / 2;
this.grantingVoters = Collections.unmodifiableSet(new HashSet<>(grantingVoters));
this.log = logContext.logger(LeaderState.class);
this.accumulator = Objects.requireNonNull(accumulator, "accumulator must be non-null");
+ this.fetchTimeoutMs = fetchTimeoutMs;
+ this.fetchTimer = time.timer(fetchTimeoutMs);
+ }
+
+ public boolean hasMajorityFollowerFetchTimeoutExpired(long currentTimeMs) {
Review Comment:
Changed to `hasMajorityFollowerFetchExpired`. Let me know if you have any better suggestion. Thanks.
--
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