m1a2st commented on code in PR #23014:
URL: https://github.com/apache/kafka/pull/23014#discussion_r3780821854
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractFetch.java:
##########
@@ -433,9 +432,14 @@ protected Map<Node, FetchSessionHandler.FetchRequestData>
prepareFetchRequests()
List<TopicPartition> unbuffered = fetchablePartitions(buffered);
if (unbuffered.isEmpty()) {
- // If there are no partitions that don't already have data locally
buffered, there's no need to issue
- // any fetch requests at the present time.
- return Collections.emptyMap();
+ // If every currently fetchable partition already has buffered
data, there is no need to issue
+ // additional fetch requests. This is a safe point to wake the
buffer immediately because progress
Review Comment:
If we also used `fetchablePartitions` in `pollForFetches`, we would shorten
the timeout in cases such as “all partitions are paused” or “all partitions are
pending revocation.” That would be undesirable:
- Paused: only the application thread calling `resume()` can change this
state, so periodic short-timeout wakeups would just waste CPU.
- Pending revocation: this is already bounded by the heartbeat manager’s
`maximumTimeToWait()`.
So `assignedPartitions().isEmpty()` is the right check in `pollForFetches`.
It only applies the shorter timeout when membership itself has not yet been
established, which is the only case where the application thread needs to react
promptly to a background event, such as an assignment arriving.
--
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]
