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