Async acks: a re-delivered record after an eager rebalance can leave the consumer paused forever (ackInOrder deferred-drain IndexOutOfBoundsException)

[yalaka-surya]

Disclosure: this write-up was drafted with Claude (Anthropic's AI assistant) from our incident investigation. The source reading and the version history were checked against the spring-kafka repository, but I have not personally verified every step, so please treat the analysis as a hypothesis to be confirmed rather than an established finding.

Before I open an issue I would like a sanity check on whether this is a bug or a misuse on my side.

Setup

• spring-kafka 3.3.x; the code path discussed below is identical in 3.3.16 and in 4.1.1
• @KafkaListener method returning CompletableFuture<Void> (so asyncReplies is true)
• AckMode.MANUAL_IMMEDIATE, ContainerProperties.asyncAcks = true, no Acknowledgment parameter (the container acks on future completion)
• max.poll.records = 5, container concurrency = 1
• default partition.assignment.strategy (resolves to the eager RangeAssignor)
• processing per record takes seconds to minutes (DB writes)

What we observe

After a group membership change (a pod replaced), one container stays paused indefinitely on the main topic. It keeps polling and heartbeating, so it is never evicted, and the committed offset never moves. Lag grows on exactly that member's partitions until the next rebalance. Nothing is logged at any level. Pausing for incomplete async acks: {...} at DEBUG is the only visible trace, and it names offsets whose work has long finished.

What I think happens (read from the sources; the code is identical in 3.3.16 and 4.1.1)

The eager assignor revokes every partition on any join or leave. onPartitionsRevoked removes the partition from offsetsInThisBatch / deferredOffsets but does not cancel in-flight futures. When the partition is re-assigned to the same consumer, captureOffsets() rebuilds the lists from the committed offset, and the uncommitted records are re-delivered while their first copies are still running. So each offset now has two futures that will each call acknowledge().

ackInOrder() handles a duplicate of a lower offset (rejected as "acknowledging a stale record"), but a duplicate of a higher offset is appended to deferred with no check. The drain loop then assumes deferred never holds more entries than offs has remaining:

while (!ObjectUtils.isEmpty(deferred)
        && !containsUnackedOffsets(offs, recordToAck.offset(), deferred.get(0).offset())) {
    recordToAck = deferred.remove(0);
    offs.remove(0);          // throws once offs is already empty
}

Minimal sequence on one partition with offsets a < b < c, each with two in-flight copies:

ack arrives	ackInOrder	state after
c (copy 1)	above head, appended	deferred = [c]
c (copy 2)	above head, appended	deferred = [c, c]
a	head match; drain blocked because b sits between a and c	offs = [b, c], a committed
b	head match; drain pops first c (offs = []), loop continues, pops second c, offs.remove(0) on an empty list throws IndexOutOfBoundsException	offs = [] but the partition key stays in the map
remaining copies of a, b	IllegalStateException("Unexpected ack ...; offsets list is empty")	no effect

The exception is thrown before processAck(recordToAck) and before the offsetsInThisBatch.remove(part) cleanup, so:

• b and c are never committed (committed offset stays at a + 1),
• the partition is mapped to an empty list forever, so offsetsInThisBatch.isEmpty() is false and doResumeConsumerIfNeccessary() never clears pausedForAsyncAcks,
• the consumer keeps polling while paused, so it heartbeats and is never removed from the group.

Both exceptions surface inside the whenComplete in MessagingMessageListenerAdapter.handleResult() on a future nobody observes, which is why nothing is logged.

Two offsets are enough to trigger it (two copies of the higher one completing before the lower one). With several records in flight per partition and rebalances from routine pod churn, the ordering hits often enough for us to see it every few days.

Why I am not sure it is a bug

The ConsumerAcknowledgment carries only the record, so the container has no way to tell an ack from the previous assignment generation apart from one for the current one. Maybe the intended contract is that an application using async acks must itself cancel or suppress completions for revoked partitions (for example via onPartitionsRevokedBeforeCommit)? I could not find that stated in the reference docs, and the container's own pause/resume logic (repauseIfNeeded re-pausing after a rebalance) suggests the rebalance-while-paused case was considered.

Version notes

• 3.3.12 drain loop only popped when deferred.get(0).offset() == recordToAck.offset() + 1, so duplicates were harmless.
• Fix offset gap larger than 1 in case async ack #4264 (for Consumer was paused infinitely since difference in async ack offsets is larger than 1 #4258, backport Consumer was paused infinitely since difference in async ack offsets is larger than 1 #4269) removed that check because offset gaps can be larger than 1; Fix consumer pausing infinitely in async ack flow #4334 (for Listener with async acks pauses indefinitely #4332) added containsUnackedOffsets, which fixed the gap reported there but left the duplicate case.
• 3.3.16 and current main have the same loop.

Questions

1. Is a duplicate ack for a re-delivered record after an eager rebalance expected to be handled by the container, or is it the application's responsibility with async acks?
2. If it is the container's, would a fix along one of these lines be acceptable as a PR: (a) stamp acks with the assignment generation and drop stale ones, or (b) de-duplicate on deferred.add and guard the offs.remove(0) in the drain loop so a duplicate cannot leave the partition mapped to an empty list?
3. Should the failure at least be surfaced? Today an exception from acknowledge() inside handleResult's whenComplete is lost, which makes this very hard to diagnose in production.

I have a deterministic reproduction in our application's test suite (Testcontainers Kafka, a plain KafkaConsumer joining and leaving the group once while records are in flight) and can distil it into a standalone project if that helps.

[AZapata27]

Your reading of ackInOrder is correct, and the sequence reproduces. I turned it into a mock-consumer test in the style of ConcurrentMessageListenerContainerMockTests (eager revoke and re-assign of the same partition, acks old2, new2, new0, new1). On 4.1.1 the last ack throws IndexOutOfBoundsException from the drain loop, and the container is left with offsetsInThisBatch={foo-0=[]}, which keeps pausedForAsyncAcks set indefinitely.

One correction: main is no longer identical. This is the eager variant of #4687, fixed in 32334d6 ("Ignore stale async acks after rebalance"). ConsumerAcknowledgment now captures the pending-offset list of the poll that delivered the record, and acknowledge() drops the ack (DEBUG "Ignoring acknowledgment for an obsolete batch") when that list has since been replaced or removed by a revoke. That's basically your option (a), using list identity as the generation. The same test passes on main. The fix is in 4.2.0-M2 and backported to 4.1.x and 4.0.x (4.1.2 / 4.0.8). I don't see it on 3.3.x, which hasn't had fixes since 3.3.16.

So for your questions: the container now owns this, which makes (b) moot. (3) still stands. On main, handleResult rethrows an exception from acknowledge() inside a whenComplete whose future nobody observes, so I'd open a separate issue for it.

Until you can upgrade, be careful with the cooperative assignor as a workaround: it stops revoking retained partitions, but then you hit your own #4708 (fix pending in #4709). Cancelling the futures on revoke doesn't help on 3.3.x/4.1.1 either, because a cancelled future still goes through asyncFailure, which acknowledges. What does work on 3.3.x is acking yourself instead of returning the future, gated on a per-partition generation:

// generations: ConcurrentHashMap<TopicPartition, AtomicLong>, incremented in
// ConsumerAwareRebalanceListener.onPartitionsRevokedBeforeCommit while holding `lock`
@KafkaListener(...)  // void return, asyncAcks=true, MANUAL_IMMEDIATE
void listen(ConsumerRecord<K, V> rec, Acknowledgment ack) {
    TopicPartition tp = new TopicPartition(rec.topic(), rec.partition());
    long g = generations.computeIfAbsent(tp, k -> new AtomicLong()).get();
    process(rec).whenComplete((r, t) -> {
        synchronized (lock) {
            if (generations.get(tp).get() == g) {
                ack.acknowledge(); // skip acks from a previous assignment
            }
        }
    });
}

Static membership (group.instance.id with stable ids) cuts down the rebalances caused by pod replacement. An alert on a ConsumerPausedEvent with reason "Incomplete out of order acks" that isn't followed by a resume would have caught this early. Processing still has to be idempotent, because the pre-rebalance copies keep running alongside the redelivered ones even with the fix.

[AZapata27]

Standalone version of the repro, if it helps for an issue: https://github.com/AZapata27/spring-kafka-async-ack-rebalance-repro

Mocked consumer, no broker needed. mvn test fails on 4.1.1 with the IndexOutOfBoundsException and offsetsInThisBatch={foo-0=[]}; mvn test -Dspring-kafka.version=4.2.0-M2 passes.