Fix build

arunpandianp · arunpandianp · commit d4242c35e333 · 2026-05-12T03:16:53.000Z
diff --git a/runners/flink/src/main/java/org/apache/beam/runners/flink/metrics/DoFnRunnerWithMetricsUpdate.java b/runners/flink/src/main/java/org/apache/beam/runners/flink/metrics/DoFnRunnerWithMetricsUpdate.java
@@ -105,6 +105,9 @@ public void finishBundle() {
     container.updateMetrics(stepName);
   }
 
+  @Override
+  public void finishKey() {}
+
   @Override
   public <KeyT> void onWindowExpiration(BoundedWindow window, Instant timestamp, KeyT key) {
     delegate.onWindowExpiration(window, timestamp, key);
diff --git a/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/ExecutableStageDoFnOperator.java b/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/ExecutableStageDoFnOperator.java
@@ -1064,6 +1064,9 @@ public void finishBundle() {
       }
     }
 
+    @Override
+    public void finishKey() {}
+
     @Override
     public <KeyT> void onWindowExpiration(BoundedWindow window, Instant timestamp, KeyT key) {}
 
diff --git a/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/stableinput/BufferingDoFnRunner.java b/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/stableinput/BufferingDoFnRunner.java
@@ -255,6 +255,9 @@ public void finishBundle() {
     Optional.ofNullable(finishBundleCallback).ifPresent(Runnable::run);
   }
 
+  @Override
+  public void finishKey() {}
+
   @Override
   public <KeyT> void onWindowExpiration(BoundedWindow window, Instant timestamp, KeyT key) {}
 
diff --git a/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/StreamingSideInputDoFnRunnerTest.java b/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/StreamingSideInputDoFnRunnerTest.java
@@ -150,6 +150,7 @@ public void testSideInputNotReady() throws Exception {
 
     runner.startBundle();
     runner.processElement(createDatum("e", 0));
+    runner.finishKey();
     runner.finishBundle();
 
     assertTrue(outputManager.getOutput(mainOutputTag).isEmpty());
@@ -214,6 +215,7 @@ public void testMultipleWindowsNotReady() throws Exception {
 
     runner.startBundle();
     runner.processElement(elem);
+    runner.finishKey();
     runner.finishBundle();
 
     assertTrue(outputManager.getOutput(mainOutputTag).isEmpty());
@@ -317,6 +319,7 @@ public void testSideInputNotification() throws Exception {
     when(mockSideInputReader.get(eq(view), any(BoundedWindow.class))).thenReturn("data");
 
     runner.startBundle();
+    runner.finishKey();
     runner.finishBundle();
 
     assertThat(outputManager.getOutput(mainOutputTag), contains(createDatum("e:data", 0)));
@@ -373,6 +376,7 @@ public void testMultipleSideInputs() throws Exception {
 
     runner.startBundle();
     runner.processElement(createDatum("e2", 2));
+    runner.finishKey();
     runner.finishBundle();
 
     assertThat(
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/DoFnRunnerFactory.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/DoFnRunnerFactory.java
@@ -284,6 +284,9 @@ public void finishBundle() {
         }
       }
 
+      @Override
+      public void finishKey() {}
+
       @Override
       public DoFn<InT, T> getFn() {
         throw new UnsupportedOperationException();
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/DoFnRunnerWithMetrics.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/DoFnRunnerWithMetrics.java
@@ -104,6 +104,9 @@ public void finishBundle() {
     }
   }
 
+  @Override
+  public void finishKey() {}
+
   @Override
   public <KeyT> void onWindowExpiration(BoundedWindow window, Instant timestamp, KeyT key) {
     delegate.onWindowExpiration(window, timestamp, key);
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/DoFnRunnerWithMetrics.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/DoFnRunnerWithMetrics.java
@@ -103,6 +103,9 @@ public void finishBundle() {
     }
   }
 
+  @Override
+  public void finishKey() {}
+
   @Override
   public <KeyT> void onWindowExpiration(BoundedWindow window, Instant timestamp, KeyT key) {
     delegate.onWindowExpiration(window, timestamp, key);

Original file line number	Diff line number	Diff line change
`@@ -105,6 +105,9 @@ public void finishBundle() {`
`105`	`105`	`container.updateMetrics(stepName);`
`106`	`106`	`}`
`107`	`107`
	`108`	`+ @Override`
	`109`	`+ public void finishKey() {}`
	`110`	`+`
`108`	`111`	`@Override`
`109`	`112`	`public <KeyT> void onWindowExpiration(BoundedWindow window, Instant timestamp, KeyT key) {`
`110`	`113`	`delegate.onWindowExpiration(window, timestamp, key);`
Original file line number	Diff line number	Diff line change
`@@ -1064,6 +1064,9 @@ public void finishBundle() {`
`1064`	`1064`	`}`
`1065`	`1065`	`}`
`1066`	`1066`
	`1067`	`+ @Override`
	`1068`	`+ public void finishKey() {}`
	`1069`	`+`
`1067`	`1070`	`@Override`
`1068`	`1071`	`public <KeyT> void onWindowExpiration(BoundedWindow window, Instant timestamp, KeyT key) {}`
`1069`	`1072`
Original file line number	Diff line number	Diff line change
`@@ -255,6 +255,9 @@ public void finishBundle() {`
`255`	`255`	`Optional.ofNullable(finishBundleCallback).ifPresent(Runnable::run);`
`256`	`256`	`}`
`257`	`257`
	`258`	`+ @Override`
	`259`	`+ public void finishKey() {}`
	`260`	`+`
`258`	`261`	`@Override`
`259`	`262`	`public <KeyT> void onWindowExpiration(BoundedWindow window, Instant timestamp, KeyT key) {}`
`260`	`263`
Original file line number	Diff line number	Diff line change
`@@ -284,6 +284,9 @@ public void finishBundle() {`
`284`	`284`	`}`
`285`	`285`	`}`
`286`	`286`
	`287`	`+ @Override`
	`288`	`+ public void finishKey() {}`
	`289`	`+`
`287`	`290`	`@Override`
`288`	`291`	`public DoFn<InT, T> getFn() {`
`289`	`292`	`throw new UnsupportedOperationException();`
Original file line number	Diff line number	Diff line change
`@@ -104,6 +104,9 @@ public void finishBundle() {`
`104`	`104`	`}`
`105`	`105`	`}`
`106`	`106`
	`107`	`+ @Override`
	`108`	`+ public void finishKey() {}`
	`109`	`+`
`107`	`110`	`@Override`
`108`	`111`	`public <KeyT> void onWindowExpiration(BoundedWindow window, Instant timestamp, KeyT key) {`
`109`	`112`	`delegate.onWindowExpiration(window, timestamp, key);`
Original file line number	Diff line number	Diff line change
`@@ -103,6 +103,9 @@ public void finishBundle() {`
`103`	`103`	`}`
`104`	`104`	`}`
`105`	`105`
	`106`	`+ @Override`
	`107`	`+ public void finishKey() {}`
	`108`	`+`
`106`	`109`	`@Override`
`107`	`110`	`public <KeyT> void onWindowExpiration(BoundedWindow window, Instant timestamp, KeyT key) {`
`108`	`111`	`delegate.onWindowExpiration(window, timestamp, key);`