Yeah you definitely need something like Feldera: https://github.com/feldera/feldera