MdTanwer opened a new pull request, #23228:
URL: https://github.com/apache/kafka/pull/23228
`AsyncKafkaConsumer.pollForFetches()` clamps its wait on the fetch buffer to
`retry.backoff.ms` whenever a fetchable partition has no buffered data. That
condition is true in the ordinary steady state of simply waiting for an
in-flight fetch response, where nothing can change until that response
arrives.
Combined with KAFKA-20780, which clears a completed inflight poll on every
iteration of the internal poll loop so a new `AsyncPollEvent` is submitted,
this makes the application and network threads cycle every
`retry.backoff.ms` for the whole `fetch.max.wait.ms` window. On each cycle
the application thread wakes with an empty buffer and submits a new poll
event, and the background thread re-runs the reconciliation check, position
validation and fetch request creation, only to find a request already in
flight and produce nothing. With default configs that is up to five wasted
round trips per fetch.
The clamp is also unnecessary. `FetchRequestManager.maximumTimeToWait()`
already returns `retryBackoffMs` when nothing is in flight, which covers
reconnect backoff, an unknown leader, and the other transient reasons a
partition may be skipped, and returns `Long.MAX_VALUE` while a request is in
flight, whose completion always wakes the buffer regardless of the outcome.
This PR drops the clamp and lets `maximumTimeToWait()` bound the wait.
Note this is separate from the fetch buffer wakeup spin reported in
KAFKA-20915, which was fixed as a duplicate of KAFKA-20854 in #23014. That
fix stopped `FetchRequestManager` from waking the buffer when it cannot
generate a request, but it also introduced the clamp removed here, so the
application thread still woke on the backoff interval while a fetch was
outstanding.
Testing:
- `AsyncKafkaConsumerTest.testPollDoesNotBoundWaitWhileFetchIsInFlight`
asserts the wait uses the full caller timeout when a fetch is in flight.
Against unmodified trunk it fails with `expected: <500> but was: <100>`,
which is the clamp firing.
- `FetchRequestManagerTest.testInflightFetchDoesNotWakeUpBuffer` guards the
behaviour this change depends on: an in-flight request must not wake the
buffer, while its completion must. This one already passes on trunk after
#23014 and is added to keep that contract covered.
- `AsyncKafkaConsumerTest`, `FetchRequestManagerTest`,
`ConsumerNetworkThreadTest` and `FetchBufferTest` pass, along with
`checkstyleMain`, `checkstyleTest` and `spotlessCheck`.
The scheduled `classic` versus `consumer` performance comparison that
reported the regression should be re-run to confirm the CPU numbers before
KAFKA-20904 is closed.
Made with [Cursor](https://cursor.com)
--
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]
