fix race condition when buffers are full (faust-streaming#237)

* fix race condition when buffers are full * fix race condition when buffers are full * Fix error messages in faust app faust-streaming#166
hamzah-jamal · Dec 13, 2021 · 7d861dc · 7d861dc
1 parent 9a298ac
commit 7d861dc
Show file tree

Hide file tree

Showing 2 changed files with 12 additions and 4 deletions.
diff --git a/faust/_cython/streams.pyx b/faust/_cython/streams.pyx
@@ -160,13 +160,17 @@ 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",
+                    "app generation_id is %r flow control.is_active %r",
                     message,
                     message.generation_id,
                     self.app.consumer_generation_id,
+                    self.app.flow_control.is_active()
                 )
                 return None, self._skipped_value, stream_state
 

diff --git a/faust/streams.py b/faust/streams.py
@@ -974,14 +974,18 @@ 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 "
-                                "app generation_id is %r",
+                                "app generation_id is %r flow_control.is_active %r",
                                 message,
                                 message.generation_id,
                                 self.app.consumer_generation_id,
+                                self.app.flow_control.is_active(),
                             )
                             break
                         if topic in acking_topics and not message.tracked: