KAFKA-20765: Fix OffsetFetch stale-epoch retry spinning in dedup loop

[aliehsaeedii]

Fixes KAFKA-20765.

The issue

In the async consumer (AsyncKafkaConsumer), when an in-flight
OffsetFetch fails with STALE_MEMBER_EPOCH and a refreshed member epoch
is already known, CommitRequestManager correctly classifies the error
as retriable — but the retry can enter a tight, non-progressing loop
instead of being re-sent:

1. Two callers fetch committed offsets for the same partitions. The
   second request is deduplicated by
   PendingRequests.addOffsetFetchRequest and chained onto the first
   (in-flight) one.
2. The in-flight request fails with STALE_MEMBER_EPOCH (e.g. another
   member joined and the coordinator bumped the group epoch while the
   request was in flight — routine under the KIP-1071 streams protocol
   during staggered startup).
3. The chained request's completion callback runs before the original
   request is removed from the in-flight buffer (CompletableFuture
   dependents run LIFO, and buffer removal is itself a completion
   callback). Its retry is handed back to addOffsetFetchRequest, which
   finds the original request still in the buffer — sameRequest()
   compares only the requested partitions — and chains the retry onto it.
4. But that request's future is already completed (with the same
   stale-epoch error), so chaining completes the retry's future immediately
   and synchronously. The error handler sees STALE_MEMBER_EPOCH again,
   retries again, chains again… a synchronous spin that only ends when the
   stack blows up.

Consequences (all observed in a system test run and reproduced in the
unit test):

• WARN "A duplicated, inflight, request was identified, but unable to find it in the outbound buffer" repeated hundreds of times within
  milliseconds,
• java.lang.StackOverflowError surfacing in SLF4J's toString() of
  OffsetFetchRequestState,
• no OffsetFetch with the new epoch ever goes on the wire from this path
  (the epoch is stamped at send time — the one stage the loop never
  reaches),
• the application-level future is never completed (orphaned).

In Kafka Streams with the streams rebalance protocol (KIP-1071) this
blocks the stream thread in committed-offset initialization for the full
default.api.timeout.ms, exceeds max.poll.interval.ms, and gets the
member evicted (observed in streams_broker_down_resilience_test with
staggered startup). Plain group.protocol=consumer apps are exposed to
the same spin; KIP-1071 just triggers the preconditions far more easily.

The fix

In PendingRequests.addOffsetFetchRequest, requests whose future is
already completed are no longer considered duplicates. A completed
request cannot deliver a result to anyone; it only appears in the
buffers transiently while its completion callbacks run. With the check
in place, the retry falls through to being enqueued and re-sent, picking
up the current member epoch at send time.

Also included:

• A unit test that deterministically reproduces the loop (fails without
  the fix with the exact production signatures) and verifies the retry is
  sent with the new epoch and completes both callers' futures.
• Log cleanup: the stale-epoch ERROR no longer claims the request
  "cannot be retried" when a retry is about to happen (only logged when
  the member has no epoch anymore), and the per-chained-caller
  "duplicated, inflight" WARN is now an accurate debug message (it fires
  for every deduplicated request, success or failure — expected, not an
  anomaly).

Testing

./gradlew :clients:test --tests "org.apache.kafka.clients.consumer.internals.CommitRequestManagerTest"
— all pass, including the new
testDuplicatedOffsetFetchFailsWithStaleEpochAndRetriesWithNewEpoch.

🤖 Generated with Claude Code

Reviewers: Lianet Magrans lmagrans@confluent.io

[lianetm]

Thanks @aliehsaeedii !

[lianetm]

dedup/chained it's certainly one of the reasons for this, but I'm not sure it's the only one. E.g, requests that remain unsent (not added to inflight) if we don't have a coordinator?. So maybe we can consider letting the log with what we're sure of (what it had before, dup found but not inflight), and we mention some of the potential reasons in a comment? wdyt?

[lianetm]

will be retried...if there's time left (worth clarifying, we cannot guarantee it will be retried at this point. The retry logic is elsewhere and based on expiration)

[lianetm]

uhm I think there is no backoff here (by the time it fails the new epoch is known already, so we just send it on the next poll). I expect the test should pass without the sleep and that we can remove the comment too.