[GSoC 2026] Kafka Streams runner: windowed GroupByKey via ReduceFnRunner

[junaiddshaukat]

Summary

Part of #18479.

Adds real windowing (and the triggering machinery that comes with it) to the Kafka Streams runner's GroupByKey, which until now only handled the global window. The approach follows the Flink and Spark portable runners: run Beam's ReduceFnRunner on the runner side, backed by Kafka Streams state and timers. Design discussed on the dev@ list and with the mentor beforehand.

What changed

The global-window GroupByKeyProcessor is replaced by WindowedGroupByKeyProcessor, which drives ReduceFnRunner per key — the same way GroupAlsoByWindowViaWindowSetNewDoFn does — over two new backends:

• KafkaStreamsStateInternals — Beam's StateInternals over a Kafka Streams KeyValueStore. Each state cell (value, bag, combining, watermark-hold) is one store entry under a composite key key | namespace | tag, so a window's state lives in its own namespace and writes go straight to the changelogged store. Modelled on the Spark runner's SparkStateInternals.
• KafkaStreamsTimerInternals — timers persisted in a store keyed by timer identity. On each input-watermark advance the processor scans for event-time timers whose fire time has passed and replays them through ReduceFnRunner.onTimers, which is what makes windows emit their panes.

GroupByKeyTranslator hydrates the input PCollection's WindowingStrategy from the pipeline proto and wires the state and timer stores. Windowing, the default trigger, panes, allowed lateness and timestamp combiners all come from ReduceFnRunner.

Test harness: dropped the manual repartition round-trip

While testing this I found that KafkaStreamsTestRunner.roundTripInternalTopics was delivering every record through the GroupByKey repartition topic twice. TopologyTestDriver (Kafka Streams 3.9) already loops an internal repartition topic from its sink back to its source within a driver step, so the manual round-trip was redundant and double-fed the topic. The previous global-window GroupByKey masked this because it fired every key once at the terminal watermark and then latched; a correctly windowed GroupByKey emits a late pane for the duplicate delivery, which surfaced the bug.

Removed the round-trip machinery. Advancing the wall clock now drives the whole pipeline to completion, and the full runner suite confirms TopologyTestDriver loops internal topics on its own.

Results

validatesRunner goes from 44 to 49 tests:

Suite	Before	After
GroupByKeyTest$BasicTests	7	9 (timestamp combiners un-sickbayed)
GroupByKeyTest$WindowTests	0 (whole class sickbayed)	3

WindowTests.testGroupByKeyMergingWindows stays sickbayed: session (merging) windows need ReduceFnRunner's merging window set to move per-window state as windows merge, which this first pass does not implement. Fixed and sliding windows, the default trigger and timestamp combiners all work. Merging windows and richer triggers land in a follow-up.

Also adds a focused runner unit test (FixedWindowGroupByKeyTest) that groups the same key across two fixed windows, and folds the #39452 reference into the LifecycleTests sickbay comment.

Testing

./gradlew :runners:kafka-streams:validatesRunner   # 49 tests, 0 failures
./gradlew :runners:kafka-streams:build             # 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.

[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).

[je-ik]

Overall this is a huge step forward. The biggest issue is that we absolutely need an efficient lookup of timers by fire instant to avoid full scans which will be expensive.

[je-ik]

This method will be a hotspot (it is called for every state access), can we optimize it a little by avoiding length-prefixing and instead just copy bytes in memory, e.g. (pseudocode I did'nt check correctness, this is just an idea)

 byte[] namespaceBytes = namespae.stringKey().getBytes(UTF_8);
 byte[] idBytes = id.getBytes(UTF_8);
 int len = encodedKey.length + namespaceBytes.length + idBytes.length + 2 /* add a separator, e.g. / between parts to avoid any possible collision */;
 byte[] res = new byte[len];
 System.arraycopy(encodedKey -> res);
 System.arraycopy(namespaceByes -> res + encodedKey.length + 1);
 System.arraycopy(idBytes -> res + encodedKey.length + namespaceBytes.length + 2);
 res[encodedKey.length] = "/";
 res[encodedKey.length + namespaceBytes.length + 1] = "/';

This might be more efficient due to pre-allocating the whole array and then just memcpy data.

It might also be worth caching the encodedKey / namespace pair, because it will be likely called several times with identical values in sequence (e.g. use code can read state 1, state 2, timer 1, write state 2, write timer 1 - all will have same encodedKey + namespace)

[junaiddshaukat]

Done, it now builds into one exactly-sized array with System.arraycopy, and caches the key | namespace prefix, which is reused across the several tags one ReduceFnRunner turn touches. That removes the stream growth and the final copy.

I kept the length prefixes rather than a separator, though: an encoded key is arbitrary coder output and a namespace stringKey can contain any byte too, so key="a/b", ns="c" and key="a", ns="b/c" would produce the same bytes with any separator. The prefixes are also what lets the timer scan read the encoded key back out (your comment below). I moved the byte helpers into StoreKeys and wrote the reasoning there.

[je-ik]

Why we return null here? Can we add comment explaining if this is going to be fixed in future and what it depends on? Same for synchronized processing time.

[junaiddshaukat]

Added. currentOutputWatermarkTime no longer returns null — it now returns the watermark the GroupByKey has actually published, which trails the input watermark by the pending holds (the watermark-hold comment below).

currentSynchronizedProcessingTime still returns null, and I documented why: it's the slowest processing time across the job's workers, which needs cross-instance coordination our watermark reports only carry for event time. TimerInternals allows null, and nothing on the paths supported today reads it — it's for processing-time triggers, which land with processing-time timer support in the follow-up, the same work that would supply it.

[je-ik]

Because we need to be able to efficiently parse key from the stored byte[] it seems this approach is a spot on here.

[junaiddshaukat]

Thanks, kept it exactly for that reason. The scan reads the encoded key back out of the key, so the prefixes have to stay here. I wrote that down in StoreKeys so it doesn't get "optimized" away later.

[je-ik]

This looks expensive. And we call it for each watermark move, we need a way to efficiently select only timers that should expire, not iterate all timers, because there will be a timer for each key and that cen be a long list.

[junaiddshaukat]

Fixed, this was the right call. Timers are now in two stores: the identity store (key/domain/family/id/namespace → index key) for overwrite and delete, and a new index store keyed domain | fireTimestamp | identity whose value is the TimerData.

Firing is now one range() over (-inf, watermark] on the index, so the cost is proportional to the timers actually due rather than to the number of keys holding a timer, and no decode is needed to decide whether a timer is due. Timestamps are written sign-flipped big-endian so the unsigned byte order Kafka Streams compares keys by matches numeric order — negative timestamps included.

KafkaStreamsTimerInternalsTest covers the range scan, negative timestamps, replacing a timer (the old index entry must go, or it would still fire at the original time), deletion, two keys sharing the index, and processing-time timers being excluded from the event-time scan.

[je-ik]

It seems we must apply watermark holds here, if I'm not missing something.

[junaiddshaukat]

You're right, this was a real bug. ReduceFnRunner takes a hold for a buffered pane that hasn't been emitted, and we were publishing the raw input watermark, so downstream was told nothing earlier was coming while panes were still buffered those elements would then arrive late against a watermark we'd already published.

WatermarkHoldState now mirrors each hold into a hold-time-ordered index, and the GroupByKey publishes min(inputWatermark, minimum hold). Reading the minimum is the first entry of that index rather than a scan, same idea as the timer index. Holds are read after firing, since firing emits the panes and releases them.