kevin-wu24 commented on code in PR #22111:
URL: https://github.com/apache/kafka/pull/22111#discussion_r3317702727
##########
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java:
##########
@@ -1725,6 +1725,13 @@ private boolean handleFetchResponse(
leaderEndpoints = Endpoints.empty();
}
+ maybeSwitchObserverFetchToLeader(
+ responseEpoch,
+ responseLeaderId,
+ leaderEndpoints,
+ currentTimeMs
+ );
Review Comment:
> Can you document the exact RPC trace for this issue? Can you document
trace in trunk vs the trace in this branch? I am interested in the RPCs and
kraft state transitions.
Sure, the trace is below. It is a very esoteric edge case, so let me know if
you have any questions.
Assume that the observer is unable to establish a TCP connection with the
leader endpoint X. This means that `requestManager.hasRequestTimedOut` in
`maybeSendFetchToBestNode` will return false, since the raft client gets an
error response from its network client when it cannot connect to the request
destination node. Also, assume that on the subsequent `poll()`, the
`requestManager.isBackingOff` is also false. This is what I am referring to
when I say the fetching destination logic needs to be independent of the
request manager state.
The observer starts up either in the `Unattached` state or the
`FollowerState` with a known leader endpoint X at the current epoch.
On trunk:
- If the observer is in the first case, we enter `pollUnattachedCommon` to
fetch from bootstrap servers, and if there is a leader, we receive its endpoint
X and transition to `FollowerState`, so we are in the second case going forward.
- Observer tries to fetch from the leader going forward. Under our
assumptions above, `maybeSendFetchToBestNode` will always execute code in the
`else if (!requestManager.hasAnyInflightRequest)` branch, which sends a fetch
to the unreachable endpoint X.
- The observer is now unable to make progress so long as
`requestManager.isBackingOff` keeps returning false and endpoint X is
partitioned from the observer.
On this branch:
- If the observer is in the first case, we `pollUnattachedCommon` to fetch
from bootstrap servers, and if there is a leader, we receive its endpoint X and
transition to `FollowerState`, so we are in the second case going forward.
- If the observer is unable to fetch from the leader within the fetch
timeout, it transitions from `FollowerState(leader X) -> Unattached(leader X)`
in the same epoch, which allows it to fetch from the bootstrap servers.
- If the bootstrap servers are still in the same epoch with the same leader
X, their completed fetch response to the observer will transition the observer
from `Unattached(leader X) -> FollowerState(leader X)`. The observer will then
fetch from endpoint X.
- If there is a leader election and an epoch bump, the observer can discover
that from the bootstrap servers. The observer cannot discover this new state if
it is stuck fetching from X to which it cannot connect.
The motivation for the proposed behavior is that on `trunk`, the observer
can be "stuck" (depending on the request manager state) fetching from a
leader's endpoint which is unreachable. This "starves" the bootstrap endpoints
and prevents the observer from discovering a new leader or fetching until X
becomes reachable.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
