junaiddshaukat commented on issue #39633: URL: https://github.com/apache/beam/issues/39633#issuecomment-5409894776
Matthias J. Sax corrected this on the Kafka dev list, and the note in the description above is wrong. There is no `commitOffsetNeeded` flag. It is `commitNeeded`, and it is set after a punctuation runs, not only from `process()` — `maybePunctuateSystemTime()` sets it when a wall-clock punctuation fires ([StreamTask.java:1209](https://github.com/apache/kafka/blob/3.9/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java#L1209)), and `maybePunctuateStreamTime()` does the same for event time. So records forwarded from a punctuator are *not* outside the commit accounting, and the KAFKA-6906 shape is not what is happening here. The cause of the duplication is still unknown. The rest of what he said is more useful than the correction. Paraphrasing: - Punctuations do not fit the EOS pattern of "read records, produce output, atomically commit the output plus the input offsets". EOS does not cover side effects, and a punctuation is most likely a side effect. - There is no supported way to run work just before a commit. - A transaction is committed by the runtime in the background and is agnostic to punctuations, so there is no way to align a flush with a commit boundary. - Transactions are an internal Kafka Streams concept, deliberately not exposed at the API level. Bundles need to work without being coupled to them. That last point is the real answer: the design was wrong, not just the implementation. Trying to close a bundle on a timer means trying to align with a boundary Kafka Streams does not expose and does not intend anyone to reason about. ## The way forward: bundle markers Jan Lukavský's suggestion, which fits the above: make the flush data-driven rather than wall-clock driven, the same way watermarks already are. The sources already punctuate, so a source emits a *flush marker* alongside its watermark. The marker travels the topology like a watermark payload, and each downstream stage closes its bundle when it **receives** one — inside `process()`, as ordinary record handling. Bundle boundaries then never touch transaction boundaries, and nothing is forwarded from a punctuator except by the sources, which already work that way. Most of the plumbing exists. `KStreamsPayload` is already a sum type carrying data and watermarks, so the marker is a third variant, and `WatermarkAggregator` already solves the "have I heard from every upstream partition" problem if the marker needs the same treatment. Open questions for whoever picks this up: - Does a marker need to be held until every upstream partition has sent one, like a watermark, or can a stage flush on the first one it sees? Flushing early is harmless for correctness — a bundle boundary is not a correctness boundary — so the simpler rule probably wins. - The marker has to be broadcast across a repartition topic like a watermark is, or downstream partitions will never see one. - `--maxBundleTimeMs` then means "how often a source emits a flush marker" rather than "how long a bundle may stay open", which is close enough to keep the option name but the documentation needs to say so. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
