Fix build

Author: arunpandianp

runners/flink/src/main/java/org/apache/beam/runners/flink/metrics/DoFnRunnerWithMetricsUpdate.java

@@ -105,6 +105,16 @@ public void finishBundle() {
     container.updateMetrics(stepName);
   }
 
+  @Override
+  public void finishKey() {
+    try (Closeable ignored =
+        MetricsEnvironment.scopedMetricsContainer(container.getMetricsContainer(stepName))) {
+      delegate.finishKey();
+    } catch (IOException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
   @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,11 @@ public void finishBundle() {
       }
     }
 
+    @Override
+    public void finishKey() {
+      // Do nothing by default.
+    }
+
     @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,11 @@ public void finishBundle() {
     Optional.ofNullable(finishBundleCallback).ifPresent(Runnable::run);
   }
 
+  @Override
+  public void finishKey() {
+    // Do nothing by default.
+  }
+
   @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(