fix race condition when buffers are full

faust-streaming · Dec 13, 2021 · 891ef26 · 891ef26
1 parent 9a298ac
commit 891ef26
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 2 deletions.
diff --git a/faust/_cython/streams.pyx b/faust/_cython/streams.pyx
@@ -160,7 +160,8 @@ cdef class StreamIterator:
             offset = message.offset
             consumer = self.consumer
 
-            if message.generation_id != self.app.consumer_generation_id:
+            if not self.app.flow_control.is_active() or\
+                message.generation_id != self.app.consumer_generation_id:
                 self.app.log.dev(
                     "Skipping message %r with generation_id %r because "
                     "app generation_id is %r",

diff --git a/faust/streams.py b/faust/streams.py
@@ -974,7 +974,8 @@ async def _py_aiter(self) -> AsyncIterator[T_co]:
                         tp = message.tp
                         offset = message.offset
 
-                        if message.generation_id != self.app.consumer_generation_id:
+                        if not self.app.flow_control.is_active() or\
+                            message.generation_id != self.app.consumer_generation_id:
                             value = skipped_value
                             self.log.dev(
                                 "Skipping message %r with generation_id %r because "