junaiddshaukat opened a new pull request, #40068: URL: https://github.com/apache/beam/pull/40068
## 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. -- 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]
