Fix error messages in faust app #166

faust-streaming · Dec 13, 2021 · 28d0db2 · 28d0db2
1 parent a86350b
commit 28d0db2
Show file tree

Hide file tree

Showing 2 changed files with 9 additions and 5 deletions.
diff --git a/faust/_cython/streams.pyx b/faust/_cython/streams.pyx
@@ -160,8 +160,10 @@ cdef class StreamIterator:
             offset = message.offset
             consumer = self.consumer
 
-            if not self.app.flow_control.is_active() or\
-                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 flow control.is_active %r",

diff --git a/faust/streams.py b/faust/streams.py
@@ -974,16 +974,18 @@ async def _py_aiter(self) -> AsyncIterator[T_co]:
                         tp = message.tp
                         offset = message.offset
 
-                        if not self.app.flow_control.is_active() or\
-                            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 flow_control.is_active %r",
                                 message,
                                 message.generation_id,
                                 self.app.consumer_generation_id,
-                                self.app.flow_control.is_active()
+                                self.app.flow_control.is_active(),
                             )
                             break
                         if topic in acking_topics and not message.tracked: