[FLINK-21469][runtime] Implement advanceToEndOfEventTime for MultipleInputStreamTask

For stop with savepoint, StreamTask#advanceToEndOfEventTime() is called (in source tasks)
to advance to the max watermark. This PR implments advanceToEndOfEventTime for
MultipleInputStreamTask chained sources.

Author: curcur

flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/MultipleInputStreamTask.java

@@ -30,13 +30,16 @@
 import org.apache.flink.streaming.api.graph.StreamConfig.InputConfig;
 import org.apache.flink.streaming.api.graph.StreamEdge;
 import org.apache.flink.streaming.api.operators.MultipleInputStreamOperator;
+import org.apache.flink.streaming.api.operators.Output;
+import org.apache.flink.streaming.api.watermark.Watermark;
 import org.apache.flink.streaming.runtime.io.CheckpointBarrierHandler;
 import org.apache.flink.streaming.runtime.io.CheckpointedInputGate;
 import org.apache.flink.streaming.runtime.io.InputProcessorUtil;
 import org.apache.flink.streaming.runtime.io.StreamMultipleInputProcessorFactory;
 import org.apache.flink.streaming.runtime.io.StreamTaskSourceInput;
 import org.apache.flink.streaming.runtime.metrics.MinWatermarkGauge;
 import org.apache.flink.streaming.runtime.metrics.WatermarkGauge;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
 
 import javax.annotation.Nullable;
 
@@ -256,4 +259,11 @@ public void abortCheckpointOnBarrier(long checkpointId, Throwable cause) throws
         }
         super.abortCheckpointOnBarrier(checkpointId, cause);
     }
+
+    @Override
+    protected void advanceToEndOfEventTime() throws Exception {
+        for (Output<StreamRecord<?>> sourceOutput : operatorChain.getChainedSourceOutputs()) {
+            sourceOutput.emitWatermark(Watermark.MAX_WATERMARK);
+        }
+    }
 }

flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/OperatorChain.java

@@ -475,6 +475,12 @@ public Output<StreamRecord<?>> getChainedSourceOutput(SourceInputConfig sourceIn
         return chainedSources.get(sourceInput).getSourceOutput();
     }
 
