wbarnha
commented
1 year ago Interesting idea, never thought about something like this. There's a feature in faust called app.crash, I don't generally recommend it, but it should be interesting for your use-case. I just tested this without issues:
import faust
app = faust.App("my-app",
broker="localhost:9092",
broker_heartbeat_interval=30,
broker_session_timeout=360,
broker_request_timeout=360,
broker_max_poll_records=1,
topic_disable_leader=True,
broker_max_poll_interval=3600)
topic = app.topic("test-stop", value_serializer="raw")
@app.agent(topic)
async def process_messages(stream):
async for message in stream.items():
await app.crash(reason=Exception("Shutdown signal received on stop topic"))
if __name__ == "__main__":
app.main()There's probably a more graceful way of shutting things down, but calling app.crash guarantees a shutdown.
cleomart
Checklist
masterbranch of Faust.Steps to reproduce
I would like to create a on-demand kafka streaming job using faust that will get terminated once a certain message is received. Here is my code. For simplicity, I terminate the faust when a message is received.
Expected behavior
I expect the faust app to be terminated gracefully and the python script would exit and the job will be completed.
Actual behavior and Full traceback
When the faust app is stopped, I get this error and the job is stuck in this infinite loop.
Versions