Fix build

Author: arunpandianp

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);

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) {}
 

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) {}
 

runners/flink/src/test/java/org/apache/beam/runners/flink/translation/wrappers/streaming/DoFnOperatorTest.java

@@ -523,6 +523,9 @@ public void finishBundle() {
                 wrappedRunner.finishBundle();
               }
 
+              @Override
+              public void finishKey() {}
+
               @Override
               public <KeyT> void onWindowExpiration(
                   BoundedWindow window, Instant timestamp, KeyT key) {

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(

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();

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);

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);

runners/spark/src/test/java/org/apache/beam/runners/spark/translation/SparkInputDataProcessorTest.java

@@ -259,6 +259,9 @@ public <KeyT> void onTimer(
     @Override
     public void finishBundle() {}
 
+    @Override
+    public void finishKey() {}
+
     @Override
     public <KeyT> void onWindowExpiration(BoundedWindow window, Instant timestamp, KeyT key) {}