fix: prevent ConsumerStoppedError exception when engine or a stream i…

…s stopped. Related to #234

examples/fastapi-webapp/fastapi_webapp/app.py

@@ -4,7 +4,7 @@
 from kstreams.streams_utils import StreamErrorPolicy
 
 from .resources import stream_engine
-from .streams import consume
+from .streams import consume, consume_2
 from .views import router
 
 app = FastAPI()
@@ -21,6 +21,7 @@ async def shutdown_event():
 
 
 stream_engine.add_stream(consume, error_policy=StreamErrorPolicy.STOP_APPLICATION)
+stream_engine.add_stream(consume_2, error_policy=StreamErrorPolicy.RESTART)
 app.include_router(router)
 app.add_middleware(PrometheusMiddleware, filter_unhandled_paths=True)
 app.add_api_route("/metrics", metrics)  # type: ignore

examples/fastapi-webapp/fastapi_webapp/streams.py

@@ -1,5 +1,6 @@
 import logging
 
+import asyncio
 from kstreams import ConsumerRecord, stream
 
 logger = logging.getLogger(__name__)
@@ -11,3 +12,10 @@ async def consume(cr: ConsumerRecord):
 
     if cr.value == b"error":
         raise ValueError("error....")
+
+
+@stream("local--hello-world", group_id="example-group-2")
+async def consume_2(cr: ConsumerRecord):
+    print(f"Event consumed: headers: {cr.headers}, payload: {cr}")
+    await asyncio.sleep(10)
+    raise ValueError

examples/graceful-shutdown-example/graceful_shutdown_example/app.py

@@ -4,6 +4,7 @@
 import aiorun
 
 import kstreams
+import kstreams.streams_utils
 
 logger = logging.getLogger(__name__)
 

examples/rqlite-example/docker-compose.yaml

@@ -0,0 +1,35 @@
+version: '3'
+services:
+  zookeeper:
+    image: "confluentinc/cp-zookeeper:7.7.0"
+    hostname: zookeeper
+    container_name: kstream_zookeeper
+    ports:
+      - 32181:32181
+    environment:
+      - ZOOKEEPER_CLIENT_PORT=32181
+  kafka:
+    image: confluentinc/cp-kafka:7.7.0
+    hostname: kafka
+    container_name: kstream_kafka
+    ports:
+    - 9092:9092
+    - 9093:9093
+    depends_on:
+    - zookeeper
+    environment:
+      - KAFKA_ZOOKEEPER_CONNECT=zookeeper:32181
+      - KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1
+      - KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=PLAINTEXT:PLAINTEXT,PLAINTEXT2:PLAINTEXT
+      - KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://127.0.0.1:9092,PLAINTEXT2://kafka:9093
+      - KAFKA_BROKER_ID=1
+      - KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS=0
+      - CONFLUENT_METRICS_ENABLE=true
+      - CONFLUENT_SUPPORT_CUSTOMER_ID=anonymous
+      - KAFKA_AUTO_CREATE_TOPICS_ENABL="true"
+  rqlite:
+    image: rqlite/rqlite:8.28.4
+    hostname: rqlite
+    container_name: kstream_rqlite
+    ports:
+      - 4001:4001