m1a2st commented on code in PR #23014:
URL: https://github.com/apache/kafka/pull/23014#discussion_r3763061740
##########
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:
Good point.
When the most recent `prepare()` produced no requests because every
candidate partition was skipped for a transient reason, such as reconnect
backoff or an in-flight request, `maximumTimeToWait()` now returns
`retryBackoffMs`. This bounds the application thread’s wait on the fetch
buffer, so it can re-evaluate fetch eligibility once the transient condition
may have cleared.
The other two empty-result paths do not need this fallback:
1. If the result is “safe to wake” because all fetchable partitions are
already buffered, we wake the buffer immediately.
2. If requests were actually sent, their completion, whether success,
failure, or session error, wakes the buffer via `removePendingFetchRequest`.
So the bounded wait only applies to the gap in between: there is nothing to
send right now, but the situation may change on its own over time.
--
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]
