[FLINK-23704][task] Apply review suggestsion for generating LatencyMarkers in FLIP-27 sources

Author: pnowojski

flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java

@@ -33,7 +33,6 @@
 import org.apache.flink.core.io.InputStatus;
 import org.apache.flink.core.io.SimpleVersionedSerializer;
 import org.apache.flink.metrics.MetricGroup;
-import org.apache.flink.runtime.jobgraph.OperatorID;
 import org.apache.flink.runtime.operators.coordination.OperatorEvent;
 import org.apache.flink.runtime.operators.coordination.OperatorEventGateway;
 import org.apache.flink.runtime.operators.coordination.OperatorEventHandler;
@@ -47,7 +46,6 @@
 import org.apache.flink.streaming.api.operators.source.TimestampsAndWatermarks;
 import org.apache.flink.streaming.api.operators.util.SimpleVersionedListState;
 import org.apache.flink.streaming.runtime.io.PushingAsyncDataInput;
-import org.apache.flink.streaming.runtime.streamrecord.LatencyMarker;
 import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
 import org.apache.flink.util.CollectionUtil;
 import org.apache.flink.util.FlinkRuntimeException;
@@ -58,10 +56,8 @@
 import java.io.IOException;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ScheduledFuture;
 
 import static org.apache.flink.util.Preconditions.checkNotNull;
-import static org.apache.flink.util.Preconditions.checkState;
 
 /**
  * Base source operator only used for integrating the source reader which is proposed by FLIP-27. It
@@ -136,7 +132,7 @@ public class SourceOperator<OUT, SplitT extends SourceSplit> extends AbstractStr
     /** Indicating whether the source operator has been closed. */
     private boolean closed;
 
-    private LatencyMarkerEmitter<OUT> latencyMarerEmitter;
+    private @Nullable LatencyMarkerEmitter<OUT> latencyMarerEmitter;
 
     public SourceOperator(
             FunctionWithException<SourceReaderContext, SourceReader<OUT, SplitT>, Exception>
@@ -237,19 +233,6 @@ public void open() throws Exception {
                             watermarkStrategy, getMetricGroup());
         }
 
