KAFKA-20904: Don't bound the poll wait while a fetch is in flight

[MdTanwer]

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
  KAFKA-20854 A more obvious busy loop due to KIP-909 #23014 and is added to keep that contract covered.
• AsyncKafkaConsumerTest, FetchRequestManagerTest,
  ConsumerNetworkThreadTest and FetchBufferTest pass, along with
  checkstyleMain, checkstyleTest and spotlessCheck.

Reviewers: Ken Huang s7133700@gmail.com

[m1a2st]

Thanks for this great patch.

With this patch, FetchRequestManager.maximumTimeToWait() returns retryBackoffMs whenever there is no in-flight fetch. This means the application thread may wake every retryBackoffMs even when all fetchable partitions are already buffered, where no useful progress can happen until the application drains the buffer.

The removed application-thread guard used to avoid this extra clamp. Should maximumTimeToWait() handle that case now?

I tried that in an earlier revision of #23014, but reverted it because it caused a stale-cache hang: maximumTimeToWait() is computed on the network thread and cached, while the application thread may drain the buffer before reading it. A cached Long.MAX_VALUE can then become stale, and with no in-flight request, nothing wakes the buffer until timer.remainingMs() expires.

The current simple retryBackoffMs constant avoids this stale-cache issue by not depending on buffer state at all, which I think is the right trade-off. I just want to confirm: is this intentional, and are we comfortable with the conservative behavior of waking every retryBackoffMs even when we could otherwise safely sleep longer?