https://ksqldb.io/ is great for this, assuming you're already using Kafka.