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")