+    public List<Output<StreamRecord<?>>> getChainedSourceOutputs() {
+        return chainedSources.values().stream()
+                .map(ChainedSource::getSourceOutput)
+                .collect(Collectors.toList());
+    }
+
     public StreamTaskSourceInput<?> getSourceTaskInput(SourceInputConfig sourceInput) {
         checkArgument(
                 chainedSources.containsKey(sourceInput),

flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/MultipleInputStreamTaskTest.java

@@ -90,6 +90,7 @@
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.contains;
 import static org.hamcrest.Matchers.containsInAnyOrder;
+import static org.hamcrest.Matchers.not;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 
@@ -510,20 +511,7 @@ public void testInputFairness() throws Exception {
     @Test
     public void testWatermark() throws Exception {
         try (StreamTaskMailboxTestHarness<String> testHarness =
-                new StreamTaskMailboxTestHarnessBuilder<>(
-                                MultipleInputStreamTask::new, BasicTypeInfo.STRING_TYPE_INFO)
-                        .modifyExecutionConfig(config -> config.enableObjectReuse())
-                        .addInput(BasicTypeInfo.STRING_TYPE_INFO, 2)
-                        .addSourceInput(
-                                new SourceOperatorFactory<>(
-                                        new MockSource(
-                                                Boundedness.CONTINUOUS_UNBOUNDED, 2, true, false),
-                                        WatermarkStrategy.forGenerator(
-                                                ctx -> new RecordToWatermarkGenerator())))
-                        .addInput(BasicTypeInfo.DOUBLE_TYPE_INFO, 2)
-                        .setupOutputForSingletonOperatorChain(
-                                new MapToStringMultipleInputOperatorFactory(3))
-                        .build()) {
+                buildWatermarkTestHarness(2, false)) {
             ArrayDeque<Object> expectedOutput = new ArrayDeque<>();
 
             int initialTime = 0;
@@ -600,20 +588,7 @@ public void testWatermark() throws Exception {
     @Test
     public void testWatermarkAndStreamStatusForwarding() throws Exception {
         try (StreamTaskMailboxTestHarness<String> testHarness =
-                new StreamTaskMailboxTestHarnessBuilder<>(
-                                MultipleInputStreamTask::new, BasicTypeInfo.STRING_TYPE_INFO)
-                        .modifyExecutionConfig(config -> config.enableObjectReuse())
-                        .addInput(BasicTypeInfo.STRING_TYPE_INFO, 2)
-                        .addSourceInput(
-                                new SourceOperatorFactory<>(
-                                        new MockSource(
-                                                Boundedness.CONTINUOUS_UNBOUNDED, 2, true, true),
-                                        WatermarkStrategy.forGenerator(
-                                                ctx -> new RecordToWatermarkGenerator())))
-                        .addInput(BasicTypeInfo.DOUBLE_TYPE_INFO, 2)
-                        .setupOutputForSingletonOperatorChain(
-                                new MapToStringMultipleInputOperatorFactory(3))
-                        .build()) {
+                buildWatermarkTestHarness(2, true)) {
             ArrayDeque<Object> expectedOutput = new ArrayDeque<>();
 
             int initialTime = 0;
@@ -676,6 +651,24 @@ public void testWatermarkAndStreamStatusForwarding() throws Exception {
         }
     }
 
+    @Test
+    public void testAdvanceToEndOfEventTime() throws Exception {
+        try (StreamTaskMailboxTestHarness<String> testHarness =
+                buildWatermarkTestHarness(2, false)) {
+            testHarness.processElement(Watermark.MAX_WATERMARK, 0, 0);
+            testHarness.processElement(Watermark.MAX_WATERMARK, 0, 1);
+
+            testHarness.getStreamTask().advanceToEndOfEventTime();
+
+            testHarness.processElement(Watermark.MAX_WATERMARK, 1, 0);
+
+            assertThat(testHarness.getOutput(), not(contains(Watermark.MAX_WATERMARK)));
+
+            testHarness.processElement(Watermark.MAX_WATERMARK, 1, 1);
+            assertThat(testHarness.getOutput(), contains(Watermark.MAX_WATERMARK));
+        }
+    }
+
     @Test
     @SuppressWarnings("unchecked")
     public void testWatermarkMetrics() throws Exception {
@@ -1026,6 +1019,27 @@ static void addSourceRecords(
                 .dispatchOperatorEvent(sourceOperatorID, new SerializedValue<>(addSplitEvent));
     }
 
+    private static StreamTaskMailboxTestHarness<String> buildWatermarkTestHarness(
+            int inputChannels, boolean readerMarkIdleOnNoSplits) throws Exception {
+        return new StreamTaskMailboxTestHarnessBuilder<>(
+                        MultipleInputStreamTask::new, BasicTypeInfo.STRING_TYPE_INFO)
+                .modifyExecutionConfig(config -> config.enableObjectReuse())
+                .addInput(BasicTypeInfo.STRING_TYPE_INFO, inputChannels)
+                .addSourceInput(
+                        new SourceOperatorFactory<>(
+                                new MockSource(
+                                        Boundedness.CONTINUOUS_UNBOUNDED,
+                                        2,
+                                        true,
+                                        readerMarkIdleOnNoSplits),
+                                WatermarkStrategy.forGenerator(
+                                        ctx -> new RecordToWatermarkGenerator())))
+                .addInput(BasicTypeInfo.DOUBLE_TYPE_INFO, inputChannels)
+                .setupOutputForSingletonOperatorChain(
+                        new MapToStringMultipleInputOperatorFactory(3))
+                .build();
+    }
+
     private static OperatorID getSourceOperatorID(
             StreamTaskMailboxTestHarness<String> testHarness, int sourceId) {
         StreamConfig.InputConfig[] inputs =