Benthos allows to declare configuration to join, filter, enrich, transform streams and support a lot of source/target like brokers, Elastic, HTTP API

https://www.benthos.dev/