Kafka Streams runner: add a flush marker payload variant

[junaiddshaukat]

Summary

First of four PRs toward bundles bounded by time (#39633). This one adds the flush marker to the payload envelope and nothing else: nothing emits a marker and nothing consumes one yet, so there is no behaviour change.

Why a marker rather than a timer

--maxBundleTimeMs is accepted today and has no effect. A bundle must be closed before its output is flushed, so it needs a time bound as well as a size one, otherwise on a sparse stream the elements already fed to it wait for the next watermark.

The natural implementation, closing the bundle from a wall-clock punctuator, 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.

Asking about this on the Kafka dev list settled why. Matthias J. Sax's answer was that punctuations do not fit the exactly-once pattern of "read records, produce output, atomically commit the output plus the input offsets", that there is no supported way to run work just before a commit, and that transactions are an internal Kafka Streams concept deliberately not exposed at the API level. So the design was wrong rather than the implementation: bundles have to work without being coupled to transaction boundaries the runner does not control and cannot see.

He also corrected a mistaken note in #39633 — there is no commitOffsetNeeded flag, it is commitNeeded, and it is set after a punctuation runs, so punctuator output is not outside the commit accounting as we had written. The issue has been corrected.

The way forward is to make the flush data-driven, the way watermarks already are. A source emits a marker on the punctuator it already runs, the marker travels the topology as an ordinary record, and each stage closes its bundle when it receives one, inside process(). Bundle boundaries then never touch transaction boundaries.

What is here

A third variant alongside data and watermark:

• FlushPayload, the narrowed view, mirroring the existing WatermarkPayload.
• KStreamsPayload.flush(sourcePartition, totalSourcePartitions), with the same range validation the watermark factory has.
• The proto variant and serde support in both directions.

The marker carries the producing partition and that transform's partition count. Those exist so it can be targeted rather than broadcast. Broadcasting would deliver one flush per upstream partition, so a downstream partition would see N times more flushes than the configured interval asks for. Instead each producing partition will address a slice of the downstream partitions, and the slices tile the whole range, so every downstream partition receives exactly one flush per interval. The rule is [floor(i*D/U), floor((i+1)*D/U)) for upstream partition i, with U upstream partitions and D downstream, and it holds for fan-out, fan-in, equal counts and a single partition on either side. That logic lands in the next PR, in the partitioner.

Unlike a watermark, a flush needs no aggregation when it arrives. There is nothing to hold and nothing to combine, because exactly one arrives.

The remaining three

2. Rename GroupByKeyBroadcastPartitioner to something that covers targeting a possibly empty subset, and add the targeting rule.
3. Sources emit the marker on their existing punctuator, at maxBundleTimeMs.
4. ExecutableStageProcessor closes and flushes its bundle when a marker arrives, then forwards it on.

Testing

./gradlew -Pwith-kafka-streams-runner :runners:kafka-streams:build
./gradlew -Pwith-kafka-streams-runner :runners:kafka-streams:validatesRunner

Both pass. Four new tests cover the serde round trip, the first and last partition as the boundaries of the range check, rejection of a partition outside its range, and that a flush cannot be read as a watermark or as data. Changing one field in the serde fails two of them, so they are testing something.

[github-actions]

Assigning reviewers:

R: @kennknowles added as fallback since no labels match configuration

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).