Hi all, TL;DR: Apache Beam has merged a Kafka Streams runner, which turns a Beam pipeline into a Kafka Streams topology. Because Beam is portable, this means a Kafka Streams application no longer has to be written in Java: the runner is Java, but your code can be Python, Go or YAML. It is an experimental skeleton, opt-in at build time and in no Beam release. I also have a question at the end about forwarding records from a punctuator under EOS, which is the one thing currently blocking us.
I wrote this over the summer as a Google Summer of Code project. The portability is the part worth explaining. Beam runs user code in a separate process over gRPC, whichever SDK it was written in, so the runner never executes your code itself. A pipeline written in Python runs on a runner written entirely in Java. So you can write what is really a Kafka Streams application in Python and deploy it like any other Kafka Streams application: no job manager, no second cluster, scale by starting more copies of the process. Java and Python are covered by tests; the other SDKs should work by construction, but I have not run them. Stateless ParDo, GroupByKey, Combine, fixed and sliding windows, Flatten and metrics work. Side inputs, stateful ParDo with user timers, merging windows and splittable DoFn do not yet. Now the question. Beam executes user code in bundles, and a bundle must be closed before its output is flushed, so it needs a time bound as well as a size one. Closing the bundle from a wall-clock punctuator, which is the natural implementation, produces duplicate output against a real broker: a test with two chained GroupByKeys across four partitions emits its single group six times, reproducibly, and the count keeps climbing after input stops. Kafka 3.9.0, EOS v2. We ruled out metrics folding and ProcessorContext.commit(): the duplication happens with the commit request removed, and does not happen with the punctuator disabled but the commit still requested. So it seems specific to producing records from a punctuator rather than from process(). We noticed commitOffsetNeeded is set inside StreamTask#process(), so punctuator output sits outside that accounting; KAFKA-6906 was a bug of that shape, fixed well before our version. So: is forwarding records from a wall-clock punctuator under EOS expected to be safe, and is there a supported way to run work just before a commit? We want to flush buffered output exactly at the commit boundary, so the records produced and the offsets consumed commit together. If there is an idiom for that, I would rather use it than work around it. The runner is at [1] <https://github.com/apache/beam/tree/master/runners/kafka-streams>, the tracking issue at [2] <https://github.com/apache/beam/issues/18479>, and the full investigation of the bundle problem at [3]. <https://github.com/apache/beam/issues/39633> Thanks, Junaid [1] https://github.com/apache/beam/tree/master/runners/kafka-streams [2] https://github.com/apache/beam/issues/18479 [3] https://github.com/apache/beam/issues/39633
