fix(Stream): handle errors.ConsumerStoppedError exception for the new…

… consumtion way
kpn · Dec 19, 2023 · 31b2742 · 31b2742
1 parent 73819c2
commit 31b2742
Showing 1 changed file with 2 additions and 0 deletions.
diff --git a/kstreams/streams.py b/kstreams/streams.py
@@ -250,6 +250,8 @@ async def func_wrapper_with_typing(self, calling_type: UDFType) -> None:
                 else:
                     # typing with cr and stream
                     await self.func(cr, self)
+        except errors.ConsumerStoppedError:
+            return
         except Exception as e:
             logger.exception(f"CRASHED Stream!!! Task {self._consumer_task} \n\n {e}")