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/30 01:38:21 UTC
[PR] [Do not merge] Test which shows observers stuck fetching from bootstrap servers [kafka]
ahuang98 opened a new pull request, #19854:
URL: https://github.com/apache/kafka/pull/19854
`testObserverSendDiscoveryFetchAfterFetchTimeoutAndResumesFetchingFromLeader` demonstrates the issue that observers will get stuck fetching from random bootstrap servers after fetch timeout.
`testUnattachedWithLeaderCanBecomeFollowerAfterFindingLeader` shows voters are not susceptible to teh same 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-19354: KRaft observer should fetch from leader after rediscovery [kafka]
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on code in PR #19854:
URL: https://github.com/apache/kafka/pull/19854#discussion_r2121590408
##########
raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientTest.java:
##########
@@ -2145,7 +2186,20 @@ public void testObserverSendDiscoveryFetchAfterFetchTimeout(boolean withKip853Rp
fetchRequest = context.assertSentFetchRequest();
assertNotEquals(leaderId, fetchRequest.destination().id());
assertTrue(context.bootstrapIds.contains(fetchRequest.destination().id()));
+ assertTrue(context.client.quorum().followerStateOrThrow().hasFetchTimeoutExpired(context.time.milliseconds()));
context.assertFetchRequestData(fetchRequest, epoch, 0L, 0);
+
+ context.deliverResponse(
+ fetchRequest.correlationId(),
+ fetchRequest.destination(),
+ context.fetchResponse(epoch, leaderId, MemoryRecords.EMPTY, 0L, Errors.NOT_LEADER_OR_FOLLOWER)
+ );
+
+ context.client.poll();
+ context.pollUntilRequest();
+ fetchRequest = context.assertSentFetchRequest();
+ assertEquals(leaderId, fetchRequest.destination().id());
+ assertFalse(context.client.quorum().followerStateOrThrow().hasFetchTimeoutExpired(context.time.milliseconds()));
Review Comment:
Let's move this line. This suite should test the protocol and not the internal implementation state.
##########
raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientTest.java:
##########
@@ -2145,7 +2186,20 @@ public void testObserverSendDiscoveryFetchAfterFetchTimeout(boolean withKip853Rp
fetchRequest = context.assertSentFetchRequest();
assertNotEquals(leaderId, fetchRequest.destination().id());
assertTrue(context.bootstrapIds.contains(fetchRequest.destination().id()));
+ assertTrue(context.client.quorum().followerStateOrThrow().hasFetchTimeoutExpired(context.time.milliseconds()));
Review Comment:
Let's not test internal kraft state. This suite should check the protocol.
##########
raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientTest.java:
##########
@@ -2033,6 +2033,47 @@ public void testUnattachedAsVoterCanBecomeFollowerAfterFindingLeader(boolean wit
assertTrue(context.client.quorum().isFollower());
}
+ @ParameterizedTest
+ @ValueSource(booleans = { true, false })
+ public void testUnattachedWithLeaderCanBecomeFollowerAfterFindingLeader(boolean withKip853Rpc) throws Exception {
+ int localId = randomReplicaId();
+ int otherNodeId = localId + 1;
+ int leaderNodeId = localId + 2;
+ int epoch = 5;
+ Set<Integer> voters = Set.of(localId, otherNodeId, leaderNodeId);
+ List<InetSocketAddress> bootstrapServers = voters
+ .stream()
+ .map(RaftClientTestContext::mockAddress)
+ .toList();
+
+ RaftClientTestContext context = new RaftClientTestContext.Builder(localId, voters)
+ .withBootstrapServers(Optional.of(bootstrapServers))
+ .withElectedLeader(epoch, leaderNodeId)
+ .withKip853Rpc(withKip853Rpc)
+ .build();
+
+ // after fetch timeout, node will become prospective with leader
+ context.time.sleep(context.fetchTimeoutMs);
+ context.client.poll();
+ assertTrue(context.client.quorum().isProspective());
+ assertEquals(leaderNodeId, context.client.quorum().leaderId().getAsInt());
Review Comment:
You can use `RaftClientTestContext#assertElectedLeader(int, int)`.
##########
raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientTest.java:
##########
@@ -2033,6 +2033,47 @@ public void testUnattachedAsVoterCanBecomeFollowerAfterFindingLeader(boolean wit
assertTrue(context.client.quorum().isFollower());
}
+ @ParameterizedTest
+ @ValueSource(booleans = { true, false })
+ public void testUnattachedWithLeaderCanBecomeFollowerAfterFindingLeader(boolean withKip853Rpc) throws Exception {
+ int localId = randomReplicaId();
+ int otherNodeId = localId + 1;
+ int leaderNodeId = localId + 2;
+ int epoch = 5;
+ Set<Integer> voters = Set.of(localId, otherNodeId, leaderNodeId);
+ List<InetSocketAddress> bootstrapServers = voters
+ .stream()
+ .map(RaftClientTestContext::mockAddress)
+ .toList();
+
+ RaftClientTestContext context = new RaftClientTestContext.Builder(localId, voters)
+ .withBootstrapServers(Optional.of(bootstrapServers))
+ .withElectedLeader(epoch, leaderNodeId)
+ .withKip853Rpc(withKip853Rpc)
+ .build();
+
+ // after fetch timeout, node will become prospective with leader
+ context.time.sleep(context.fetchTimeoutMs);
+ context.client.poll();
+ assertTrue(context.client.quorum().isProspective());
+ assertEquals(leaderNodeId, context.client.quorum().leaderId().getAsInt());
+ assertEquals(epoch, context.currentEpoch());
+
+ // after election loss node will become follower because it had a last known leader
+ context.time.sleep(context.electionTimeoutMs() * 2L);
+ context.client.poll();
+ assertTrue(context.client.quorum().isFollower());
+ assertEquals(leaderNodeId, context.client.quorum().leaderId().getAsInt());
+ assertEquals(epoch, context.currentEpoch());
+
+ // node will send fetch request to leader
+ context.pollUntilRequest();
Review Comment:
Why not `pollUntilRequest` and check that the destination is the leader and the epoch is `epoch`?
We should try to avoid testing internal KRaft state in protocol suites like `KafkaRaftClientTest`.
##########
raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientTest.java:
##########
@@ -2033,6 +2033,47 @@ public void testUnattachedAsVoterCanBecomeFollowerAfterFindingLeader(boolean wit
assertTrue(context.client.quorum().isFollower());
}
+ @ParameterizedTest
+ @ValueSource(booleans = { true, false })
+ public void testUnattachedWithLeaderCanBecomeFollowerAfterFindingLeader(boolean withKip853Rpc) throws Exception {
+ int localId = randomReplicaId();
+ int otherNodeId = localId + 1;
+ int leaderNodeId = localId + 2;
+ int epoch = 5;
+ Set<Integer> voters = Set.of(localId, otherNodeId, leaderNodeId);
+ List<InetSocketAddress> bootstrapServers = voters
+ .stream()
+ .map(RaftClientTestContext::mockAddress)
+ .toList();
+
+ RaftClientTestContext context = new RaftClientTestContext.Builder(localId, voters)
+ .withBootstrapServers(Optional.of(bootstrapServers))
+ .withElectedLeader(epoch, leaderNodeId)
+ .withKip853Rpc(withKip853Rpc)
+ .build();
+
+ // after fetch timeout, node will become prospective with leader
+ context.time.sleep(context.fetchTimeoutMs);
+ context.client.poll();
+ assertTrue(context.client.quorum().isProspective());
Review Comment:
Same comment here. The test should instead expect VOTE requests with pre-vote set to true.
--
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-19354: KRaft observer should fetch from leader after rediscovery [kafka]
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on code in PR #19854:
URL: https://github.com/apache/kafka/pull/19854#discussion_r2228734466
##########
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java:
##########
@@ -2569,13 +2569,15 @@ private void maybeTransition(
} else {
transitionToUnattached(epoch, OptionalInt.empty());
}
- } else if (
- leaderId.isPresent() &&
- (!quorum.hasLeader() || leaderEndpoints.size() > quorum.leaderEndpoints().size())
- ) {
- // The request or response indicates the leader of the current epoch
- // which are currently unknown or the replica has discovered more endpoints
- transitionToFollower(epoch, leaderId.getAsInt(), leaderEndpoints, currentTimeMs);
+ } else if (leaderId.isPresent()) {
+ if (quorum.isFollowerObserver()) {
Review Comment:
Got it. What about the first question? Why don't we also need to do this for voters?
--
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-19354: KRaft observer should fetch from leader after rediscovery [kafka]
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on code in PR #19854:
URL: https://github.com/apache/kafka/pull/19854#discussion_r2232057083
##########
raft/src/test/java/org/apache/kafka/raft/RequestManagerTest.java:
##########
@@ -245,6 +245,41 @@ public void testFindReadyWithRequestTimedOut() {
assertFalse(cache.isResponseExpected(otherNode, 1));
}
+ @Test
+ public void testHasRequestTimedOut() {
+ List<Node> bootstrapList = makeBootstrapList(2);
+ RequestManager cache = new RequestManager(
+ bootstrapList,
+ retryBackoffMs,
+ requestTimeoutMs,
+ random
+ );
+
+ // Find a ready node with the starting state
+ Node bootstrapNode1 = cache.findReadyBootstrapServer(time.milliseconds()).get();
+ assertTrue(
+ bootstrapList.contains(bootstrapNode1),
+ String.format("%s is not in %s", bootstrapNode1, bootstrapList)
+ );
+ // Before sending a request, no request should have timed out
+ assertFalse(cache.hasRequestTimedOut(bootstrapNode1, time.milliseconds()));
+
+ // Send a request
+ cache.onRequestSent(bootstrapNode1, 1, time.milliseconds());
+ assertEquals(
+ Optional.empty(),
+ cache.findReadyBootstrapServer(time.milliseconds())
+ );
+ assertFalse(cache.hasRequestTimedOut(bootstrapNode1, time.milliseconds()));
+
+ time.sleep(requestTimeoutMs / 2);
+ assertFalse(cache.hasRequestTimedOut(bootstrapNode1, time.milliseconds()));
+
+ // Timeout the request
+ time.sleep(requestTimeoutMs / 2);
Review Comment:
In integer division, `value / 2 + value / 2` does not equal `value`. You can see this when value is 3. `3/2` is `1` and `1 + 1` is not 3. A more obvious sleep logic is:
```java
time.sleep(requestTimeoutMs - 1);
assertFalse(cache.hasRequestTimedOut(bootstrapNode1, time.milliseconds()));
// Timeout the request
time.sleep(1);
```
##########
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java:
##########
@@ -2570,7 +2570,7 @@ private void maybeTransition(
transitionToUnattached(epoch, OptionalInt.empty());
}
} else if (
- leaderId.isPresent() &&
+ leaderId.isPresent() &&
(!quorum.hasLeader() || leaderEndpoints.size() > quorum.leaderEndpoints().size())
Review Comment:
How about removing the extra spaces from this line too?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19354: KRaft observer should fetch from leader after rediscovery [kafka]
Posted by "ahuang98 (via GitHub)" <gi...@apache.org>.
ahuang98 commented on PR #19854:
URL: https://github.com/apache/kafka/pull/19854#issuecomment-2931563818
Thanks for the review Jose
> Does this mean that voters can never rediscover the leader from the bootstrap server? Or this doesn't impact voter because they transition to prospective and back to follower which resets the fetch timer?
Yes, this doesn't impact voters because prospectives which have a leaderId would transition back to follower (and refreshed fetch timeout)
--
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-19354: KRaft observer should fetch from leader after rediscovery [kafka]
Posted by "ahuang98 (via GitHub)" <gi...@apache.org>.
ahuang98 commented on code in PR #19854:
URL: https://github.com/apache/kafka/pull/19854#discussion_r2232070935
##########
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java:
##########
@@ -2570,7 +2570,7 @@ private void maybeTransition(
transitionToUnattached(epoch, OptionalInt.empty());
}
} else if (
- leaderId.isPresent() &&
+ leaderId.isPresent() &&
(!quorum.hasLeader() || leaderEndpoints.size() > quorum.leaderEndpoints().size())
Review Comment:
this is technically the correct indexing, going to just leave this in
--
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-19354: KRaft observer should fetch from leader after rediscovery [kafka]
Posted by "ahuang98 (via GitHub)" <gi...@apache.org>.
ahuang98 commented on code in PR #19854:
URL: https://github.com/apache/kafka/pull/19854#discussion_r2226490247
##########
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java:
##########
@@ -2569,13 +2569,15 @@ private void maybeTransition(
} else {
transitionToUnattached(epoch, OptionalInt.empty());
}
- } else if (
- leaderId.isPresent() &&
- (!quorum.hasLeader() || leaderEndpoints.size() > quorum.leaderEndpoints().size())
- ) {
- // The request or response indicates the leader of the current epoch
- // which are currently unknown or the replica has discovered more endpoints
- transitionToFollower(epoch, leaderId.getAsInt(), leaderEndpoints, currentTimeMs);
+ } else if (leaderId.isPresent()) {
+ if (quorum.isFollowerObserver()) {
Review Comment:
There were a few reasons - I'm curious to know what you think
- if the node re-transitions to follower, we need to remove the safeguard which currently prevents follower to follower transitions without an epoch bump
- we save some extraneous actions which occur in transitionToFollower (e.g. no need to reset connections or use durable transition, we don't have any new state like leaderId or votedKey to persist)
--
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-19354: KRaft observer should send fetch to best node [kafka]
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio merged PR #19854:
URL: https://github.com/apache/kafka/pull/19854
--
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-19354: KRaft observer should fetch from leader after rediscovery [kafka]
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on code in PR #19854:
URL: https://github.com/apache/kafka/pull/19854#discussion_r2195283553
##########
raft/src/main/java/org/apache/kafka/raft/FollowerState.java:
##########
@@ -137,6 +137,11 @@ public void resetFetchTimeoutForSuccessfulFetch(long currentTimeMs) {
hasFetchedFromLeader = true;
}
+ public void resetFetchTimeoutAfterDiscoveringLeader(long currentTimeMs) {
Review Comment:
Let's document these 3 methods: `resetFetchTimeoutForSuccessfulFetch`, `resetFetchTimeoutAfterDiscoveringLeader` and `overrideFetchTimeout`.
How about removing code duplication with:
```java
public void resetFetchTimeoutForSuccessfulFetch(long currentTimeMs) {
resetFetchTimeout(currentTimeMs);
hasFetchedFromLeader = true;
}
public void resetFetchTimeout(long currentTimeMs) {
overrideFetchTimeout(currentTimeMs, fetchTimeoutMs);
}
```
##########
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java:
##########
@@ -2569,13 +2569,15 @@ private void maybeTransition(
} else {
transitionToUnattached(epoch, OptionalInt.empty());
}
- } else if (
- leaderId.isPresent() &&
- (!quorum.hasLeader() || leaderEndpoints.size() > quorum.leaderEndpoints().size())
- ) {
- // The request or response indicates the leader of the current epoch
- // which are currently unknown or the replica has discovered more endpoints
- transitionToFollower(epoch, leaderId.getAsInt(), leaderEndpoints, currentTimeMs);
+ } else if (leaderId.isPresent()) {
+ if (quorum.isFollowerObserver()) {
Review Comment:
Why don't we also need to do this for voters? If so, should the logic always re-transition to follower?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19354: KRaft observer should fetch from leader after rediscovery [kafka]
Posted by "ahuang98 (via GitHub)" <gi...@apache.org>.
ahuang98 commented on code in PR #19854:
URL: https://github.com/apache/kafka/pull/19854#discussion_r2231846413
##########
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java:
##########
@@ -2569,13 +2569,15 @@ private void maybeTransition(
} else {
transitionToUnattached(epoch, OptionalInt.empty());
}
- } else if (
- leaderId.isPresent() &&
- (!quorum.hasLeader() || leaderEndpoints.size() > quorum.leaderEndpoints().size())
- ) {
- // The request or response indicates the leader of the current epoch
- // which are currently unknown or the replica has discovered more endpoints
- transitionToFollower(epoch, leaderId.getAsInt(), leaderEndpoints, currentTimeMs);
+ } else if (leaderId.isPresent()) {
+ if (quorum.isFollowerObserver()) {
Review Comment:
discussed offline, pushed a new commit with a more clean method!
--
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-19354: KRaft observer should fetch from leader after rediscovery [kafka]
Posted by "ahuang98 (via GitHub)" <gi...@apache.org>.
ahuang98 commented on code in PR #19854:
URL: https://github.com/apache/kafka/pull/19854#discussion_r2136234763
##########
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java:
##########
@@ -2543,13 +2543,16 @@ private void maybeTransition(
} else {
transitionToUnattached(epoch, OptionalInt.empty());
}
- } else if (
- leaderId.isPresent() &&
- (!quorum.hasLeader() || leaderEndpoints.size() > quorum.leaderEndpoints().size())
- ) {
- // The request or response indicates the leader of the current epoch
- // which are currently unknown or the replica has discovered more endpoints
- transitionToFollower(epoch, leaderId.getAsInt(), leaderEndpoints, currentTimeMs);
+ } else if (leaderId.isPresent()) {
+ if (quorum.isFollowerObserver()) {
+ // This allows observers to resume fetching from leader after discovering the leader
+// transitionToFollower(epoch, leaderId.getAsInt(), leaderEndpoints, currentTimeMs);
Review Comment:
alternative solution
--
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-19354: KRaft observer should fetch from leader after rediscovery [kafka]
Posted by "ahuang98 (via GitHub)" <gi...@apache.org>.
ahuang98 commented on code in PR #19854:
URL: https://github.com/apache/kafka/pull/19854#discussion_r2136234763
##########
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java:
##########
@@ -2543,13 +2543,16 @@ private void maybeTransition(
} else {
transitionToUnattached(epoch, OptionalInt.empty());
}
- } else if (
- leaderId.isPresent() &&
- (!quorum.hasLeader() || leaderEndpoints.size() > quorum.leaderEndpoints().size())
- ) {
- // The request or response indicates the leader of the current epoch
- // which are currently unknown or the replica has discovered more endpoints
- transitionToFollower(epoch, leaderId.getAsInt(), leaderEndpoints, currentTimeMs);
+ } else if (leaderId.isPresent()) {
+ if (quorum.isFollowerObserver()) {
+ // This allows observers to resume fetching from leader after discovering the leader
+// transitionToFollower(epoch, leaderId.getAsInt(), leaderEndpoints, currentTimeMs);
Review Comment:
alternative solution, with allowance for follower to follower transition for observers only
##########
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java:
##########
@@ -2543,13 +2543,16 @@ private void maybeTransition(
} else {
transitionToUnattached(epoch, OptionalInt.empty());
}
- } else if (
- leaderId.isPresent() &&
- (!quorum.hasLeader() || leaderEndpoints.size() > quorum.leaderEndpoints().size())
- ) {
- // The request or response indicates the leader of the current epoch
- // which are currently unknown or the replica has discovered more endpoints
- transitionToFollower(epoch, leaderId.getAsInt(), leaderEndpoints, currentTimeMs);
+ } else if (leaderId.isPresent()) {
+ if (quorum.isFollowerObserver()) {
+ // This allows observers to resume fetching from leader after discovering the leader
+// transitionToFollower(epoch, leaderId.getAsInt(), leaderEndpoints, currentTimeMs);
Review Comment:
alternative solution, needing allowance for follower to follower transition for observers only
--
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] [Do not merge] Test which shows observers stuck fetching from bootstrap servers [kafka]
Posted by "ahuang98 (via GitHub)" <gi...@apache.org>.
ahuang98 commented on code in PR #19854:
URL: https://github.com/apache/kafka/pull/19854#discussion_r2114996046
##########
raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientTest.java:
##########
@@ -2145,7 +2186,19 @@ public void testObserverSendDiscoveryFetchAfterFetchTimeout(boolean withKip853Rp
fetchRequest = context.assertSentFetchRequest();
assertNotEquals(leaderId, fetchRequest.destination().id());
assertTrue(context.bootstrapIds.contains(fetchRequest.destination().id()));
+ assertTrue(context.client.quorum().followerStateOrThrow().hasFetchTimeoutExpired(context.time.milliseconds()));
context.assertFetchRequestData(fetchRequest, epoch, 0L, 0);
+
+ context.deliverResponse(
+ fetchRequest.correlationId(),
+ fetchRequest.destination(),
+ context.fetchResponse(epoch, leaderId, MemoryRecords.EMPTY, 0L, Errors.NOT_LEADER_OR_FOLLOWER)
+ );
+
+ context.pollUntilRequest();
+ fetchRequest = context.assertSentFetchRequest();
+ assertEquals(leaderId, fetchRequest.destination().id());
Review Comment:
this check fails
--
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] [Do not merge] Test which shows observers stuck fetching from bootstrap servers [kafka]
Posted by "ahuang98 (via GitHub)" <gi...@apache.org>.
ahuang98 commented on code in PR #19854:
URL: https://github.com/apache/kafka/pull/19854#discussion_r2114996046
##########
raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientTest.java:
##########
@@ -2145,7 +2186,19 @@ public void testObserverSendDiscoveryFetchAfterFetchTimeout(boolean withKip853Rp
fetchRequest = context.assertSentFetchRequest();
assertNotEquals(leaderId, fetchRequest.destination().id());
assertTrue(context.bootstrapIds.contains(fetchRequest.destination().id()));
+ assertTrue(context.client.quorum().followerStateOrThrow().hasFetchTimeoutExpired(context.time.milliseconds()));
context.assertFetchRequestData(fetchRequest, epoch, 0L, 0);
+
+ context.deliverResponse(
+ fetchRequest.correlationId(),
+ fetchRequest.destination(),
+ context.fetchResponse(epoch, leaderId, MemoryRecords.EMPTY, 0L, Errors.NOT_LEADER_OR_FOLLOWER)
+ );
+
+ context.pollUntilRequest();
+ fetchRequest = context.assertSentFetchRequest();
+ assertEquals(leaderId, fetchRequest.destination().id());
Review Comment:
this fails locally
--
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-19354: KRaft observer should fetch from leader after rediscovery [kafka]
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on code in PR #19854:
URL: https://github.com/apache/kafka/pull/19854#discussion_r2232057083
##########
raft/src/test/java/org/apache/kafka/raft/RequestManagerTest.java:
##########
@@ -245,6 +245,41 @@ public void testFindReadyWithRequestTimedOut() {
assertFalse(cache.isResponseExpected(otherNode, 1));
}
+ @Test
+ public void testHasRequestTimedOut() {
+ List<Node> bootstrapList = makeBootstrapList(2);
+ RequestManager cache = new RequestManager(
+ bootstrapList,
+ retryBackoffMs,
+ requestTimeoutMs,
+ random
+ );
+
+ // Find a ready node with the starting state
+ Node bootstrapNode1 = cache.findReadyBootstrapServer(time.milliseconds()).get();
+ assertTrue(
+ bootstrapList.contains(bootstrapNode1),
+ String.format("%s is not in %s", bootstrapNode1, bootstrapList)
+ );
+ // Before sending a request, no request should have timed out
+ assertFalse(cache.hasRequestTimedOut(bootstrapNode1, time.milliseconds()));
+
+ // Send a request
+ cache.onRequestSent(bootstrapNode1, 1, time.milliseconds());
+ assertEquals(
+ Optional.empty(),
+ cache.findReadyBootstrapServer(time.milliseconds())
+ );
+ assertFalse(cache.hasRequestTimedOut(bootstrapNode1, time.milliseconds()));
+
+ time.sleep(requestTimeoutMs / 2);
+ assertFalse(cache.hasRequestTimedOut(bootstrapNode1, time.milliseconds()));
+
+ // Timeout the request
+ time.sleep(requestTimeoutMs / 2);
Review Comment:
In integer division, `value / 2 + value / 2` does not equal `value`. You can see this when value is 3. `3/2` is `1` and `1 + 1` is not `3`. A more obvious sleep logic is:
```java
time.sleep(requestTimeoutMs - 1);
assertFalse(cache.hasRequestTimedOut(bootstrapNode1, time.milliseconds()));
// Timeout the request
time.sleep(1);
```
--
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-19354: KRaft observer should fetch from leader after rediscovery [kafka]
Posted by "ahuang98 (via GitHub)" <gi...@apache.org>.
ahuang98 commented on PR #19854:
URL: https://github.com/apache/kafka/pull/19854#issuecomment-3120475711
btw, build is green! I'm pushing a commit for a single test fix which I validated still passes locally
--
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-19354: KRaft observer should fetch from leader after rediscovery [kafka]
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on code in PR #19854:
URL: https://github.com/apache/kafka/pull/19854#discussion_r2121590408
##########
raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientTest.java:
##########
@@ -2145,7 +2186,20 @@ public void testObserverSendDiscoveryFetchAfterFetchTimeout(boolean withKip853Rp
fetchRequest = context.assertSentFetchRequest();
assertNotEquals(leaderId, fetchRequest.destination().id());
assertTrue(context.bootstrapIds.contains(fetchRequest.destination().id()));
+ assertTrue(context.client.quorum().followerStateOrThrow().hasFetchTimeoutExpired(context.time.milliseconds()));
context.assertFetchRequestData(fetchRequest, epoch, 0L, 0);
+
+ context.deliverResponse(
+ fetchRequest.correlationId(),
+ fetchRequest.destination(),
+ context.fetchResponse(epoch, leaderId, MemoryRecords.EMPTY, 0L, Errors.NOT_LEADER_OR_FOLLOWER)
+ );
+
+ context.client.poll();
+ context.pollUntilRequest();
+ fetchRequest = context.assertSentFetchRequest();
+ assertEquals(leaderId, fetchRequest.destination().id());
+ assertFalse(context.client.quorum().followerStateOrThrow().hasFetchTimeoutExpired(context.time.milliseconds()));
Review Comment:
Let's remove this line. This suite should test the protocol and not the internal implementation state.
--
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-19354: KRaft observer should fetch from leader after rediscovery [kafka]
Posted by "ahuang98 (via GitHub)" <gi...@apache.org>.
ahuang98 commented on code in PR #19854:
URL: https://github.com/apache/kafka/pull/19854#discussion_r2114996046
##########
raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientTest.java:
##########
@@ -2145,7 +2186,19 @@ public void testObserverSendDiscoveryFetchAfterFetchTimeout(boolean withKip853Rp
fetchRequest = context.assertSentFetchRequest();
assertNotEquals(leaderId, fetchRequest.destination().id());
assertTrue(context.bootstrapIds.contains(fetchRequest.destination().id()));
+ assertTrue(context.client.quorum().followerStateOrThrow().hasFetchTimeoutExpired(context.time.milliseconds()));
context.assertFetchRequestData(fetchRequest, epoch, 0L, 0);
+
+ context.deliverResponse(
+ fetchRequest.correlationId(),
+ fetchRequest.destination(),
+ context.fetchResponse(epoch, leaderId, MemoryRecords.EMPTY, 0L, Errors.NOT_LEADER_OR_FOLLOWER)
+ );
+
+ context.pollUntilRequest();
+ fetchRequest = context.assertSentFetchRequest();
+ assertEquals(leaderId, fetchRequest.destination().id());
Review Comment:
this check would fail without the fix
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19354: KRaft observer should fetch from leader after rediscovery [kafka]
Posted by "ahuang98 (via GitHub)" <gi...@apache.org>.
ahuang98 commented on code in PR #19854:
URL: https://github.com/apache/kafka/pull/19854#discussion_r2226443533
##########
raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientTest.java:
##########
@@ -2033,6 +2033,40 @@ public void testUnattachedAsVoterCanBecomeFollowerAfterFindingLeader(boolean wit
assertTrue(context.client.quorum().isFollower());
}
+ @ParameterizedTest
+ @ValueSource(booleans = { true, false })
+ public void testUnattachedWithLeaderCanBecomeFollowerAfterFindingLeader(boolean withKip853Rpc) throws Exception {
Review Comment:
this name was incorrect and should have been something like `testProspectiveWithLeaderCanBecomeFollower` - but in that case it already duplicates a test in KafkaRaftClientPreVoteTest so I'm deleting this
--
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