From 2b9c16af5391045e06215767f4e0c5663c735d28 Mon Sep 17 00:00:00 2001 From: William Barnhart Date: Sun, 19 Jul 2026 10:54:08 +0000 Subject: [PATCH] Fix #446: don't crash livelock detector when consumer not yet started The commit-livelock detector task periodically calls verify_all_partitions_active(), which iterates self.assignment(). When the consumer thread has not finished starting (observed with alternative event loops such as eventlet), assignment() raises ConsumerNotStarted and the unhandled exception crashes the app. There is nothing to verify until the consumer thread is up, so catch ConsumerNotStarted and skip the cycle instead of propagating it. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01HHPL4VFWQRQPpjR1gXSKyL --- faust/transport/consumer.py | 12 ++++++++++-- tests/unit/transport/test_consumer.py | 17 ++++++++++++++++- 2 files changed, 26 insertions(+), 3 deletions(-) diff --git a/faust/transport/consumer.py b/faust/transport/consumer.py index d5a82b1b8..6c8490c01 100644 --- a/faust/transport/consumer.py +++ b/faust/transport/consumer.py @@ -80,7 +80,7 @@ from mode.utils.text import pluralize from mode.utils.times import Seconds -from faust.exceptions import ProducerSendError +from faust.exceptions import ConsumerNotStarted, ProducerSendError from faust.types import TP, AppT, ConsumerMessage, Message, RecordMetadata from faust.types.core import HeadersArg from faust.types.transports import ( @@ -900,7 +900,15 @@ async def _commit_livelock_detector(self) -> None: # pragma: no cover async def verify_all_partitions_active(self) -> None: now = monotonic() - for tp in self.assignment(): + try: + assignment = self.assignment() + except ConsumerNotStarted: + # The commit-livelock detector may run before the consumer + # thread has finished starting (observed with alternative event + # loops such as eventlet). There is nothing to verify yet, so + # skip this cycle instead of crashing the app. See issue #446. + return + for tp in assignment: await self.sleep(0) if not self.should_stop: self.verify_event_path(now, tp) diff --git a/tests/unit/transport/test_consumer.py b/tests/unit/transport/test_consumer.py index 070066984..46cfcdfd8 100644 --- a/tests/unit/transport/test_consumer.py +++ b/tests/unit/transport/test_consumer.py @@ -9,7 +9,7 @@ from faust import App from faust.app._attached import Attachments -from faust.exceptions import AlreadyConfiguredWarning +from faust.exceptions import AlreadyConfiguredWarning, ConsumerNotStarted from faust.tables.manager import TableManager from faust.transport.base import Producer, Transport from faust.transport.conductor import Conductor @@ -1413,6 +1413,21 @@ async def test_verify_all_partitions_active(self, *, consumer): ] ) + @pytest.mark.asyncio + async def test_verify_all_partitions_active__consumer_not_started( + self, *, consumer + ): + # The livelock detector can run before the consumer thread has + # started; assignment() then raises ConsumerNotStarted. This must + # be swallowed rather than crash the app. See issue #446. + consumer.assignment = Mock(name="assignment", side_effect=ConsumerNotStarted()) + consumer.verify_event_path = Mock(name="verify_event_path") + + with patch("faust.transport.consumer.monotonic"): + await consumer.verify_all_partitions_active() + + consumer.verify_event_path.assert_not_called() + @pytest.mark.asyncio async def test_verify_all_partitions_active__bail_on_sleep(self, *, consumer): consumer.assignment = Mock(name="assignment")