Kafka Streams runner: bundles are not bounded by time

[junaiddshaukat]

--maxBundleTimeMs is accepted but has no effect. A bundle is closed when it reaches --maxBundleSize elements or when a watermark arrives, so on a sparse stream the elements already fed to a bundle are not emitted until the next watermark, and the watermark cannot advance past them.

The natural implementation — closing the bundle from a wall-clock punctuator — produces duplicate output against a real broker. With it enabled, an integration test running two chained GroupByKeys across four partitions emits its single group six times, reproducibly, and the count keeps climbing after the input has stopped.

What has been ruled out:

• Metrics folding. Splitting the same input across many bundles does not change a user counter (MetricsAcrossBundlesTest).
• 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 appears to be closing a Fn-API bundle from a punctuator rather than from record processing.

Note that Kafka Streams sets commitOffsetNeeded inside StreamTask#process(), so data produced from a punctuator sits outside that accounting; KAFKA-6906 was a bug of that shape.

That note was wrong — see the comment below. There is no commitOffsetNeeded flag; it is commitNeeded, and it is set after a punctuation runs as well as from process(), so punctuator output is not outside the commit accounting. Corrected by Matthias J. Sax on the Kafka dev list. The cause of the duplication is still unknown, and the way forward is data-driven bundle markers rather than a wall-clock punctuator.

Part of #18479.

[junaiddshaukat]

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), 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.