fix: report the cause when the committer main task dies

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

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)

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)