From aed971d8457d32857497898ea15f1c31968269a3 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Sun, 4 Oct 2026 11:44:30 +0300 Subject: [PATCH] fix: hold a commit that hits a transient KafkaError and back off its retries A transient commit error re-queued the batch. That either ended the flush (commit_all timed out at the default 10s/10s and offsets were redelivered), hot-looped commit() with a traceback per attempt while other partitions had work in flight, or, during close(), left the offsets uncommitted. The failed commit now stays in PendingCommits and merges into the next take_ready(); CommitScheduler spaces retries 0.1s doubling to 2s; a flush is released once every commit_all waiting on it has given up; a streak logs one ERROR, then WARNINGs, then INFO on recovery. Closes #62 --- AGENTS.md | 2 +- CONTEXT.md | 9 +- .../adr/0004-a-failed-commit-stays-pending.md | 15 ++ .../_commit_scheduler.py | 42 +++- .../_pending_state.py | 33 ++- .../batch_committer.py | 61 ++++-- faststream_concurrent_aiokafka/consts.py | 2 + tests/test_commit_scheduler.py | 86 +++++++- tests/test_kafka_committer.py | 193 +++++++++++++++--- tests/test_pending_commits.py | 51 +++++ 10 files changed, 430 insertions(+), 64 deletions(-) create mode 100644 docs/adr/0004-a-failed-commit-stays-pending.md diff --git a/AGENTS.md b/AGENTS.md index 5773ebd..0af554d 100644 --- a/AGENTS.md +++ b/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 diff --git a/CONTEXT.md b/CONTEXT.md index 38bfc3d..449ea8e 100644 --- a/CONTEXT.md +++ b/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 diff --git a/docs/adr/0004-a-failed-commit-stays-pending.md b/docs/adr/0004-a-failed-commit-stays-pending.md new file mode 100644 index 0000000..68ee78a --- /dev/null +++ b/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. diff --git a/faststream_concurrent_aiokafka/_commit_scheduler.py b/faststream_concurrent_aiokafka/_commit_scheduler.py index a6e7f86..281b4d8 100644 --- a/faststream_concurrent_aiokafka/_commit_scheduler.py +++ b/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 diff --git a/faststream_concurrent_aiokafka/_pending_state.py b/faststream_concurrent_aiokafka/_pending_state.py index e8c036c..81fa579 100644 --- a/faststream_concurrent_aiokafka/_pending_state.py +++ b/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 diff --git a/faststream_concurrent_aiokafka/batch_committer.py b/faststream_concurrent_aiokafka/batch_committer.py index 6471b46..e5ffe16 100644 --- a/faststream_concurrent_aiokafka/batch_committer.py +++ b/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,10 +202,11 @@ 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( @@ -194,6 +214,7 @@ async def _streaming_iteration(self, tasks: "_LoopTasks") -> None: 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``). diff --git a/faststream_concurrent_aiokafka/consts.py b/faststream_concurrent_aiokafka/consts.py index 148908a..406c2c7 100644 --- a/faststream_concurrent_aiokafka/consts.py +++ b/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 diff --git a/tests/test_commit_scheduler.py b/tests/test_commit_scheduler.py index 583847e..cb5ca24 100644 --- a/tests/test_commit_scheduler.py +++ b/tests/test_commit_scheduler.py @@ -2,6 +2,8 @@ import inspect import pathlib +import pytest + from faststream_concurrent_aiokafka._commit_scheduler import CommitScheduler @@ -58,7 +60,7 @@ def test_flush_without_stop_commits_until_pending_drains() -> None: assert d.drain_queue_now is False d2 = s.evaluate(now=101.0, absorbed=False, flush_fired=False, stop_requested=False, pending_len=1) assert d2.should_commit is True # keeps committing while flush_in_progress - s.note_committed(now=102.0, committed=True, timeout_fired=False, pending_empty=True) + s.note_committed(now=102.0, committed=True, timeout_fired=False, pending_empty=True, transient_error=False) d3 = s.evaluate(now=103.0, absorbed=False, flush_fired=False, stop_requested=False, pending_len=0) assert d3.should_commit is False # flag cleared once pending drained @@ -76,28 +78,28 @@ def test_flush_with_stop_sets_shutdown_and_drain() -> None: def test_deadline_reset_keeps_ticking_when_pending_remains() -> None: s = _sched(timeout=10.0) s.evaluate(now=100.0, absorbed=True, flush_fired=False, stop_requested=False, pending_len=5) - s.note_committed(now=104.0, committed=True, timeout_fired=False, pending_empty=False) + s.note_committed(now=104.0, committed=True, timeout_fired=False, pending_empty=False, transient_error=False) assert s.wait_timeout(now=104.0) == 10.0 # re-armed at fresh now + timeout def test_deadline_cleared_when_pending_drains() -> None: s = _sched(timeout=10.0) s.evaluate(now=100.0, absorbed=True, flush_fired=False, stop_requested=False, pending_len=1) - s.note_committed(now=104.0, committed=True, timeout_fired=False, pending_empty=True) + s.note_committed(now=104.0, committed=True, timeout_fired=False, pending_empty=True, transient_error=False) assert s.wait_timeout(now=104.0) is None # invariant: pending empty ⇒ no deadline def test_note_committed_resets_on_timeout_even_without_commit() -> None: s = _sched(timeout=10.0) s.evaluate(now=100.0, absorbed=True, flush_fired=False, stop_requested=False, pending_len=1) - s.note_committed(now=110.0, committed=False, timeout_fired=True, pending_empty=False) + s.note_committed(now=110.0, committed=False, timeout_fired=True, pending_empty=False, transient_error=False) assert s.wait_timeout(now=110.0) == 10.0 def test_note_committed_no_reset_when_neither_committed_nor_timeout() -> None: s = _sched(timeout=10.0) s.evaluate(now=100.0, absorbed=True, flush_fired=False, stop_requested=False, pending_len=1) - s.note_committed(now=103.0, committed=False, timeout_fired=False, pending_empty=False) + s.note_committed(now=103.0, committed=False, timeout_fired=False, pending_empty=False, transient_error=False) assert s.wait_timeout(now=103.0) == 7.0 # deadline left ticking, not reset @@ -108,6 +110,80 @@ def test_no_trigger_when_idle_below_batch() -> None: assert d.should_commit is False +def _flush(s: CommitScheduler, *, now: float, released: bool = False, fired: bool = False) -> bool: + if released: + s.release_flush() + return s.evaluate(now=now, absorbed=False, flush_fired=fired, stop_requested=False, pending_len=1).should_commit + + +def _fail(s: CommitScheduler, *, now: float) -> None: + s.note_committed(now=now, committed=True, timeout_fired=False, pending_empty=False, transient_error=True) + + +def test_transient_error_backs_off_the_next_flush_commit() -> None: + s = _sched() + assert _flush(s, now=100.0, fired=True) is True + _fail(s, now=100.0) + assert _flush(s, now=100.05) is False + assert s.wait_timeout(now=100.05) == pytest.approx(0.05) + assert _flush(s, now=100.1) is True + + +def test_backoff_doubles_up_to_the_cap() -> None: + s = _sched() + _flush(s, now=0.0, fired=True) + delays = [] + now = 0.0 + for _ in range(7): + _fail(s, now=now) + delay = s.wait_timeout(now=now) + assert delay is not None + delays.append(delay) + now += delay + assert delays == pytest.approx([0.1, 0.2, 0.4, 0.8, 1.6, 2.0, 2.0]) + + +def test_a_round_without_transient_error_resets_the_backoff() -> None: + s = _sched() + _flush(s, now=100.0, fired=True) + _fail(s, now=100.0) + _fail(s, now=100.1) + s.note_committed(now=100.3, committed=True, timeout_fired=False, pending_empty=False, transient_error=False) + assert _flush(s, now=100.3) is True + _fail(s, now=100.3) + assert s.wait_timeout(now=100.3) == pytest.approx(0.1) + + +def test_backoff_gates_the_shutdown_flush() -> None: + s = _sched() + d = s.evaluate(now=100.0, absorbed=False, flush_fired=True, stop_requested=True, pending_len=1) + assert d.should_commit is True + _fail(s, now=100.0) + d = s.evaluate(now=100.05, absorbed=False, flush_fired=False, stop_requested=True, pending_len=1) + assert d.should_commit is False + assert s.wait_timeout(now=100.05) == pytest.approx(0.05) + + +def test_retry_time_does_not_shorten_the_wait_outside_a_flush() -> None: + s = _sched(timeout=10.0) + s.evaluate(now=100.0, absorbed=True, flush_fired=False, stop_requested=False, pending_len=1) + _fail(s, now=110.0) + assert s.wait_timeout(now=110.0) == 10.0 + + +def test_released_flush_stops_committing() -> None: + s = _sched() + _flush(s, now=100.0, fired=True) + assert _flush(s, now=101.0, released=True) is False + assert _flush(s, now=102.0) is False + + +def test_flush_fired_with_release_reopens_the_flush() -> None: + s = _sched() + _flush(s, now=100.0, fired=True) + assert _flush(s, now=101.0, released=True, fired=True) is True + + def test_the_commit_scheduler_reads_no_clock_and_touches_no_asyncio() -> None: """INVARIANT: `_commit_scheduler` decides synchronously, from arguments alone. diff --git a/tests/test_kafka_committer.py b/tests/test_kafka_committer.py index 0be32af..f9b9787 100644 --- a/tests/test_kafka_committer.py +++ b/tests/test_kafka_committer.py @@ -217,10 +217,10 @@ async def test_committer_commits_to_kafka(committer: KafkaBatchCommitter, mock_c assert call_args == partitions_to_offsets -async def test_committer_retries_on_kafka_error( +async def test_call_committer_returns_a_transient_error( committer: KafkaBatchCommitter, mock_consumer: MockAIOKafkaConsumer ) -> None: - """KafkaError re-queues the batch for retry on the next cycle.""" + """KafkaError is handed back to _commit_ready, which holds the batch; nothing is re-queued.""" mock_task: typing.Final = MockAsyncioTask(result="success") sample_task: typing.Final = KafkaCommitTask( asyncio_task=mock_task, # ty: ignore[invalid-argument-type] @@ -228,22 +228,21 @@ async def test_committer_retries_on_kafka_error( consumer=mock_consumer, topic_partition=TopicPartition(topic="test-topic", partition=0), ) - mock_consumer.commit.side_effect = KafkaError("transient broker error") + error: typing.Final = KafkaError("transient broker error") + mock_consumer.commit.side_effect = error partitions_to_offsets: typing.Final = {sample_task.topic_partition: 101} rc = _pending_state.ReadyCommit(consumer=mock_consumer, offsets=partitions_to_offsets, tasks=[sample_task]) result: typing.Final = await committer._call_committer(rc) - assert result is False - assert not committer._messages_queue.empty() - requeued_task: typing.Final = await committer._messages_queue.get() - assert requeued_task == sample_task + assert result is error + assert committer._messages_queue.empty() async def test_committer_ignores_commit_failed_error( committer: KafkaBatchCommitter, mock_consumer: MockAIOKafkaConsumer, sample_task: KafkaCommitTask ) -> None: - """CommitFailedError (rebalance in progress) is silently ignored — no re-queue.""" + """CommitFailedError (rebalance in progress) is discarded, not held for retry.""" mock_consumer.commit.side_effect = CommitFailedError() partitions_to_offsets: typing.Final = {sample_task.topic_partition: 101} rc = _pending_state.ReadyCommit(consumer=mock_consumer, offsets=partitions_to_offsets, tasks=[sample_task]) @@ -642,7 +641,7 @@ def test_insert_sorted_appends_in_order(mock_consumer: MockAIOKafkaConsumer) -> def test_insert_sorted_bisects_out_of_order(mock_consumer: MockAIOKafkaConsumer) -> None: - """A re-queued task with a lower offset slides into the right position.""" + """An out-of-order task with a lower offset slides into the right position.""" tp: typing.Final = TopicPartition(topic="t", partition=0) pending: list[KafkaCommitTask] = [] for offset in (10, 11): @@ -695,9 +694,8 @@ async def test_commit_ready_calls_commit_per_partition_max( offsets=_pending_state.map_offsets_per_partition(id(mock_consumer), tasks, {}), tasks=tasks, ) - result: typing.Final = await committer._commit_ready([rc]) + await committer._commit_ready([rc]) - assert result is True mock_consumer.commit.assert_called_once_with({tp: expected_offset + 2}) @@ -781,10 +779,10 @@ async def test_commit_ready_handles_multiple_consumers(committer: KafkaBatchComm consumer_b.commit.assert_called_once_with({tp_b: 21}) -async def test_commit_ready_returns_false_on_commit_failure( +async def test_commit_ready_discarded_commit_is_not_a_transient_error( committer: KafkaBatchCommitter, mock_consumer: MockAIOKafkaConsumer ) -> None: - """_commit_ready returns False when _call_committer fails.""" + """A commit _call_committer discards is finished, not held for retry.""" task: typing.Final = MockAsyncioTask(result="ok") tp: typing.Final = TopicPartition(topic="t", partition=0) commit_task: typing.Final = KafkaCommitTask( @@ -801,12 +799,13 @@ async def test_commit_ready_returns_false_on_commit_failure( result: typing.Final = await committer._commit_ready([rc]) assert result is False + assert len(committer._pending) == 0 -async def test_commit_ready_returns_false_if_any_consumer_group_fails( +async def test_commit_ready_holds_only_the_consumer_that_hit_a_transient_error( committer: KafkaBatchCommitter, ) -> None: - """If any consumer's commit slice fails, the overall return is False.""" + """One consumer's transient error holds its commit; the other consumer still commits.""" consumer_a: typing.Final = MockAIOKafkaConsumer() consumer_b: typing.Final = MockAIOKafkaConsumer() consumer_a.commit.side_effect = KafkaError("transient") # consumer_a fails @@ -842,8 +841,9 @@ async def test_commit_ready_returns_false_if_any_consumer_group_fails( ) result: typing.Final = await committer._commit_ready([rc_a, rc_b]) - assert result is False - consumer_b.commit.assert_called_once() # b still committed independently + assert result is True + consumer_b.commit.assert_called_once() + assert committer._pending.take_ready() == [rc_a] async def test_commit_ready_cancelled_task_not_logged_as_error( @@ -1223,14 +1223,13 @@ async def hangs() -> None: await committer.close() -async def test_committer_streaming_handles_requeue_offset_order() -> None: - """Lazy offset sort tolerates re-queued tasks landing after higher-offset arrivals. +async def test_committer_streaming_merges_held_batch_with_later_offsets() -> None: + """A batch held after a transient KafkaError merges with higher-offset arrivals. - Transient KafkaError re-queues a batch; meanwhile new same-partition tasks arrive - with higher offsets. The final commit must reflect the correct max offset. + The final commit must reflect the max offset across the held batch and the new tasks. """ consumer: typing.Final = MockAIOKafkaConsumer() - # First commit attempt: transient KafkaError → re-queue. Second attempt: succeeds. + # First commit attempt: transient KafkaError → held. Second attempt: succeeds. consumer.commit.side_effect = [KafkaError("transient"), None, None] committer: typing.Final = KafkaBatchCommitter(commit_batch_timeout_sec=10.0, commit_batch_size=2) @@ -1253,15 +1252,15 @@ async def quick() -> None: ) ) - # Wait until the failing commit attempt has occurred and the batch was re-queued. + # Wait until the failing commit attempt has occurred and the batch is held. await _drive_until(lambda: consumer.commit.call_count >= 1, deadline_sec=1.0) - # Now send a task with a higher offset BEFORE the re-queued tasks land back in pending. + # Now send a task with a higher offset while the batch is held. late_task: typing.Final = asyncio.create_task(quick()) await committer.send_task( KafkaCommitTask( asyncio_task=late_task, - offset=200, # much higher than the re-queued 100/101 + offset=200, # much higher than the held 100/101 consumer=consumer, topic_partition=tp, ) @@ -1270,7 +1269,7 @@ async def quick() -> None: await committer.close() # The final commit must reflect the max processed offset (200 → next-to-fetch 201) - # despite the requeued 100/101 arriving after offset 200 in queue order. + # merged with the held 100/101. final_call: typing.Final = consumer.commit.call_args_list[-1] assert final_call.args[0] == {tp: 201} @@ -1552,3 +1551,145 @@ async def test_committer_cancellation_is_not_reported_as_a_death(caplog: pytest. await asyncio.sleep(0) assert not _death_records(caplog) + + +# ---------- transient commit errors ---------- + + +async def _never_finishes() -> None: + await asyncio.Event().wait() + + +async def _finishes() -> None: + return None + + +def _commit_task(consumer: MockAIOKafkaConsumer, partition: int, task: asyncio.Task[None]) -> KafkaCommitTask: + return KafkaCommitTask( + asyncio_task=task, offset=100, consumer=consumer, topic_partition=TopicPartition(topic="t", partition=partition) + ) + + +async def _cancel_all(committer: KafkaBatchCommitter, *tasks: asyncio.Task[typing.Any]) -> None: + assert committer._commit_task is not None + for task in (committer._commit_task, *tasks): + task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await task + + +async def test_commit_all_retries_a_transient_error_within_the_flush_timeout(caplog: pytest.LogCaptureFixture) -> None: + consumer: typing.Final = MockAIOKafkaConsumer() + consumer.commit.side_effect = [KafkaError("blip"), None] + committer: typing.Final = KafkaBatchCommitter(commit_batch_timeout_sec=0.5, commit_batch_size=100) + committer.spawn() + await committer.send_task(_commit_task(consumer, 0, asyncio.create_task(_finishes()))) + + await committer.commit_all(flush_timeout_sec=0.5) + + assert "commit_all flush timed out" not in caplog.text + assert consumer.commit.call_args_list[-1].args[0] == {TopicPartition(topic="t", partition=0): 101} + await committer.close() + + +async def test_commit_retries_back_off_during_a_flush() -> None: + consumer: typing.Final = MockAIOKafkaConsumer() + consumer.commit.side_effect = KafkaError("outage") + committer: typing.Final = KafkaBatchCommitter(commit_batch_timeout_sec=10.0, commit_batch_size=100) + committer.spawn() + in_flight: typing.Final = asyncio.create_task(_never_finishes()) + await committer.send_task(_commit_task(consumer, 0, asyncio.create_task(_finishes()))) + await committer.send_task(_commit_task(consumer, 1, in_flight)) + await _drive_until(lambda: len(committer._pending) == 2) + + await committer.commit_all(flush_timeout_sec=0.5) + + assert consumer.commit.call_count <= 4 # attempts at about 0, 0.1 and 0.3s + await _cancel_all(committer, in_flight) + + +async def test_flush_urgency_ends_when_commit_all_times_out() -> None: + consumer: typing.Final = MockAIOKafkaConsumer() + consumer.commit.side_effect = KafkaError("outage") + committer: typing.Final = KafkaBatchCommitter(commit_batch_timeout_sec=10.0, commit_batch_size=100) + committer.spawn() + in_flight: typing.Final = asyncio.create_task(_never_finishes()) + await committer.send_task(_commit_task(consumer, 0, asyncio.create_task(_finishes()))) + await committer.send_task(_commit_task(consumer, 1, in_flight)) + await _drive_until(lambda: len(committer._pending) == 2) + + await committer.commit_all(flush_timeout_sec=0.25) + calls_when_flush_ended: typing.Final = consumer.commit.call_count + await asyncio.sleep(0.5) + + assert consumer.commit.call_count == calls_when_flush_ended + await _cancel_all(committer, in_flight) + + +async def test_one_commit_all_timing_out_keeps_the_flush_for_another_waiter() -> None: + consumer: typing.Final = MockAIOKafkaConsumer() + consumer.commit.side_effect = [KafkaError("blip"), None] + committer: typing.Final = KafkaBatchCommitter(commit_batch_timeout_sec=10.0, commit_batch_size=100) + committer.spawn() + in_flight: typing.Final = asyncio.create_task(_never_finishes()) + await committer.send_task(_commit_task(consumer, 0, asyncio.create_task(_finishes()))) + await committer.send_task(_commit_task(consumer, 1, in_flight)) + await _drive_until(lambda: len(committer._pending) == 2) + + long_wait: typing.Final = asyncio.create_task(committer.commit_all(flush_timeout_sec=1.0)) + await committer.commit_all(flush_timeout_sec=0.05) + await _drive_until(lambda: consumer.commit.call_count >= 2, deadline_sec=0.5) + + assert consumer.commit.call_args_list[1].args[0] == {TopicPartition(topic="t", partition=0): 101} + await _cancel_all(committer, in_flight, long_wait) + + +async def test_close_retries_a_transient_error_before_exiting() -> None: + consumer: typing.Final = MockAIOKafkaConsumer() + consumer.commit.side_effect = [KafkaError("blip"), None] + committer: typing.Final = KafkaBatchCommitter(commit_batch_timeout_sec=10.0, commit_batch_size=100) + committer.spawn() + await committer.send_task(_commit_task(consumer, 0, asyncio.create_task(_finishes()))) + + await committer.close() + + assert consumer.commit.call_count == 2 + assert consumer.commit.call_args_list[-1].args[0] == {TopicPartition(topic="t", partition=0): 101} + + +async def test_transient_error_outside_a_flush_retries_after_the_batch_timeout() -> None: + consumer: typing.Final = MockAIOKafkaConsumer() + consumer.commit.side_effect = [KafkaError("blip"), None] + committer: typing.Final = KafkaBatchCommitter(commit_batch_timeout_sec=0.2, commit_batch_size=100) + committer.spawn() + loop: typing.Final = asyncio.get_running_loop() + await committer.send_task(_commit_task(consumer, 0, asyncio.create_task(_finishes()))) + + await _drive_until(lambda: consumer.commit.call_count >= 1) + failed_at: typing.Final = loop.time() + await _drive_until(lambda: consumer.commit.call_count >= 2) + + assert loop.time() - failed_at >= 0.15 + await committer.close() + + +async def test_a_streak_of_transient_errors_logs_one_error(caplog: pytest.LogCaptureFixture) -> None: + consumer: typing.Final = MockAIOKafkaConsumer() + consumer.commit.side_effect = [KafkaError("a"), KafkaError("b"), KafkaError("c"), None] + committer: typing.Final = KafkaBatchCommitter(commit_batch_timeout_sec=10.0, commit_batch_size=100) + committer.spawn() + await committer.send_task(_commit_task(consumer, 0, asyncio.create_task(_finishes()))) + + with caplog.at_level(logging.INFO, logger="faststream_concurrent_aiokafka.batch_committer"): + await committer.commit_all(flush_timeout_sec=2.0) + + records: typing.Final = [r for r in caplog.records if r.name == "faststream_concurrent_aiokafka.batch_committer"] + errors: typing.Final = [r for r in records if r.levelno == logging.ERROR] + warnings: typing.Final = [r for r in records if r.levelno == logging.WARNING] + infos: typing.Final = [r for r in records if r.levelno == logging.INFO] + assert len(errors) == 1 + assert errors[0].exc_info is not None + assert len(warnings) == 2 + assert all(r.exc_info is None for r in warnings) + assert len(infos) == 1 + await committer.close() diff --git a/tests/test_pending_commits.py b/tests/test_pending_commits.py index 0c1b97b..089ce47 100644 --- a/tests/test_pending_commits.py +++ b/tests/test_pending_commits.py @@ -117,3 +117,54 @@ def test_clear_watermarks_clears_every_consumer_on_partition() -> None: ready = {id(rc.consumer): rc for rc in pending.take_ready()} assert ready[id(a)].offsets == {_tp(): 2} assert ready[id(b)].offsets == {_tp(): 2} + + +def test_held_commit_counts_as_pending(mock_consumer: MockAIOKafkaConsumer) -> None: + pending = PendingCommits() + pending.absorb(make_commit_task(mock_consumer, _tp(), offset=0, done=True)) + pending.absorb(make_commit_task(mock_consumer, _tp(), offset=1, done=True)) + [rc] = pending.take_ready() + + pending.hold(rc) + + assert len(pending) == 2 + + +def test_take_ready_returns_a_held_commit(mock_consumer: MockAIOKafkaConsumer) -> None: + pending = PendingCommits() + pending.absorb(make_commit_task(mock_consumer, _tp(), offset=0, done=True)) + [rc] = pending.take_ready() + pending.hold(rc) + + assert pending.take_ready() == [rc] + assert len(pending) == 0 + + +def test_take_ready_merges_a_held_commit_with_new_ready_work(mock_consumer: MockAIOKafkaConsumer) -> None: + pending = PendingCommits() + pending.absorb(make_commit_task(mock_consumer, _tp(0), offset=0, done=True)) + [held] = pending.take_ready() + pending.hold(held) + pending.absorb(make_commit_task(mock_consumer, _tp(0), offset=1, done=True)) + pending.absorb(make_commit_task(mock_consumer, _tp(1), offset=5, done=True)) + + [rc] = pending.take_ready() + + assert rc.consumer is mock_consumer + assert rc.offsets == {_tp(0): 2, _tp(1): 6} + assert [t.offset for t in rc.tasks] == [0, 1, 5] + assert len(pending) == 0 + + +def test_held_commits_stay_per_consumer() -> None: + a, b = MockAIOKafkaConsumer(), MockAIOKafkaConsumer() + pending = PendingCommits() + pending.absorb(make_commit_task(a, _tp(), offset=0, done=True)) + [held] = pending.take_ready() + pending.hold(held) + pending.absorb(make_commit_task(b, _tp(), offset=3, done=True)) + + ready = {id(rc.consumer): rc for rc in pending.take_ready()} + + assert ready[id(a)].offsets == {_tp(): 1} + assert ready[id(b)].offsets == {_tp(): 4}