fix for consumer errors in app #166

faust-streaming · Jun 29, 2021 · 75a1231 · 75a1231
1 parent 6f8f0b7
commit 75a1231
Showing 1 changed file with 3 additions and 4 deletions.
diff --git a/faust/transport/consumer.py b/faust/transport/consumer.py
@@ -848,13 +848,12 @@ async def _commit_livelock_detector(self) -> None:  # pragma: no cover
         await self.sleep(interval)
         async for sleep_time in self.itertimer(interval, name="livelock"):
             if not self.app.rebalancing:
-                await self.app.loop.run_in_executor(
-                    None, self.verify_all_partitions_active
-                )
+                await self.verify_all_partitions_active()
 
-    def verify_all_partitions_active(self) -> None:
+    async def verify_all_partitions_active(self) -> None:
         now = monotonic()
         for tp in self.assignment():
+            await self.sleep(0)
             if not self.should_stop:
                 self.verify_event_path(now, tp)