-        latencyMarerEmitter =
-                new LatencyMarkerEmitter<>(
-                        getProcessingTimeService(),
-                        getExecutionConfig().isLatencyTrackingConfigured()
-                                ? getExecutionConfig().getLatencyTrackingInterval()
-                                : getContainingTask()
-                                        .getEnvironment()
-                                        .getTaskManagerInfo()
-                                        .getConfiguration()
-                                        .getLong(MetricOptions.LATENCY_INTERVAL),
-                        getOperatorID(),
-                        getRuntimeContext().getIndexOfThisSubtask());
-
         // restore the state if necessary.
         final List<SplitT> splits = CollectionUtil.iterableToList(readerState.get());
         if (!splits.isEmpty()) {
@@ -263,7 +246,6 @@ public void open() throws Exception {
         sourceReader.start();
 
         eventTimeLogic.startPeriodicWatermarkEmits();
-        latencyMarerEmitter.startLatencyMarkerEmit();
     }
 
     @Override
@@ -272,7 +254,7 @@ public void close() throws Exception {
             eventTimeLogic.stopPeriodicWatermarkEmits();
         }
         if (latencyMarerEmitter != null) {
-            latencyMarerEmitter.stopLatencyMarkerEmit();
+            latencyMarerEmitter.close();
         }
         if (sourceReader != null) {
             sourceReader.close();
@@ -289,7 +271,7 @@ public void dispose() throws Exception {
             sourceReader.close();
         }
         if (latencyMarerEmitter != null) {
-            latencyMarerEmitter.stopLatencyMarkerEmit();
+            latencyMarerEmitter.close();
         }
     }
 
@@ -306,11 +288,31 @@ public InputStatus emitNext(DataOutput<OUT> output) throws Exception {
 
         // this creates a batch or streaming output based on the runtime mode
         currentMainOutput = eventTimeLogic.createMainOutput(output);
-        latencyMarerEmitter.emitMainOutput(output);
+        initializeLatencyMarkerEmitter(output);
         lastInvokedOutput = output;
         return sourceReader.pollNext(currentMainOutput);
     }
 
+    private void initializeLatencyMarkerEmitter(DataOutput<OUT> output) {
+        long latencyTrackingInterval =
+                getExecutionConfig().isLatencyTrackingConfigured()
+                        ? getExecutionConfig().getLatencyTrackingInterval()
+                        : getContainingTask()
+                                .getEnvironment()
+                                .getTaskManagerInfo()
+                                .getConfiguration()
+                                .getLong(MetricOptions.LATENCY_INTERVAL);
+        if (latencyTrackingInterval > 0) {
+            latencyMarerEmitter =
+                    new org.apache.flink.streaming.api.operators.LatencyMarkerEmitter<>(
+                            getProcessingTimeService(),
+                            output::emitLatencyMarker,
+                            latencyTrackingInterval,
+                            getOperatorID(),
+                            getRuntimeContext().getIndexOfThisSubtask());
+        }
+    }
+
     @Override
     public void snapshotState(StateSnapshotContext context) throws Exception {
         long checkpointId = context.getCheckpointId();
@@ -377,80 +379,4 @@ public SourceReader<OUT, SplitT> getSourceReader() {
     ListState<SplitT> getReaderState() {
         return readerState;
     }
-
-    private static class LatencyMarkerEmitter<OUT> {
-
-        private final ProcessingTimeService timeService;
-
-        private final long latencyTrackingInterval;
-
-        private final OperatorID operatorId;
-
-        private final int subtaskIndex;
-
-        @Nullable private DataOutput<OUT> currentMainOutput;
-
-        @Nullable
-        private ScheduledFuture<?> latencyMarkerTimer;
-
-        public LatencyMarkerEmitter(
-                final ProcessingTimeService timeService,
-                long latencyTrackingInterval,
-                final OperatorID operatorId,
-                final int subtaskIndex) {
-            this.timeService = timeService;
-            this.latencyTrackingInterval = latencyTrackingInterval;
-            this.operatorId = operatorId;
-            this.subtaskIndex = subtaskIndex;
-        }
-
-        // ------------------------------------------------------------------------
-
-        public void emitMainOutput(PushingAsyncDataInput.DataOutput<OUT> output) {
-            // At the moment, we assume only one output is ever created!
-            // This assumption is strict, currently, because many of the classes in this
-            // implementation
-            // do not support re-assigning the underlying output
-            checkState(currentMainOutput == null, "Main output has already been set.");
-            currentMainOutput = output;
-        }
-
-        public void startLatencyMarkerEmit() {
-            checkState(
-                    latencyMarkerTimer == null, "Latency marker emitter has already been started");
-            if (latencyTrackingInterval == 0) {
-                // a value of zero means not activated
-                return;
-            }
-            latencyMarkerTimer =
-                    timeService.scheduleWithFixedDelay(
-                            this::triggerLatencyMarkerEmit, 0L, latencyTrackingInterval);
-        }
-
-        public void stopLatencyMarkerEmit() {
-            if (latencyMarkerTimer != null) {
-                latencyMarkerTimer.cancel(false);
-                latencyMarkerTimer = null;
-            }
-        }
-
-        void triggerLatencyMarkerEmit(@SuppressWarnings("unused") long wallClockTimestamp) {
-            if (currentMainOutput != null) {
-                try {
-                    // ProcessingTimeService callbacks are executed under the
-                    // checkpointing lock
-                    currentMainOutput.emitLatencyMarker(
-                            new LatencyMarker(
-                                    timeService.getCurrentProcessingTime(),
-                                    operatorId,
-                                    subtaskIndex));
-                } catch (Throwable t) {
-                    // we catch the Throwable here so that we don't trigger the
-                    // processing
-                    // timer services async exception handler
-                    LOG.warn("Error while emitting latency marker.", t);
-                }
-            }
-        }
-    }
 }