Re: Beam's new Kafka Streams runner: write KS applications in any Beam SDK

[Matthias J. Sax]

Btw, I did just double checked the code. The `commitNeeded` flag is set 
after a punctuation run: 
https://github.com/apache/kafka/blob/3.9/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java#L1209

There is no `commitOffsetNeeded` flag...

-Matthias


On 8/24/26 2:53 PM, Matthias J. Sax wrote:
Thanks for reaching out:

is forwarding records from a wall-clock punctuator under EOS expected 
to be safe

I don't know anything about Beam, or "bundles", but in general 
punctuation in combination with EOS are tricky. EOS guarantees that you 
read some messaged, producer the corresponding output, and atomically 
commit the result records plus the advanced input topic offsets.

Punctuations, wall-clock and event-time, don't really fit into this 
pattern. It might still be ok to use them, but w/o more details it's 
very hard to say if this could work or not. -- EOS in general does not 
cover side-effects, and punctuation (depending on their implementation) 
are most likely a side-effect.


and is there a supported way to run work just before a commit

No, that's not supported.


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.

Given how Kafka transactions and punctuation work, there is no good way 
to do this. A transaction gets committed by the runtime in the 
background, and this process is totally agnostic to punctuations. I 
cannot say why you would see duplicates though (assuming your buffer is 
persistent storage), but there is no guarantee when a punctuation runs 
exactly, and if you take input records and buffer them, if the TX 
commits the offset will get advanced, so the buffer must ensure it does 
not lose data (using a built-in persistent store would guarantee this, 
because of the writes into the changelog topic).

And if there is an ongoing transaction and you flush record from the 
buffer (ie, forward and delete from the store), all store updates should 
be part of the same TX, so if the transaction commits, the record should 
be in the result topic and the deletes in the changelog topic (and it 
the TX fails, the buffer should roll back to the previous state).


Overall it seems that you try to align a Kafka TX with a "bundle", but 
in KS TX are are internal concept (implementation details) not exposed 
at the API level because users should not need to think/reason about it. 
I believe you will need to find a way to do bundles w/o coupling them on 
TX boundaries which you cannot control, and which are not exposed.


Hope this helps.


-Matthias


On 8/23/26 6:25 PM, Junaid wrote:
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