fix: hold a commit that hits a transient KafkaError and back off its retries

AGENTS.md

@@ -7,7 +7,7 @@ This file provides guidance to Claude Code (claude.ai/code) when working with co
 `faststream-concurrent-aiokafka` gives FastStream's Kafka broker bounded concurrent message
 processing without giving up at-least-once delivery. [`CONTEXT.md`](CONTEXT.md) opens with what it
 does and owns the vocabulary — read it before naming a concept in code, a test name, or an issue
-title. Most of the vocabulary is FastStream's and aiokafka's; only six terms are this package's.
+title. Most of the vocabulary is FastStream's and aiokafka's; only seven terms are this package's.
 
 ## Commands
 

CONTEXT.md

@@ -15,7 +15,7 @@ Most of the vocabulary is **not this package's**. `broker`, `subscriber`, `middl
 `ack policy` and `ContextRepo` are [FastStream](https://faststream.ag2.ai/)'s; `consumer`,
 `consumer group`, `partition`, `offset`, `commit` and `rebalance` are Kafka's, reached through
 [aiokafka](https://aiokafka.readthedocs.io/). Those projects are the authority for all of them and
-nothing here redefines one. The six below are local to this package's concurrency model, and each
+nothing here redefines one. The seven below are local to this package's concurrency model, and each
 exists because getting the word wrong here has a concrete cost.
 
 **Handler**:
@@ -51,6 +51,13 @@ must not advance until a rebalance clears it. Distinct from the cancelled task i
 **hard boundary** within one commit round: the boundary stops that round, the watermark is what
 makes the stop survive later rounds.
 
+**Flush**:
+A period in which the committer commits every ready offset as soon as it can, instead of waiting
+for a full batch or the batch timeout. Three things open one: a rebalance revoking partitions
+(`commit_all`), backpressure at the uncommitted-task ceiling, and shutdown (`close`). A bare
+"flush" can mean any of the three; name the trigger when the difference matters. A rebalance flush
+ends when pending work drains or when every `commit_all` waiting on it gives up.
+
 **Control signal**:
 A member of FastStream's `IgnoredException` family — `AckMessage`, `RejectMessage`, `SkipMessage`,
 `NackMessage`, `StopConsume`, `StopApplication` — raised inside a dispatched user task. The family

docs/adr/0004-a-failed-commit-stays-pending.md

@@ -0,0 +1,15 @@
+# A commit that hits a transient error stays pending; it is never re-queued
+
+When `consumer.commit()` raises a transient `KafkaError`, the committer used to put the batch's
+tasks back onto its intake queue. That is the obvious retry, and it caused two opposite failures
+during a flush. If the re-queue emptied pending work, the flush ended, and the re-absorbed tasks
+waited a full batch timeout, so with the default `commit_batch_timeout_sec == flush_timeout_sec`
+the rebalance flush timed out and the offsets were redelivered. If other partitions still had
+work in flight, the flush stayed open and every re-put woke the loop to commit again: a hot retry
+loop with a traceback per attempt, outliving the `commit_all` that opened it. The failed
+`ReadyCommit` is now held inside `PendingCommits` and merged into the next `take_ready()`, so
+pending work never looks drained while a commit is outstanding, and nothing re-enters the queue
+to wake the loop; retries are spaced by a capped exponential backoff owned by the
+`CommitScheduler`. The rejected alternative was to keep re-queuing and add a retry-not-before
+guard: it patches both symptoms but leaves the root cause, the queue carrying work that was
+already absorbed, in place.

faststream_concurrent_aiokafka/_commit_scheduler.py

@@ -1,5 +1,7 @@
 import dataclasses
 
+from faststream_concurrent_aiokafka import consts
+
 
 @dataclasses.dataclass(frozen=True, slots=True)
 class Decision:
@@ -11,7 +13,8 @@ class Decision:
 class CommitScheduler:
     """Owns the streaming loop's when-to-commit decision state.
 
-    Manages the timeout deadline, the flush lifecycle, and the shutdown lifecycle.
+    Manages the timeout deadline, the flush lifecycle, the shutdown lifecycle,
+    and the backoff between commit rounds that hit a transient error.
 
     Synchronous, I/O-free, single-owner: the committer's async driver is the
     sole caller, on one asyncio task. Reads no clock and touches no asyncio
@@ -23,7 +26,9 @@ class CommitScheduler:
     Invariants:
       * pending empty ⇒ timeout_deadline is None.
       * flush_in_progress is set only when a flush fired without a stop request,
-        and cleared once pending drains.
+        and cleared once pending drains or the flush is released.
+      * retry_not_before is set only by a round that hit a transient error, and
+        cleared by the next round that did not; until it passes, no trigger commits.
       * should_shutdown is set only when a flush fired with a stop request; once
         set, is_finished() returns True as soon as pending drains.
     """
@@ -34,17 +39,31 @@ def __init__(self, *, commit_batch_size: int, commit_batch_timeout_sec: float) -
         self._timeout_deadline: float | None = None
         self._should_shutdown: bool = False
         self._flush_in_progress: bool = False
+        self._retry_attempts: int = 0
+        self._retry_not_before: float | None = None
+
+    @property
+    def retrying(self) -> bool:
+        return self._retry_attempts > 0
 
     def accepts_new_work(self) -> bool:
         # While shutting down, the driver stops pulling new items from the queue.
         return not self._should_shutdown
 
     def wait_timeout(self, now: float) -> float | None:
-        # Remaining time until the batch-timeout fires, for asyncio.wait. None when
-        # no deadline is armed (pending empty), so the select blocks until an event.
-        if self._timeout_deadline is None:
+        # Time until the batch-timeout fires or, during a flush, a backed-off retry is due.
+        # None when neither is armed, so the select blocks until an event.
+        wake_at = [] if self._timeout_deadline is None else [self._timeout_deadline]
+        urgent = self._flush_in_progress or self._should_shutdown
+        if urgent and self._retry_not_before is not None and self._retry_not_before > now:
+            wake_at.append(self._retry_not_before)
+        if not wake_at:
             return None
-        return max(self._timeout_deadline - now, 0.0)
+        return max(min(wake_at) - now, 0.0)
+
+    def release_flush(self) -> None:
+        # Fed before evaluate(), so a flush that fired since the release still opens.
+        self._flush_in_progress = False
 
     def evaluate(
         self,
@@ -69,7 +88,8 @@ def evaluate(
             else:
                 self._flush_in_progress = True
 
-        should_commit = (
+        backing_off = self._retry_not_before is not None and now < self._retry_not_before
+        should_commit = not backing_off and (
             pending_len >= self._batch_size or timeout_fired or self._flush_in_progress or self._should_shutdown
         )
         return Decision(
@@ -85,6 +105,7 @@ def note_committed(
         committed: bool,
         timeout_fired: bool,
         pending_empty: bool,
+        transient_error: bool,
     ) -> None:
         # An active commit_all (flush without stop) keeps committing until pending
         # drains; clear the flag once it does so messages_queue.join() can return.
@@ -94,6 +115,13 @@ def note_committed(
         # let it keep ticking. Invariant: pending empty ⇒ deadline None.
         if committed or timeout_fired:
             self._timeout_deadline = (now + self._batch_timeout) if not pending_empty else None
+        if transient_error:
+            self._retry_attempts += 1
+            delay = consts.COMMIT_RETRY_BACKOFF_BASE_SEC * 2 ** (self._retry_attempts - 1)
+            self._retry_not_before = now + min(delay, consts.COMMIT_RETRY_BACKOFF_MAX_SEC)
+        elif committed:
+            self._retry_attempts = 0
+            self._retry_not_before = None
 
     def is_finished(self, *, pending_empty: bool) -> bool:
         return self._should_shutdown and pending_empty

faststream_concurrent_aiokafka/_pending_state.py

@@ -20,8 +20,7 @@ class KafkaCommitTask:
 
 def insert_sorted(partition_pending: list[KafkaCommitTask], new_ct: KafkaCommitTask) -> None:
     # Common case: tasks arrive from the broker in offset order, so append is correct and
-    # the list stays sorted. Out-of-order arrivals only happen when _call_committer
-    # re-queues a batch on transient KafkaError; bisect handles the rare case in O(log N).
+    # the list stays sorted; bisect handles a rare out-of-order arrival in O(log N).
     if not partition_pending or partition_pending[-1].offset <= new_ct.offset:
         partition_pending.append(new_ct)
     else:
@@ -113,7 +112,7 @@ class ReadyCommit:
 
 
 class PendingCommits:
-    """Owns per-partition pending commit tasks, pending count, and cancellation watermarks.
+    """Owns per-partition pending commit tasks, held failed commits, pending count, and cancellation watermarks.
 
     Synchronous and single-owner: the committer's streaming loop is the sole
     mutator, so no locking is needed. Reads asyncio task state (done/cancelled)
@@ -124,6 +123,8 @@ def __init__(self) -> None:
         self._pending: dict[TopicPartition, list[KafkaCommitTask]] = {}
         self._count: int = 0
         self._watermarks: dict[tuple[int, TopicPartition], int] = {}
+        # Commits that hit a transient error; take_ready() returns them again, merged per consumer.
+        self._held: list[ReadyCommit] = []
 
     def __len__(self) -> int:
         return self._count
@@ -132,6 +133,10 @@ def absorb(self, ct: KafkaCommitTask) -> None:
         insert_sorted(self._pending.setdefault(ct.topic_partition, []), ct)
         self._count += 1
 
+    def hold(self, rc: ReadyCommit) -> None:
+        self._held.append(rc)
+        self._count += len(rc.tasks)
+
     def take_ready(self) -> list[ReadyCommit]:
         # Extract each partition's contiguous-done prefix (cancelled = hard
         # boundary), then group by consumer and apply the watermark floor.
@@ -140,16 +145,28 @@ def take_ready(self) -> list[ReadyCommit]:
         ready, ready_count = extract_ready_prefixes(self._pending)
         self._count -= ready_count
         flat: list[KafkaCommitTask] = [t for tasks in ready.values() for t in tasks]
-        if not flat:
-            return []
         by_consumer: dict[int, list[KafkaCommitTask]] = {}
         for task in flat:
             by_consumer.setdefault(id(task.consumer), []).append(task)
-        result: list[ReadyCommit] = []
+        fresh: list[ReadyCommit] = []
         for consumer_id, tasks in by_consumer.items():
             offsets = map_offsets_per_partition(consumer_id, tasks, self._watermarks)
-            result.append(ReadyCommit(consumer=tasks[0].consumer, offsets=offsets, tasks=tasks))
-        return result
+            fresh.append(ReadyCommit(consumer=tasks[0].consumer, offsets=offsets, tasks=tasks))
+        held, self._held = self._held, []
+        self._count -= sum(len(rc.tasks) for rc in held)
+        merged: dict[int, ReadyCommit] = {}
+        for rc in (*held, *fresh):
+            prior = merged.get(id(rc.consumer))
+            if prior is None:
+                merged[id(rc.consumer)] = rc
+                continue
+            offsets = dict(prior.offsets)
+            for partition, offset in rc.offsets.items():
+                offsets[partition] = max(offset, offsets.get(partition, offset))
+            merged[id(rc.consumer)] = ReadyCommit(
+                consumer=rc.consumer, offsets=offsets, tasks=[*prior.tasks, *rc.tasks]
+            )
+        return list(merged.values())
 
     def clear_watermarks(self, partitions: typing.Iterable[TopicPartition] | None = None) -> None:
         # Drops every consumer's floor for the given partitions, not just one owner's. This

faststream_concurrent_aiokafka/batch_committer.py

@@ -47,6 +47,9 @@ def __init__(
         # iteration. Fan-in cost is O(1) regardless of partition count or pending depth.
         self._task_completed_event = asyncio.Event()
         self._stop_requested: bool = False
+        # commit_all callers still awaiting the flush; when the last one gives up, the flush is released.
+        self._flush_waiters: int = 0
+        self._flush_released: bool = False
 
         self._shutdown_timeout = shutdown_timeout_sec
         # Owns per-partition pending commit tasks, count, and cancellation watermarks.
@@ -83,7 +86,7 @@ def _check_is_commit_task_running(self) -> None:
             msg: typing.Final = "Committer main task is not running"
             raise CommitterIsDeadError(msg)
 
-    async def _call_committer(self, rc: _pending_state.ReadyCommit) -> bool:
+    async def _call_committer(self, rc: _pending_state.ReadyCommit) -> bool | KafkaError:
         if not rc.offsets:
             return True
         assigned: typing.Final = rc.consumer.assignment()
@@ -103,29 +106,41 @@ async def _call_committer(self, rc: _pending_state.ReadyCommit) -> bool:
             # Partition no longer assigned (rebalance/revocation) — discard batch, not retryable
             logger.warning("Cannot commit due to partition loss or rebalancing, ignoring batch: %r", exc)
             return False
-        except KafkaError:
-            # Transient error — re-queue batch for retry on next cycle
-            logger.exception("Error during commit to kafka, re-queuing batch")
-            for task in rc.tasks:
-                self._uncommitted_count += 1
-                await self._messages_queue.put(task)
-            return False
+        except KafkaError as exc:
+            return exc
         else:
             return not revoked
 
     async def _commit_ready(self, ready_commits: list[_pending_state.ReadyCommit]) -> bool:
         # One commit per consumer, concurrently — each AIOKafkaConsumer commits its
-        # own partitions. task_done()/uncommitted_count balance the queue regardless
-        # of commit success (re-queued tasks are re-counted inside _call_committer).
+        # own partitions. A commit that hit a transient error stays pending, so its
+        # tasks are neither task_done() nor uncounted. Returns whether any did.
         results: typing.Final = await asyncio.gather(*(self._call_committer(rc) for rc in ready_commits))
-        committed_count = 0
-        for rc in ready_commits:
-            committed_count += len(rc.tasks)
+        transient_errors: typing.Final[list[KafkaError]] = []
+        finished_count = 0
+        for rc, result in zip(ready_commits, results, strict=True):
+            if isinstance(result, KafkaError):
+                transient_errors.append(result)
+                self._pending.hold(rc)
+                continue
+            finished_count += len(rc.tasks)
             for _ in rc.tasks:
                 self._messages_queue.task_done()
-        self._uncommitted_count -= committed_count
+        self._uncommitted_count -= finished_count
         self._uncommitted_drained.set()
-        return all(results)
+        self._log_transient_errors(transient_errors)
+        return bool(transient_errors)
+
+    def _log_transient_errors(self, errors: list[KafkaError]) -> None:
+        if not errors:
+            if self._scheduler.retrying:
+                logger.info("Commit to kafka succeeded after transient errors")
+            return
+        for index, exc in enumerate(errors):
+            if index == 0 and not self._scheduler.retrying:
+                logger.error("Error during commit to kafka, retrying the batch", exc_info=exc)
+            else:
+                logger.warning("Commit to kafka still failing, retrying the batch: %r", exc)
 
     async def _run_commit_process(self) -> None:
         tasks: typing.Final = _LoopTasks(
@@ -171,6 +186,10 @@ async def _streaming_iteration(self, tasks: "_LoopTasks") -> None:
 
         flush_fired: typing.Final = tasks.flush_wait_task.done()
 
+        if self._flush_released:
+            self._flush_released = False
+            self._scheduler.release_flush()
+
         decision: typing.Final = self._scheduler.evaluate(
             now=now,
             absorbed=absorbed,
@@ -183,17 +202,19 @@ async def _streaming_iteration(self, tasks: "_LoopTasks") -> None:
             self._handle_flush_fired(tasks, drain_queue=decision.drain_queue_now)
 
         committed = False
+        transient_error = False
         if decision.should_commit:
             ready = self._pending.take_ready()
             if ready:
-                await self._commit_ready(ready)
+                transient_error = await self._commit_ready(ready)
                 committed = True
 
         self._scheduler.note_committed(
             now=loop.time(),
             committed=committed,
             timeout_fired=decision.timeout_fired,
             pending_empty=not self._pending,
+            transient_error=transient_error,
         )
 
     def _handle_flush_fired(self, tasks: "_LoopTasks", *, drain_queue: bool) -> None:
@@ -224,15 +245,23 @@ async def commit_all(self, flush_timeout_sec: float = consts.DEFAULT_REBALANCE_F
         reassignment — at-least-once). Safe to call during rebalance
         (on_partitions_revoked); the committer keeps running after this returns.
         """
+        self._flush_waiters += 1
+        self._flush_released = False
         self._flush_batch_event.set()
+        drained = False
         try:
             await asyncio.wait_for(self._messages_queue.join(), timeout=flush_timeout_sec)
+            drained = True
         except TimeoutError:
             logger.warning(
                 "Kafka middleware. commit_all flush timed out after %.1fs; "
                 "in-flight offsets will be redelivered on restart/reassignment",
                 flush_timeout_sec,
             )
+        finally:
+            self._flush_waiters -= 1
+            if not drained and self._flush_waiters == 0:
+                self._flush_released = True
 
     def clear_cancellation_watermarks(self, partitions: typing.Iterable[TopicPartition] | None = None) -> None:
         """Forget cancellation watermarks for ``partitions`` (or all if ``None``).

faststream_concurrent_aiokafka/consts.py

@@ -8,3 +8,5 @@
 DEFAULT_SHUTDOWN_TIMEOUT_SEC: typing.Final = 20.0
 PROCESSING_CONTEXT_KEY: typing.Final = "concurrent_processing"
 DEFAULT_MAX_UNCOMMITTED_TASKS: typing.Final = 10_000
+COMMIT_RETRY_BACKOFF_BASE_SEC: typing.Final = 0.1
+COMMIT_RETRY_BACKOFF_MAX_SEC: typing.Final = 2.0