m1a2st commented on code in PR #23014:
URL: https://github.com/apache/kafka/pull/23014#discussion_r3763016872
##########
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:
I learned this point from your patch :)
--
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]
