fix(stream): recreate consumer when a stream instance is restarted (#77)

This reverts commit f0dca3a.

Co-authored-by: Marcos Schroh <[email protected]>

.github/workflows/docs-preview.yaml

@@ -39,6 +39,6 @@ jobs:
           poetry run mkdocs build
 
       - name: Deploy preview
-        uses: rossjrw/pr-preview-action@v1
+        uses: rossjrw/pr-preview-action@v1.2.0
         with:
           source-dir: ./site/

kstreams/streams.py

@@ -87,9 +87,8 @@ async def func_wrapper(func):
                     f"CRASHED Stream!!! Task {self._consumer_task} \n\n {e}"
                 )
 
-        if self.consumer is None:
-            self.consumer = self._create_consumer()
-
+        # Always create a consumer on stream.start
+        self.consumer = self._create_consumer()
         func = self.func(self)
         await self.consumer.start()
         self.running = True

tests/test_client.py

@@ -1,4 +1,3 @@
-from unittest import mock
 from unittest.mock import Mock
 
 import pytest
@@ -267,21 +266,3 @@ async def test_e2e_consume_multiple_topics():
         assert topic_2.total_events == total_events
 
         assert TopicManager.all_messages_consumed()
-
-
-@pytest.mark.asyncio
-async def test_create_consumer_if_consumer_is_none(stream_engine: StreamEngine):
-    with mock.patch("kstreams.clients.aiokafka.AIOKafkaConsumer.start"):
-        topic_name = "local--kstreams"
-
-        @stream_engine.stream(topic_name, name="my-stream")
-        async def consume(stream):
-            async for cr in stream:
-                ...
-
-        stream = stream_engine.get_stream("my-stream")
-        await stream.start()
-        temp = stream.consumer
-        await stream.stop()
-        await stream.start()
-        assert temp == stream.consumer

tests/test_stream_engine.py

@@ -100,6 +100,25 @@ async def stream(_):
             Consumer.stop.assert_awaited()
 
 
+@pytest.mark.asyncio
+async def test_recreate_consumer_on_re_tart_stream(stream_engine: StreamEngine):
+    with mock.patch("kstreams.clients.aiokafka.AIOKafkaConsumer.start"):
+        topic_name = "local--kstreams"
+        stream_name = "my-stream"
+
+        @stream_engine.stream(topic_name, name=stream_name)
+        async def consume(stream):
+            async for _ in stream:
+                ...
+
+        stream = stream_engine.get_stream(stream_name)
+        await stream.start()
+        consumer = stream.consumer
+        await stream.stop()
+        await stream.start()
+        assert consumer is not stream.consumer
+
+
 @pytest.mark.asyncio
 async def test_engine_not_started(stream_engine: StreamEngine):
     topic = "local--hello-kpn"