chia7712 commented on code in PR #23014:
URL: https://github.com/apache/kafka/pull/23014#discussion_r3774795872
##########
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:
This comment is interesting. We have a similar consideration when
calculating the suitable poll timeout. Should we apply the same logic there?
```java
// With the non-blocking poll design, it's possible that at this
point the background thread is
// concurrently working to update positions. Therefore, a _copy_ of
the current assignment is retrieved
// and iterated looking for any partitions with invalid positions.
This is done to avoid being stuck
// in poll for an unnecessarily long amount of time if we are
missing some positions since the offset
// lookup may be backing off after a failure.
if (pollTimeout > retryBackoffMs) {
Set<TopicPartition> partitions =
subscriptions.assignedPartitions(); // this one
```
--
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]
