From 761eec4bb0279f7ceb8bdc3255bae48b53dc18ae Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Tue, 22 Sep 2026 16:02:02 +0300 Subject: [PATCH] fix: report the cause when the committer main task dies MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An unhandled exception from consumer.commit() ends the streaming loop, and spawn() registered no done-callback, so the death itself logged nothing. The only ERROR a dead committer produced was the send_task guard's CommitterIsDeadError on the next message — a symptom, not a cause — while the real exception surfaced later at WARNING from close(). Error reporters promote ERROR to an event and keep lower levels as breadcrumbs, so a production incident arrived with no root cause attached. _on_committer_done now reports the cause at ERROR when the loop ends, and is the only place that does; close()'s duplicate branch is gone. Behaviour is unchanged: a dead committer stays dead. Widening _call_committer's except list to Exception would keep the loop alive by discarding batches it cannot commit, but an error that repeats every round becomes an infinite redelivery — the consumer looks healthy, offsets never advance, and nothing reaches the liveness probe. Respawning is worse, because _pending, _messages_queue and _uncommitted_count may already describe offsets partly committed before the throw, so a fresh loop can commit past work that never finished. --- .../batch_committer.py | 22 +++-- tests/test_healthcheck.py | 28 ++++++ tests/test_kafka_committer.py | 99 +++++++++++++++++-- 3 files changed, 137 insertions(+), 12 deletions(-) diff --git a/faststream_concurrent_aiokafka/batch_committer.py b/faststream_concurrent_aiokafka/batch_committer.py index 4a9eb90..a59a5bb 100644 --- a/faststream_concurrent_aiokafka/batch_committer.py +++ b/faststream_concurrent_aiokafka/batch_committer.py @@ -243,9 +243,24 @@ async def send_task(self, new_task: KafkaCommitTask) -> None: self._uncommitted_count += 1 await self._messages_queue.put(new_task) + def _on_committer_done(self, task: asyncio.Task[typing.Any]) -> None: + """Report the exception that ended the streaming loop, at the moment it ends it. + + Without this the only ERROR a dead committer produces is the `send_task` guard's + `CommitterIsDeadError`, raised whenever the next message arrives. That names the symptom + and not the cause, so an error reporter that promotes ERROR to an event and keeps lower + levels as breadcrumbs captures a committer death with nothing attached explaining it. + """ + if task.cancelled(): + return + exc: typing.Final = task.exception() + if exc is not None: + logger.error("Committer main task died; offsets will no longer be committed", exc_info=exc) + def spawn(self) -> None: if not self._commit_task: self._commit_task = asyncio.create_task(self._run_commit_process()) + self._commit_task.add_done_callback(self._on_committer_done) else: logger.error("Committer main task already running") @@ -256,12 +271,7 @@ async def close(self) -> None: return if self._commit_task.done(): - # Task already terminated (cancelled or raised). Nothing to wait on; surface - # any non-cancellation exception so it gets logged, then continue shutdown. - if not self._commit_task.cancelled(): - exc = self._commit_task.exception() - if exc is not None: - logger.warning("Committer task had already died before close()", exc_info=exc) + # Nothing to wait on. _on_committer_done already reported any exception. return self._stop_requested = True diff --git a/tests/test_healthcheck.py b/tests/test_healthcheck.py index 8b27970..92af5db 100644 --- a/tests/test_healthcheck.py +++ b/tests/test_healthcheck.py @@ -1,3 +1,6 @@ +# ruff: noqa: SLF001 +import asyncio +import contextlib import typing from unittest.mock import MagicMock @@ -41,3 +44,28 @@ async def test_unhealthy_when_is_healthy_returns_false() -> None: mock_handler.is_healthy = False test_broker.context.set_global("concurrent_processing", mock_handler) assert is_kafka_handler_healthy(test_broker.context) is False + + +async def test_unhealthy_when_the_committer_died_under_a_running_handler() -> None: + """A dead committer must fail the probe on its own, before a message meets the send_task guard. + + The handler is still running and still accepting dispatches, so `_is_running` says nothing + here; only the committer's liveness does. This is the chain an operator's liveness probe + depends on to restart a consumer that can no longer commit offsets. + """ + broker: typing.Final = KafkaBroker("localhost:9092") + async with TestKafkaBroker(broker) as test_broker: + handler: typing.Final = await initialize_concurrent_processing(context=test_broker.context) + try: + assert is_kafka_handler_healthy(test_broker.context) is True + + commit_task: typing.Final = handler._committer._commit_task + assert commit_task is not None + commit_task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await commit_task + + assert handler.is_running is True + assert is_kafka_handler_healthy(test_broker.context) is False + finally: + await stop_concurrent_processing(test_broker.context) diff --git a/tests/test_kafka_committer.py b/tests/test_kafka_committer.py index e7aecd0..cf5f3ba 100644 --- a/tests/test_kafka_committer.py +++ b/tests/test_kafka_committer.py @@ -147,8 +147,8 @@ async def test_committer_uses_shutdown_timeout_kwarg() -> None: assert committer._shutdown_timeout == 0.05 -async def test_committer_close_logs_when_task_already_died(caplog: pytest.LogCaptureFixture) -> None: - """If the committer task crashed before close() is called, the exception is logged.""" +async def test_committer_close_is_a_noop_when_the_task_already_died() -> None: + """close() on an already-dead committer returns quietly; the death was reported when it happened.""" committer: typing.Final = KafkaBatchCommitter(commit_batch_timeout_sec=0.1, commit_batch_size=10) async def crashing() -> typing.Never: @@ -160,7 +160,7 @@ async def crashing() -> typing.Never: await committer._commit_task await committer.close() - assert "Committer task had already died before close()" in caplog.text + assert committer._commit_task.done() async def test_committer_close_but_timeout_error(caplog: pytest.LogCaptureFixture) -> None: @@ -1344,7 +1344,7 @@ async def done() -> None: async def test_committer_close_when_task_already_finished_cleanly(caplog: pytest.LogCaptureFixture) -> None: """close() on a committer whose task already completed normally (not crashed, not cancelled). - Must return without logging a warning — exercise the exc-is-None branch. + Must report nothing — exercise the done-callback's exc-is-None branch. """ committer: typing.Final = KafkaBatchCommitter(commit_batch_timeout_sec=0.01, commit_batch_size=10) committer.spawn() @@ -1358,9 +1358,8 @@ async def test_committer_close_when_task_already_finished_cleanly(caplog: pytest assert not committer._commit_task.cancelled() assert committer._commit_task.exception() is None # normal exit, not a crash - # close() must not log "had already died" since there was no exception. await committer.close() - assert "Committer task had already died before close()" not in caplog.text + assert not _death_records(caplog) # ---------- _streaming_iteration: accepts_new_work() is False (post-shutdown loop) ---------- @@ -1406,3 +1405,91 @@ async def slow() -> None: assert not committer.is_healthy consumer.commit.assert_called_once_with({tp: 2}) + + +# ---------- committer death is reported at the moment it happens ---------- + + +def _death_records(caplog: pytest.LogCaptureFixture) -> list[logging.LogRecord]: + return [ + record + for record in caplog.records + if record.name == "faststream_concurrent_aiokafka.batch_committer" and record.levelno == logging.ERROR + ] + + +async def test_committer_reports_the_cause_when_the_main_task_dies(caplog: pytest.LogCaptureFixture) -> None: + """INVARIANT: the exception that kills the streaming loop is logged at ERROR, carrying itself. + + `_call_committer` handles only `CommitFailedError`, `IllegalStateError` and `KafkaError`; + anything else ends `_run_commit_process` and the committer never commits again. Every later + message then hits the `send_task` guard and raises `CommitterIsDeadError`, which names the + symptom and not the cause. An error reporter that promotes ERROR to an event and keeps lower + levels as breadcrumbs would otherwise capture only the guard, which is how a released + `TypeError` from `AIOKafkaConsumer.commit` reached production with no cause attached. + """ + consumer: typing.Final = MockAIOKafkaConsumer() + consumer.commit.side_effect = TypeError("Key should be TopicPartition instance") + committer: typing.Final = KafkaBatchCommitter(commit_batch_timeout_sec=0.05, commit_batch_size=1) + committer.spawn() + + async def quick() -> None: + return None + + await committer.send_task( + KafkaCommitTask( + asyncio_task=asyncio.create_task(quick()), + offset=10, + consumer=consumer, + topic_partition=TopicPartition(topic="t", partition=0), + ) + ) + + assert committer._commit_task is not None + await _drive_until(committer._commit_task.done) + + records: typing.Final = _death_records(caplog) + assert len(records) == 1 + exc_info: typing.Final = records[0].exc_info + assert exc_info is not None + assert isinstance(exc_info[1], TypeError) + + +async def test_committer_death_is_reported_before_any_later_message(caplog: pytest.LogCaptureFixture) -> None: + """The cause is on the record before the first CommitterIsDeadError, not after it.""" + consumer: typing.Final = MockAIOKafkaConsumer() + consumer.commit.side_effect = TypeError("Key should be TopicPartition instance") + committer: typing.Final = KafkaBatchCommitter(commit_batch_timeout_sec=0.05, commit_batch_size=1) + committer.spawn() + + async def quick() -> None: + return None + + tp: typing.Final = TopicPartition(topic="t", partition=0) + await committer.send_task( + KafkaCommitTask(asyncio_task=asyncio.create_task(quick()), offset=10, consumer=consumer, topic_partition=tp) + ) + + assert committer._commit_task is not None + await _drive_until(committer._commit_task.done) + + assert _death_records(caplog) + assert not committer.is_healthy + with pytest.raises(CommitterIsDeadError): + await committer.send_task( + KafkaCommitTask(asyncio_task=asyncio.create_task(quick()), offset=11, consumer=consumer, topic_partition=tp) + ) + + +async def test_committer_cancellation_is_not_reported_as_a_death(caplog: pytest.LogCaptureFixture) -> None: + """Shutdown cancels the loop; that is not a failure and must not reach an error reporter.""" + committer: typing.Final = KafkaBatchCommitter(commit_batch_timeout_sec=0.05, commit_batch_size=1) + committer.spawn() + + assert committer._commit_task is not None + committer._commit_task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await committer._commit_task + await asyncio.sleep(0) + + assert not _death_records(caplog)