Thanks Matthias, that is very helpful, and thank you for the second mail too, you are right, and I have corrected the issue.
Your wider point is the useful one: we were trying to align bundles with a boundary Kafka Streams does not expose and does not intend anyone to reason about, so the design was wrong rather than the implementation. The plan now is to make the flush data-driven instead. The sources already punctuate, so a source emits a flush marker that travels the topology like a watermark, and each stage closes its bundle when it receives one, inside process(). That keeps bundle boundaries away from transaction boundaries entirely. On Tue, Aug 25, 2026 at 5:14 AM Matthias J. Sax <[email protected]> wrote: > 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 > >> > > > >
