chia7712 commented on code in PR #23014:
URL: https://github.com/apache/kafka/pull/23014#discussion_r3774699031
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -1983,9 +1983,7 @@ private Fetch<K, V> pollForFetches(Timer timer) {
return fetch;
}
- long pollTimeout = isCommittedOffsetsManagementEnabled()
- ? Math.min(applicationEventHandler.maximumTimeToWait(),
timer.remainingMs())
- : timer.remainingMs();
+ long pollTimeout =
Math.min(applicationEventHandler.maximumTimeToWait(), timer.remainingMs());
// With the non-blocking poll design, it's possible that at this point
the background thread is
Review Comment:
Would you mind updating the comments around line 1996?
```java
if (partitions.isEmpty()) {
// If there aren't any assigned partitions, this could mean
that this consumer's group membership
// has not been established or assignments have been removed
and not yet reassigned. In either case,
// reduce the poll time for the fetch buffer wait.
pollTimeout = retryBackoffMs;
```
The reasons for empty partitions can also include DNS resolution failures.
##########
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
+ // can be made by consuming the buffered data. If no partitions
are fetchable at all (for example,
+ // no assignment yet, invalid positions, paused, or pending
revocation/callback), the state will
+ // not change until some external event occurs, so an immediate
wakeup would only busy-loop the
+ // caller rather than allowing the normal backoff to apply.
+ boolean canWakeBufferIfNoFetchRequestsToSend =
!subscriptions.fetchablePartitions(tp -> true).isEmpty();
+ return new FetchRequestPreparationResult(Collections.emptyMap(),
canWakeBufferIfNoFetchRequestsToSend);
Review Comment:
`Collections.emptyMap()` -> `Map.of`
--
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]
