fixed recovery hang and undefined set_close method in aiokafka

faust-streaming · Nov 8, 2020 · 06755bc · 06755bc
1 parent 41e358c
commit 06755bc
Showing 1 changed file with 2 additions and 1 deletion.
diff --git a/faust/tables/recovery.py b/faust/tables/recovery.py
@@ -246,6 +246,7 @@ def on_partitions_revoked(self, revoked: Set[TP]) -> None:
         T = traced_from_parent_span()
         T(self.flush_buffers)()
         self.signal_recovery_reset.set()
+        self.signal_recovery_start.set()
 
     async def on_rebalance(
         self, assigned: Set[TP], revoked: Set[TP], newly_assigned: Set[TP]
@@ -294,7 +295,7 @@ async def on_rebalance(
                 child_of=rebalancing_span,
             )
             app._span_add_default_tags(self._recovery_span)
-        self.signal_recovery_reset.clear()
+        self.signal_recovery_reset.set()
         self.signal_recovery_start.set()
 
     async def _resume_streams(self) -> None: