junaiddshaukat opened a new pull request, #39610: URL: https://github.com/apache/beam/pull/39610
## Summary Part of #18479. Enables `CombineTest` in the ValidatesRunner suite, and picks up two review follow-ups from #39546 and #39578. ## Combine coverage Combine has always been expected to work here without a translator of its own: the fuser expands `Combine.perKey` into a GroupByKey with the combining logic running as ordinary ParDos in the SDK harness, all of which the runner already executes. That was an assumption though — nothing exercised it. Enabling `CombineTest` takes the suite from 49 to 59 tests and makes it a tested claim. | Suite | Tests | | --- | --- | | `CombineTest$BasicTests` | 8 | | `CombineTest$WindowingTests` | 2 | `BasicTests` passes in full, including hot-key fanout and the accumulation-mode variant; `WindowingTests` contributes the fixed-window and empty-window cases. The rest falls out on category excludes the task already declares — `CombineWithContextTests` and `AccumulationTests` need side inputs, and most of `WindowingTests` needs side inputs, triggers or `TestStream`. `CombineTest$WindowingTests.testSessionsCombine` is sickbayed. Session windows are merging windows, which the first windowing pass did not implement, so it joins the existing `testGroupByKeyMergingWindows` entry under a comment now worded to cover Combine too. That is the *only* Combine failure, and it is the known windowing gap rather than anything specific to Combine. ## Flatten: what the `Math.max` is actually doing The review question on #39546 was whether the `Math.max` over the inputs' partition counts is redundant, since the comment above it asserted the inputs are co-partitioned. Neither half of that was quite right, so both are now fixed. The max is not a no-op in principle: Kafka Streams merges the subtopologies of every parent a processor is wired to and gives the merged subtopology as many tasks as its largest source topic has partitions, so the max is what that task count comes to. But the mismatched case does not reach this translator at all. A Flatten whose branches would disagree — one through a GroupByKey, one straight from a source — is folded into the SDK harness stage by the fuser rather than becoming a node here, and the runner Flattens that do arrive come from the fuser deduplicating partial outputs of a single PCollection. I tried to build the mismatched shape to see what the runner does with it, and could not: it never becomes a runner Flatten. `FlattenParallelismTest` records that, so if a change ever lets that shape through, it starts failing and the partition-count handling gets revisited. The comment now describes the situation instead of asserting an invariant nothing enforces. Worth stating why this is not just tidying. If such a Flatten ever did reach the translator, the branch with fewer partitions would only produce on that many of the merged subtopology's tasks, and the remaining Flatten instances would wait forever for a watermark report from it. A stalled pipeline is a bad failure mode to leave undetected, which is what the test is guarding. ## Partitioner: guard the null key `GroupByKeyBroadcastPartitioner.partition()` guarded against a null record key, but `partitions()` — the method Kafka Streams actually calls when it is present — hashed it unguarded. Nothing reaches that today, because data arriving at a repartition sink has been re-keyed by `ShuffleByKeyProcessor` first, but the guard belongs on the method that runs. That is the groundwork for the other half of the review point: that a stateless stage should carry a null key rather than the empty-array placeholder Impulse and Read emit. That change is more invasive than it looks — every processor is declared `Processor<byte[], …, byte[], …>`, so emitting a null key means moving all of them, plus the payload serde and the partitioner generics, to `byte @Nullable []`, or adding nullness suppressions in several places. It is worth doing, but as its own change rather than folded in here. ## Testing ``` ./gradlew :runners:kafka-streams:validatesRunner # 59 tests, 0 failures ./gradlew :runners:kafka-streams:build # 82 unit tests, spotless + checker + errorprone ``` -- 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]
