Fix #446: don't crash livelock detector when consumer not yet started

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)

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