junaiddshaukat opened a new pull request, #40186: URL: https://github.com/apache/beam/pull/40186
## Summary Second PR toward bundles bounded by time (#39633), after #40068. It makes the shuffle and the repartition partitioner handle a flush marker. Nothing emits markers yet, so there is no behaviour change. ## What changed - `GroupByKeyBroadcastPartitioner` is renamed to `KStreamsPayloadPartitioner`, since it no longer only broadcasts. Data is still hashed by key and watermarks still go to every partition. A flush goes to exactly the partitions it names. - `ShuffleByKeyProcessor` picks those partitions. Upstream partition `i` of `U` addresses downstream partitions `[floor(i*D/U), floor((i+1)*D/U))` of `D`. Over all `i` this covers every downstream partition exactly once, for fan-out, fan-in and equal counts, so each downstream partition gets one flush per interval and not one per upstream partition. It only depends on partition numbers, so a rebalance does not change it. - When the shuffle fans in, some upstream partitions have nothing to address (partition 0 of 10 into 8, for example). Those forward no marker. - The partitioner fails on a flush if the topic does not have `internalParallelism` partitions. The targets are computed for that count, so on a topic left by an earlier run with a different parallelism they would point at partitions that don't exist, or skip some. The topic manager reuses an existing topic without checking its partition count, so nothing catches this earlier today. Handling stale topics properly belongs to #39566. ## Order of the remaining PRs I swapped the last two compared to #40068. Today `ExecutableStageProcessor` and `WindowedGroupByKeyProcessor` would throw on a flush, because each treats anything that is not one kind as the other kind. So consumers have to handle the marker before any source emits it: 3. Consumers handle a flush: `ExecutableStageProcessor` closes its bundle and forwards the marker, `WindowedGroupByKeyProcessor` forwards it. `StageOutputProcessor` and `FlattenProcessor` already pass it through. 4. Sources emit the marker on their existing punctuator, at `maxBundleTimeMs`. One open question for step 4: a marker created by a source has no edge yet, but `KStreamsPayload.flush` does not accept empty targets. The shuffle recomputes targets anyway, so what a source puts there is ignored. I'd rather settle that in step 4 than change the payload now. ## Testing ``` ./gradlew -Pwith-kafka-streams-runner :runners:kafka-streams:build ./gradlew -Pwith-kafka-streams-runner :runners:kafka-streams:validatesRunner ``` Both pass, 119 unit tests and 59 ValidatesRunner tests. New tests: `KStreamsPayloadPartitionerTest` (watermark broadcast, flush targets, data by key, keyless data, partition count mismatch) and four in `ShuffleByKeyProcessorTest`, including one that checks every downstream partition is hit exactly once for several upstream/downstream shapes. I checked they catch real mistakes: an off-by-one in the range rule plus removing the partition count check fails five of them, and forwarding a flush with no targets fails the fan-in test. -- 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]
