lianetm commented on code in PR #23014:
URL: https://github.com/apache/kafka/pull/23014#discussion_r3758928516
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractFetch.java:
##########
@@ -298,6 +290,11 @@ public void handleCloseFetchSessionFailure(final Node
fetchTarget,
private void removePendingFetchRequest(Node fetchTarget, int sessionId) {
log.debug("Removing pending request for fetch session: {} for node:
{}", sessionId, fetchTarget);
nodesWithPendingFetchRequests.remove(fetchTarget.id());
+
+ // Wake the buffer whenever a node stops having a request in flight,
whatever the outcome was: data, an
+ // empty response, a fetch session error, or a failure. This ensures
the caller is not left waiting on a
+ // wakeup that only a completed request could have delivered.
+ fetchBuffer.wakeup();
Review Comment:
nice, this was a big gap I just realized (yesterday) that we had (not waking
up on all responses)
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractFetch.java:
##########
@@ -484,7 +488,48 @@ protected Map<Node, FetchSessionHandler.FetchRequestData>
prepareFetchRequests()
}
}
- return convert(fetchable);
+ // If every fetchable-but-unbuffered partition was skipped (for
example, due to reconnect backoff,
Review Comment:
uhm interesting, in this case, that we used to wake up before but won't now,
we leave the app thread to block, but how does that get bounded? I don't see
the fetchReqMgr overrides the `maximumTimeToWait` (should we?), so I guess
whatever other managers bring will win? (e.g, hb interval if in a group).
https://github.com/apache/kafka/blob/e2aa6d676f260d147302179a7a9cf3af369f01ad/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java#L1986-L1988
Getting that time boundary right is key here imo, we're moving from waking
up too much to just waking up in specific cases (challenge is to make sure we
don't leave the app thread blocking needlessly then, would probably turn into
fetch latency not generating next requests soon 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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
