[GSoC 2026] Kafka Streams runner: CombineTest coverage and two review follow-ups

[junaiddshaukat]

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

[gemini-code-assist]

Caution

The consumer version of Gemini Code Assist on GitHub has been sunset. All code review activity has officially ceased.

[je-ik]

Good to see the combine tests pass, I have left a few questions.

[je-ik]

Should we test how the topology looks-like, not only that it is non-empty?

[junaiddshaukat]

Agreed, isEmpty() == false would have passed no matter what the fuser did, so it wasn't testing the thing the file exists for. Both now assert the shape: that no processor node stands for the Flatten, and that the branches stay in three separate subtopologies instead of being merged into one. That is the actual claim, since a Flatten node over branches of differing parallelism is exactly the case where the smaller branch can't reach all instances and the rest stall.

The second test asserts the same shape deliberately, because whether the Flatten is fused is a property of the fused graph rather than of the parallelism — which is what makes raising the parallelism safe. I printed the translated topology to write these, and there's no Flatten node in either case; both branches just end.

[je-ik]

Same here?