m1a2st commented on code in PR #23014:
URL: https://github.com/apache/kafka/pull/23014#discussion_r3809283539
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -1985,29 +1985,27 @@ private Fetch<K, V> pollForFetches(Timer timer) {
return fetch;
}
- long pollTimeout = isCommittedOffsetsManagementEnabled()
- ? Math.min(applicationEventHandler.maximumTimeToWait(),
timer.remainingMs())
- : timer.remainingMs();
- // 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();
+ long pollTimeout =
Math.min(applicationEventHandler.maximumTimeToWait(), timer.remainingMs());
- 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.
+ // Bound the wait when background progress may make fetching possible
soon.
+ // Use the current application-thread state to avoid relying on stale
state from the network thread.
+ if (pollTimeout > retryBackoffMs) {
+ if (subscriptions.numAssignedPartitions() == 0) {
+ // If there are no assigned partitions, reduce the fetch
buffer wait time. This may happen when
+ // group membership has not been established yet, assignments
have been revoked but not reassigned,
+ // bootstrap DNS resolution is still in progress, or manual
assignment has not happened yet.
+ pollTimeout = retryBackoffMs;
+ } else if (!subscriptions.hasAllFetchPositions()) {
+ // If some partitions do not have valid positions, the
background thread may still be resolving them,
+ // for example by fetching committed offsets, looking up
offsets by timestamp, or backing off after a
+ // failure. Reduce the wait time so the application thread can
consume data promptly once positions are
+ // resolved.
+ pollTimeout = retryBackoffMs;
+ } else if (subscriptions.hasFetchablePartitions(tp ->
!fetchBuffer.bufferedPartitions().contains(tp))) {
Review Comment:
I’ve moved this check back to `FetchRequestManager.maximumTimeToWait()`,
where it can avoid applying the bound when in-flight requests already guarantee
a wakeup.
The application thread now only bounds the timeout for conditions it can
observe from fresh state: no assigned partitions, or positions that have not
been resolved yet.
update:
I initially moved this check to `FetchRequestManager.maximumTimeToWait()` to
avoid applying the bound when in-flight requests already guarantee a wakeup.
However, that introduced a stale-cache issue and caused a CI timeout.
The issue is that `maximumTimeToWait()` runs on the network thread, and its
result is cached. If the network thread observes that all fetchable partitions
are buffered and returns `Long.MAX_VALUE`, but the application thread then
consumes that data, the cached value becomes stale. Without the
application-thread guard, `pollTimeout` may use the stale `Long.MAX_VALUE`,
causing the application thread to block in `awaitWakeup()` until the next full
network round trip delivers a wakeup.
The application-thread check acts as a freshness guard: it re-evaluates the
buffer state right before blocking, so a stale `Long.MAX_VALUE` cache cannot
cause a long wait. The trade-off is that when in-flight requests cover all
unbuffered partitions, the application thread may still bound the wait to
`retryBackoffMs`.
--
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]
