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

runners/kafka-streams/build.gradle

@@ -87,11 +87,11 @@ dependencies {
 
 // Known-failing @ValidatesRunner tests, excluded until the feature they need lands.
 def sickbayTests = [
-  // Non-global windowing (FixedWindows, merging windows, timestamp combiners) is not supported
-  // yet; these apply a window and assert on window-derived output, hitting a GlobalWindow cast.
-  'org.apache.beam.sdk.transforms.GroupByKeyTest$WindowTests',
-  'org.apache.beam.sdk.transforms.GroupByKeyTest$BasicTests.testTimestampCombinerLatest',
-  'org.apache.beam.sdk.transforms.GroupByKeyTest$BasicTests.testTimestampCombinerEarliest',
+  // Merging (session) windows are not supported yet: ReduceFnRunner drives them through a merging
+  // window set that moves per-window state as windows merge, which this first windowing pass does
+  // not implement. Non-merging windows (fixed, sliding), the default trigger and timestamp
+  // combiners do work. Lands with the follow-up windowing PR.
+  'org.apache.beam.sdk.transforms.GroupByKeyTest$WindowTests.testGroupByKeyMergingWindows',
   // A DoFn whose @StartBundle throws never gets to report its error: SdkHarnessClient.newBundle
   // sends the ProcessBundleRequest and then blocks in GrpcDataService.createOutboundAggregator
   // waiting for the SDK harness to open its data stream, which a bundle that failed during setup
@@ -100,6 +100,7 @@ def sickbayTests = [
   // anything specific to this runner; the Flink runner sickbays all of LifecycleTests and the
   // Prism runner sickbays each of its three error tests. The @ProcessElement and @FinishBundle
   // variants do pass here, because by then the data stream is established.
+  // Tracked by https://github.com/apache/beam/issues/39452.
   'org.apache.beam.sdk.transforms.ParDoTest$LifecycleTests.testParDoWithErrorInStartBatch',
 ]
 

runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/GroupByKeyTranslator.java

@@ -22,8 +22,12 @@
 import org.apache.beam.model.pipeline.v1.RunnerApi;
 import org.apache.beam.sdk.coders.Coder;
 import org.apache.beam.sdk.coders.KvCoder;
+import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+import org.apache.beam.sdk.util.construction.RehydratedComponents;
+import org.apache.beam.sdk.util.construction.WindowingStrategyTranslation;
 import org.apache.beam.sdk.values.KV;
 import org.apache.beam.sdk.values.WindowedValues;
+import org.apache.beam.sdk.values.WindowingStrategy;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableSet;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Iterables;
 import org.apache.kafka.common.serialization.Serdes;
@@ -35,10 +39,11 @@
  * Translates the {@code beam:transform:group_by_key:v1} URN — the runner's first stateful,
  * shuffle-bearing transform.
  *
- * <p>This is the simplest GroupByKey: GlobalWindow, default trigger, no allowed lateness (per the
- * plan agreed with the mentor). Each key's values are buffered in a Kafka Streams state store and
- * emitted once as {@code KV<K, Iterable<V>>} when the watermark reaches {@link
- * org.apache.beam.sdk.transforms.windowing.BoundedWindow#TIMESTAMP_MAX_VALUE}.
+ * <p>Windowing and triggering are executed by Beam's {@link
+ * org.apache.beam.runners.core.ReduceFnRunner} inside {@link WindowedGroupByKeyProcessor}, the same
+ * way the Flink and Spark portable runners do it — so fixed/sliding windows, the default trigger,
+ * allowed lateness and timestamp combiners all work. The input PCollection's windowing strategy is
+ * hydrated from the pipeline proto and handed to the processor.
  *
  * <p>Topology added (the Beam key becomes the Kafka record key so Kafka Streams shuffles by it):
  *
@@ -49,7 +54,8 @@
  *       via {@link KStreamsPayloadSerde} and a {@link GroupByKeyBroadcastPartitioner} that hashes
  *       data by key and fans watermark reports out to every partition;
  *   <li>a {@link Topology#addSource source} reading the repartition topic back;
- *   <li>the {@link GroupByKeyProcessor} plus a persistent state store, wired to the source.
+ *   <li>the {@link WindowedGroupByKeyProcessor} plus persistent state and timer stores, wired to
+ *       the source.
  * </ul>
  *
  * <p>The repartition topic is expected to exist on the broker before the job starts (same
@@ -62,6 +68,9 @@ class GroupByKeyTranslator implements PTransformTranslator {
   static final String SINK_SUFFIX = "-repartition-sink";
   static final String SOURCE_SUFFIX = "-repartition-source";
   static final String STATE_STORE_SUFFIX = "-state";
+  static final String HOLDS_INDEX_STORE_SUFFIX = "-holds-index";
+  static final String TIMER_STORE_SUFFIX = "-timers";
+  static final String TIMER_INDEX_STORE_SUFFIX = "-timers-index";
   static final String REPARTITION_TOPIC_PREFIX = "__beam_gbk_";
 
   @Override
@@ -82,12 +91,18 @@ public void translate(
     Coder<@Nullable Object> valueCoder =
         (Coder<@Nullable Object>) (Coder<?>) kvCoder.getValueCoder();
 
+    WindowingStrategy<?, BoundedWindow> windowingStrategy =
+        hydrateWindowingStrategy(pipeline, inputPCollectionId);
+
     String parentProcessor = context.getProcessorNameForPCollection(inputPCollectionId);
 
     String shuffleName = transformId + SHUFFLE_SUFFIX;
     String sinkName = transformId + SINK_SUFFIX;
     String sourceName = transformId + SOURCE_SUFFIX;
     String stateStoreName = transformId + STATE_STORE_SUFFIX;
+    String holdsIndexStoreName = transformId + HOLDS_INDEX_STORE_SUFFIX;
+    String timerStoreName = transformId + TIMER_STORE_SUFFIX;
+    String timerIndexStoreName = transformId + TIMER_INDEX_STORE_SUFFIX;
     String repartitionTopic = repartitionTopic(transformId);
 
     KStreamsPayloadSerde<KV<Object, Object>> payloadSerde = new KStreamsPayloadSerde<>(inputCoder);
@@ -111,27 +126,70 @@ public void translate(
         payloadSerde.deserializer(),
         repartitionTopic);
 
-    // Buffer values per key and fire KV<K, Iterable<V>> at the terminal watermark. Watermark
-    // reports cross the repartition topic unchanged, so they still carry the id of the transform
-    // that produced this GroupByKey's input — the parent the shuffle is attached to.
+    // Group by key and window through Beam's ReduceFnRunner, backed by the state and timer stores.
+    // Watermark reports cross the repartition topic unchanged, so they still carry the id of the
+    // transform that produced this GroupByKey's input — the parent the shuffle is attached to.
     topology.addProcessor(
         transformId,
         () ->
-            new GroupByKeyProcessor(
+            new WindowedGroupByKeyProcessor<Object, @Nullable Object, BoundedWindow>(
                 stateStoreName,
+                holdsIndexStoreName,
+                timerStoreName,
+                timerIndexStoreName,
                 transformId,
                 ImmutableSet.of(parentProcessor),
                 keyCoder,
-                valueCoder),
+                valueCoder,
+                windowingStrategy,
+                context.getPipelineOptions()),
         sourceName);
     topology.addStateStore(
         Stores.keyValueStoreBuilder(
             Stores.persistentKeyValueStore(stateStoreName), Serdes.ByteArray(), Serdes.ByteArray()),
         transformId);
+    topology.addStateStore(
+        Stores.keyValueStoreBuilder(
+            Stores.persistentKeyValueStore(timerStoreName), Serdes.ByteArray(), Serdes.ByteArray()),
+        transformId);
+    // Indexes ordered by timestamp, so due timers and the minimum watermark hold are range scans
+    // rather than scans of every timer or every held window.
+    topology.addStateStore(
+        Stores.keyValueStoreBuilder(
+            Stores.persistentKeyValueStore(timerIndexStoreName),
+            Serdes.ByteArray(),
+            Serdes.ByteArray()),
+        transformId);
+    topology.addStateStore(
+        Stores.keyValueStoreBuilder(
+            Stores.persistentKeyValueStore(holdsIndexStoreName),
+            Serdes.ByteArray(),
+            Serdes.ByteArray()),
+        transformId);
 
     context.registerPCollectionProducer(outputPCollectionId, transformId);
   }
 
+  /** Hydrates the input PCollection's windowing strategy from the pipeline proto. */
+  private static WindowingStrategy<?, BoundedWindow> hydrateWindowingStrategy(
+      RunnerApi.Pipeline pipeline, String inputPCollectionId) {
+    RunnerApi.Components components = pipeline.getComponents();
+    String windowingStrategyId =
+        components.getPcollectionsOrThrow(inputPCollectionId).getWindowingStrategyId();
+    try {
+      @SuppressWarnings("unchecked")
+      WindowingStrategy<?, BoundedWindow> strategy =
+          (WindowingStrategy<?, BoundedWindow>)
+              WindowingStrategyTranslation.fromProto(
+                  components.getWindowingStrategiesOrThrow(windowingStrategyId),
+                  RehydratedComponents.forComponents(components));
+      return strategy;
+    } catch (Exception e) {
+      throw new IllegalStateException(
+          "Failed to hydrate GroupByKey windowing strategy " + windowingStrategyId, e);
+    }
+  }
+
   /** The internal repartition topic name for a GroupByKey transform. */
   static String repartitionTopic(String transformId) {
     return REPARTITION_TOPIC_PREFIX + transformId.replaceAll("[^a-zA-Z0-9._-]", "_");