currentWatermark) {
+ return new InternalSourceSplitMetricGroup(
+ operatorMetricGroup, clock, splitId, currentWatermark);
+ }
+
+ public void markSplitStart() {
+ splitStartTime = clock.absoluteTimeMillis();
+ }
+
+ public void maybeMarkSplitStart() {
+ if (splitStartTime == SPLIT_NOT_STARTED) {
+ markSplitStart();
+ }
+ }
+
+ public long getCurrentWatermark() {
+ return this.currentWatermarkGauge.getValue();
+ }
+
+ public void markPaused() {
+ maybeMarkSplitStart();
+ if (isIdle()) {
+ // If a split got paused it means it emitted records,
+ // hence it shouldn't be considered idle anymore
+ markNotIdle();
+ LOG.warn("[{}] Split marked paused while still idle", splitId);
+ }
+ this.pausedTimePerSecond.markStart();
+ }
+
+ public void markIdle() {
+ maybeMarkSplitStart();
+ if (isPaused()) {
+ markNotPaused();
+ // This is benign: idleness takes over paused state if they race
+ LOG.info("[{}] Split marked idle while still paused", splitId);
+ }
+ this.idleTimePerSecond.markStart();
+ }
+
+ public void markNotPaused() {
+ maybeMarkSplitStart();
+ this.pausedTimePerSecond.markEnd();
+ }
+
+ public void markNotIdle() {
+ maybeMarkSplitStart();
+ this.idleTimePerSecond.markEnd();
+ }
+
+ public double getActiveTimePerSecond() {
+ if (splitStartTime == SPLIT_NOT_STARTED) {
+ return 0L;
+ }
+ double activeTimePerSecond = 1000.0 - getPausedTimePerSecond() - getIdleTimePerSecond();
+ return Math.max(activeTimePerSecond, 0);
+ }
+
+ public double getAccumulatedActiveTime() {
+ if (splitStartTime == SPLIT_NOT_STARTED) {
+ return 0L;
+ }
+ return Math.max(
+ clock.absoluteTimeMillis()
+ - splitStartTime
+ - getAccumulatedPausedTime()
+ - getAccumulatedIdleTime(),
+ 0);
+ }
+
+ public long getAccumulatedIdleTime() {
+ return idleTimePerSecond.getAccumulatedCount();
+ }
+
+ public long getIdleTimePerSecond() {
+ return idleTimePerSecond.getValue();
+ }
+
+ public long getPausedTimePerSecond() {
+ return pausedTimePerSecond.getValue();
+ }
+
+ public long getAccumulatedPausedTime() {
+ return pausedTimePerSecond.getAccumulatedCount();
+ }
+
+ public Boolean isPaused() {
+ return pausedTimePerSecond.isMeasuring();
+ }
+
+ public Boolean isIdle() {
+ return idleTimePerSecond.isMeasuring();
+ }
+
+ public Boolean isActive() {
+ return !isPaused() && !isIdle();
+ }
+
+ public void onSplitFinished() {
+ if (splitWatermarkMetricGroup instanceof AbstractMetricGroup) {
+ ((AbstractMetricGroup) splitWatermarkMetricGroup).close();
+ } else {
+ if (splitWatermarkMetricGroup != null) {
+ LOG.warn(
+ "Split watermark metric group can not be closed, expecting an instance of AbstractMetricGroup but got: ",
+ splitWatermarkMetricGroup.getClass().getName());
+ }
+ }
+ }
+
+ @VisibleForTesting
+ public MetricGroup getSplitWatermarkMetricGroup() {
+ return splitWatermarkMetricGroup;
+ }
+
+ @VisibleForTesting
+ public void updateTimers() {
+ this.idleTimePerSecond.update();
+ this.pausedTimePerSecond.update();
+ }
+}
diff --git a/flink-sql-runner/src/main/java/org/apache/flink/runtime/metrics/groups/package-info.java b/flink-sql-runner/src/main/java/org/apache/flink/runtime/metrics/groups/package-info.java
new file mode 100644
index 00000000..c02846c4
--- /dev/null
+++ b/flink-sql-runner/src/main/java/org/apache/flink/runtime/metrics/groups/package-info.java
@@ -0,0 +1,24 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * Backporting FLINK-40093 bugfix until Flink 2.3.1 released.
+ *
+ * @see FLINK-40093
+ */
+package org.apache.flink.runtime.metrics.groups;
diff --git a/flink-sql-runner/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java b/flink-sql-runner/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java
new file mode 100644
index 00000000..c487aba5
--- /dev/null
+++ b/flink-sql-runner/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java
@@ -0,0 +1,985 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.api.operators;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.eventtime.WatermarkAlignmentParams;
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeutils.base.array.BytePrimitiveArraySerializer;
+import org.apache.flink.api.connector.source.ReaderOutput;
+import org.apache.flink.api.connector.source.RichSourceReaderContext;
+import org.apache.flink.api.connector.source.SourceEvent;
+import org.apache.flink.api.connector.source.SourceReader;
+import org.apache.flink.api.connector.source.SourceReaderContext;
+import org.apache.flink.api.connector.source.SourceSplit;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.MetricOptions;
+import org.apache.flink.core.io.InputStatus;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.metrics.groups.SourceReaderMetricGroup;
+import org.apache.flink.runtime.event.WatermarkEvent;
+import org.apache.flink.runtime.io.AvailabilityProvider;
+import org.apache.flink.runtime.io.network.api.StopMode;
+import org.apache.flink.runtime.metrics.groups.InternalSourceReaderMetricGroup;
+import org.apache.flink.runtime.metrics.groups.InternalSourceSplitMetricGroup;
+import org.apache.flink.runtime.metrics.groups.TaskIOMetricGroup;
+import org.apache.flink.runtime.operators.coordination.OperatorEvent;
+import org.apache.flink.runtime.operators.coordination.OperatorEventGateway;
+import org.apache.flink.runtime.operators.coordination.OperatorEventHandler;
+import org.apache.flink.runtime.source.event.AddSplitEvent;
+import org.apache.flink.runtime.source.event.IsProcessingBacklogEvent;
+import org.apache.flink.runtime.source.event.NoMoreSplitsEvent;
+import org.apache.flink.runtime.source.event.ReaderRegistrationEvent;
+import org.apache.flink.runtime.source.event.ReportedWatermarkEvent;
+import org.apache.flink.runtime.source.event.RequestSplitEvent;
+import org.apache.flink.runtime.source.event.SourceEventWrapper;
+import org.apache.flink.runtime.source.event.WatermarkAlignmentEvent;
+import org.apache.flink.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+import org.apache.flink.streaming.api.graph.StreamConfig;
+import org.apache.flink.streaming.api.operators.source.TimestampsAndWatermarks;
+import org.apache.flink.streaming.api.operators.source.WatermarkSampler;
+import org.apache.flink.streaming.api.operators.util.PausableRelativeClock;
+import org.apache.flink.streaming.api.operators.util.SimpleVersionedListState;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.io.DataInputStatus;
+import org.apache.flink.streaming.runtime.io.MultipleFuturesAvailabilityHelper;
+import org.apache.flink.streaming.runtime.io.PushingAsyncDataInput;
+import org.apache.flink.streaming.runtime.streamrecord.RecordAttributesBuilder;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
+import org.apache.flink.streaming.runtime.tasks.StreamTask;
+import org.apache.flink.streaming.runtime.tasks.StreamTask.CanEmitBatchOfRecordsChecker;
+import org.apache.flink.util.CollectionUtil;
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.UserCodeClassLoader;
+import org.apache.flink.util.function.FunctionWithException;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+
+import static org.apache.flink.configuration.PipelineOptions.ALLOW_UNALIGNED_SOURCE_SPLITS;
+import static org.apache.flink.configuration.PipelineOptions.WATERMARK_ALIGNMENT_BUFFER_SIZE;
+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
+ * implements the interface of {@link PushingAsyncDataInput} which is naturally compatible with one
+ * input processing in runtime stack.
+ *
+ * Important Note on Serialization: The SourceOperator inherits the {@link
+ * java.io.Serializable} interface from the StreamOperator, but is in fact NOT serializable. The
+ * operator must only be instantiated in the StreamTask from its factory.
+ *
+ * @param The output type of the operator.
+ */
+@Internal
+public class SourceOperator extends AbstractStreamOperator
+ implements OperatorEventHandler,
+ PushingAsyncDataInput,
+ TimestampsAndWatermarks.WatermarkUpdateListener {
+ private static final long serialVersionUID = 1405537676017904695L;
+
+ // Package private for unit test.
+ static final ListStateDescriptor SPLITS_STATE_DESC =
+ new ListStateDescriptor<>("SourceReaderState", BytePrimitiveArraySerializer.INSTANCE);
+
+ /**
+ * The factory for the source reader. This is a workaround, because currently the SourceReader
+ * must be lazily initialized, which is mainly because the metrics groups that the reader relies
+ * on is lazily initialized.
+ */
+ private final FunctionWithException, Exception>
+ readerFactory;
+
+ /**
+ * The serializer for the splits, applied to the split types before storing them in the reader
+ * state.
+ */
+ private final SimpleVersionedSerializer splitSerializer;
+
+ /** The event gateway through which this operator talks to its coordinator. */
+ private final OperatorEventGateway operatorEventGateway;
+
+ /** The factory for timestamps and watermark generators. */
+ private final WatermarkStrategy watermarkStrategy;
+
+ private final WatermarkAlignmentParams watermarkAlignmentParams;
+
+ /** The Flink configuration. */
+ private final Configuration configuration;
+
+ /**
+ * Host name of the machine where the operator runs, to support locality aware work assignment.
+ */
+ private final String localHostname;
+
+ /** Whether to emit intermediate watermarks or only one final watermark at the end of input. */
+ private final boolean emitProgressiveWatermarks;
+
+ // ---- lazily initialized fields (these fields are the "hot" fields) ----
+
+ /** The source reader that does most of the work. */
+ private SourceReader sourceReader;
+
+ private ReaderOutput currentMainOutput;
+
+ private DataOutput lastInvokedOutput;
+
+ private boolean idle = false;
+
+ /** The state that holds the currently assigned splits. */
+ private ListState readerState;
+
+ /**
+ * The event time and watermarking logic. Ideally this would be eagerly passed into this
+ * operator, but we currently need to instantiate this lazily, because the metric groups exist
+ * only later.
+ */
+ private TimestampsAndWatermarks eventTimeLogic;
+
+ /** A mode to control the behaviour of the {@link #emitNext(DataOutput)} method. */
+ private OperatingMode operatingMode;
+
+ /** The timestamp when {#operatingMode} was last changed. */
+ private long operatingModeChangeTs;
+
+ private final CompletableFuture finished = new CompletableFuture<>();
+ private final SourceOperatorAvailabilityHelper availabilityHelper =
+ new SourceOperatorAvailabilityHelper();
+
+ private final List splitsToInitializeOutput = new ArrayList<>();
+
+ private final Set currentlyPausedSplits = new HashSet<>();
+ private final Set currentlyIdleSplits = new HashSet<>();
+
+ private boolean waitingForCheckpoint;
+
+ private final CompletableFuture checkpointsStartedFuture;
+
+ private final Map splitMetricGroups = new HashMap<>();
+
+ private enum OperatingMode {
+ READING,
+ WAITING_FOR_ALIGNMENT,
+ OUTPUT_NOT_INITIALIZED,
+ SOURCE_DRAINED,
+ SOURCE_STOPPED,
+ DATA_FINISHED
+ }
+
+ private InternalSourceReaderMetricGroup sourceMetricGroup;
+
+ private long currentMaxDesiredWatermark = Watermark.MAX_WATERMARK.getTimestamp();
+
+ /**
+ * {@link #currentMaxDesiredWatermark} is checked against the minimum of sample split
+ * watermarks.
+ */
+ private final Map sampledSplitWatermarks = new HashMap<>();
+
+ private final WatermarkSampler sampledLatestWatermark;
+
+ /** Can be not completed only in {@link OperatingMode#WAITING_FOR_ALIGNMENT} mode. */
+ private CompletableFuture waitingForAlignmentFuture =
+ CompletableFuture.completedFuture(null);
+
+ private @Nullable LatencyMarkerEmitter latencyMarkerEmitter;
+
+ private final boolean allowUnalignedSourceSplits;
+
+ private final int watermarkBufferSize;
+
+ private final CanEmitBatchOfRecordsChecker canEmitBatchOfRecords;
+
+ /**
+ * {@link PausableRelativeClock} tracking activity of the operator's main input. It's paused on
+ * backpressure. Note, each split output has its own independent {@link PausableRelativeClock}.
+ */
+ private transient PausableRelativeClock mainInputActivityClock;
+
+ /** Watermark identifier to whether the watermark are aligned. */
+ private final Map watermarkIsAlignedMap;
+
+ private final boolean supportsSplitReassignmentOnRecovery;
+
+ public SourceOperator(
+ StreamOperatorParameters parameters,
+ FunctionWithException, Exception>
+ readerFactory,
+ OperatorEventGateway operatorEventGateway,
+ SimpleVersionedSerializer splitSerializer,
+ WatermarkStrategy watermarkStrategy,
+ ProcessingTimeService timeService,
+ Configuration configuration,
+ String localHostname,
+ boolean emitProgressiveWatermarks,
+ CanEmitBatchOfRecordsChecker canEmitBatchOfRecords,
+ Map watermarkIsAlignedMap,
+ boolean supportsSplitReassignmentOnRecovery,
+ boolean pauseSourcesUntilFirstCheckpoint) {
+ super(parameters);
+ this.watermarkBufferSize = configuration.get(WATERMARK_ALIGNMENT_BUFFER_SIZE);
+ this.sampledLatestWatermark = new WatermarkSampler(watermarkBufferSize);
+ this.readerFactory = checkNotNull(readerFactory);
+ this.operatorEventGateway = checkNotNull(operatorEventGateway);
+ this.splitSerializer = checkNotNull(splitSerializer);
+ this.watermarkStrategy = checkNotNull(watermarkStrategy);
+ this.processingTimeService = timeService;
+ this.configuration = checkNotNull(configuration);
+ this.localHostname = checkNotNull(localHostname);
+ this.emitProgressiveWatermarks = emitProgressiveWatermarks;
+ setOperatingMode(OperatingMode.OUTPUT_NOT_INITIALIZED);
+ this.watermarkAlignmentParams = watermarkStrategy.getAlignmentParameters();
+ this.allowUnalignedSourceSplits = configuration.get(ALLOW_UNALIGNED_SOURCE_SPLITS);
+ this.canEmitBatchOfRecords = checkNotNull(canEmitBatchOfRecords);
+ this.watermarkIsAlignedMap = watermarkIsAlignedMap;
+ this.supportsSplitReassignmentOnRecovery = supportsSplitReassignmentOnRecovery;
+ this.waitingForCheckpoint = pauseSourcesUntilFirstCheckpoint;
+ //noinspection unchecked
+ this.checkpointsStartedFuture =
+ waitingForCheckpoint
+ ? new CompletableFuture<>()
+ : (CompletableFuture) AVAILABLE;
+ LOG.info("SourceOperator initialized, wait for 1st checkpoint: {}", waitingForCheckpoint);
+ }
+
+ @Override
+ protected void setup(
+ StreamTask containingTask,
+ StreamConfig config,
+ Output> output) {
+ super.setup(containingTask, config, output);
+ initSourceMetricGroup();
+ // Metric "numRecordsIn" & "numBytesIn" is defined as the total number of records/bytes
+ // read from the external system in FLIP-33, reuse them for task to account for traffic
+ // with external system
+ this.metrics.getIOMetricGroup().reuseInputMetricsForTask();
+ this.metrics.getIOMetricGroup().reuseBytesInputMetricsForTask();
+ }
+
+ @VisibleForTesting
+ protected void initSourceMetricGroup() {
+ sourceMetricGroup = InternalSourceReaderMetricGroup.wrap(getMetricGroup());
+ }
+
+ /**
+ * Initializes the reader. The code from this method should ideally happen in the constructor or
+ * in the operator factory even. It has to happen here at a slightly later stage, because of the
+ * lazy metric initialization.
+ *
+ * Calling this method explicitly is an optional way to have the reader initialization a bit
+ * earlier than in open(), as needed by the {@link
+ * org.apache.flink.streaming.runtime.tasks.SourceOperatorStreamTask}
+ *
+ *
This code should move to the constructor once the metric groups are available at task
+ * setup time.
+ */
+ public void initReader() throws Exception {
+ if (sourceReader != null) {
+ return;
+ }
+
+ StreamingRuntimeContext runtimeContext = getRuntimeContext();
+ final int subtaskIndex = runtimeContext.getTaskInfo().getIndexOfThisSubtask();
+
+ final RichSourceReaderContext context =
+ new RichSourceReaderContext() {
+ @Override
+ public SourceReaderMetricGroup metricGroup() {
+ return sourceMetricGroup;
+ }
+
+ @Override
+ public Configuration getConfiguration() {
+ return configuration;
+ }
+
+ @Override
+ public String getLocalHostName() {
+ return localHostname;
+ }
+
+ @Override
+ public int getIndexOfSubtask() {
+ return subtaskIndex;
+ }
+
+ @Override
+ public void sendSplitRequest() {
+ operatorEventGateway.sendEventToCoordinator(
+ new RequestSplitEvent(getLocalHostName()));
+ }
+
+ @Override
+ public void sendSourceEventToCoordinator(SourceEvent event) {
+ operatorEventGateway.sendEventToCoordinator(new SourceEventWrapper(event));
+ }
+
+ @Override
+ public UserCodeClassLoader getUserCodeClassLoader() {
+ return new UserCodeClassLoader() {
+ @Override
+ public ClassLoader asClassLoader() {
+ return getRuntimeContext().getUserCodeClassLoader();
+ }
+
+ @Override
+ public void registerReleaseHookIfAbsent(
+ String releaseHookName, Runnable releaseHook) {
+ getRuntimeContext()
+ .registerUserCodeClassLoaderReleaseHookIfAbsent(
+ releaseHookName, releaseHook);
+ }
+ };
+ }
+
+ @Override
+ public int currentParallelism() {
+ return getRuntimeContext().getTaskInfo().getNumberOfParallelSubtasks();
+ }
+
+ @Override
+ public void emitWatermark(
+ org.apache.flink.api.common.watermark.Watermark watermark) {
+ checkState(watermarkIsAlignedMap.containsKey(watermark.getIdentifier()));
+ output.emitWatermark(
+ new WatermarkEvent(
+ watermark,
+ watermarkIsAlignedMap.get(watermark.getIdentifier())));
+ }
+
+ @Override
+ public RuntimeContext getRuntimeContext() {
+ return runtimeContext;
+ }
+ };
+
+ sourceReader = readerFactory.apply(context);
+ }
+
+ public InternalSourceReaderMetricGroup getSourceMetricGroup() {
+ return sourceMetricGroup;
+ }
+
+ protected InternalSourceSplitMetricGroup getOrCreateSplitMetricGroup(String splitId) {
+ sampledSplitWatermarks.computeIfAbsent(
+ splitId, k -> new WatermarkSampler(watermarkBufferSize));
+ if (!this.splitMetricGroups.containsKey(splitId)) {
+ InternalSourceSplitMetricGroup splitMetricGroup =
+ InternalSourceSplitMetricGroup.wrap(
+ getMetricGroup(),
+ processingTimeService.getClock(),
+ splitId,
+ () -> sampledSplitWatermarks.get(splitId).getLatest());
+ splitMetricGroup.markSplitStart();
+ this.splitMetricGroups.put(splitId, splitMetricGroup);
+ }
+ return this.splitMetricGroups.get(splitId);
+ }
+
+ @VisibleForTesting
+ public InternalSourceSplitMetricGroup getSplitMetricGroup(String splitId) {
+ return this.splitMetricGroups.get(splitId);
+ }
+
+ @Override
+ public void open() throws Exception {
+ mainInputActivityClock = new PausableRelativeClock(getProcessingTimeService().getClock());
+ TaskIOMetricGroup taskIOMetricGroup =
+ getContainingTask().getEnvironment().getMetricGroup().getIOMetricGroup();
+ taskIOMetricGroup.registerBackPressureListener(mainInputActivityClock);
+
+ initReader();
+
+ // in the future when we this one is migrated to the "eager initialization" operator
+ // (StreamOperatorV2), then we should evaluate this during operator construction.
+ if (emitProgressiveWatermarks) {
+ eventTimeLogic =
+ TimestampsAndWatermarks.createProgressiveEventTimeLogic(
+ watermarkStrategy,
+ sourceMetricGroup,
+ getProcessingTimeService(),
+ getExecutionConfig().getAutoWatermarkInterval(),
+ mainInputActivityClock,
+ getProcessingTimeService().getClock(),
+ taskIOMetricGroup);
+ } else {
+ eventTimeLogic =
+ TimestampsAndWatermarks.createNoOpEventTimeLogic(
+ watermarkStrategy, sourceMetricGroup, mainInputActivityClock);
+ }
+
+ // restore the state if necessary.
+ final List splits = CollectionUtil.iterableToList(readerState.get());
+ if (!splits.isEmpty() && !supportsSplitReassignmentOnRecovery) {
+ LOG.info("Restoring state for {} split(s) to reader.", splits.size());
+ for (SplitT s : splits) {
+ getOrCreateSplitMetricGroup(s.splitId());
+ }
+ splitsToInitializeOutput.addAll(splits);
+ sourceReader.addSplits(splits);
+ }
+
+ // Register the reader to the coordinator.
+ registerReader(supportsSplitReassignmentOnRecovery ? splits : Collections.emptyList());
+
+ sourceMetricGroup.idlingStarted();
+ // Start the reader after registration, sending messages in start is allowed.
+ sourceReader.start();
+
+ eventTimeLogic.startPeriodicWatermarkEmits();
+ }
+
+ @Override
+ public void finish() throws Exception {
+ stopInternalServices();
+ super.finish();
+
+ finished.complete(null);
+ }
+
+ private void stopInternalServices() {
+ if (eventTimeLogic != null) {
+ eventTimeLogic.stopPeriodicWatermarkEmits();
+ }
+ if (latencyMarkerEmitter != null) {
+ latencyMarkerEmitter.close();
+ }
+ }
+
+ public CompletableFuture stop(StopMode mode) {
+ switch (operatingMode) {
+ case WAITING_FOR_ALIGNMENT:
+ case OUTPUT_NOT_INITIALIZED:
+ case READING:
+ setOperatingMode(
+ mode == StopMode.DRAIN
+ ? OperatingMode.SOURCE_DRAINED
+ : OperatingMode.SOURCE_STOPPED);
+ availabilityHelper.forceStop();
+ if (this.operatingMode == OperatingMode.SOURCE_STOPPED) {
+ stopInternalServices();
+ finished.complete(null);
+ return finished;
+ }
+ break;
+ }
+ return finished;
+ }
+
+ @Override
+ public void close() throws Exception {
+ getContainingTask()
+ .getEnvironment()
+ .getMetricGroup()
+ .getIOMetricGroup()
+ .unregisterBackPressureListener(mainInputActivityClock);
+
+ if (sourceReader != null) {
+ sourceReader.close();
+ }
+ super.close();
+ }
+
+ @Override
+ public DataInputStatus emitNext(DataOutput output) throws Exception {
+ if (waitingForCheckpoint && operatingMode == OperatingMode.SOURCE_DRAINED) {
+ return DataInputStatus.END_OF_DATA;
+ }
+ // guarding an assumptions we currently make due to the fact that certain classes
+ // assume a constant output, this assumption does not need to stand if we emitted all
+ // records. In that case the output will change to FinishedDataOutput
+ assert lastInvokedOutput == output
+ || lastInvokedOutput == null
+ || this.operatingMode == OperatingMode.DATA_FINISHED;
+
+ // short circuit the hot path. Without this short circuit (READING handled in the
+ // switch/case) InputBenchmark.mapSink was showing a performance regression.
+ if (operatingMode != OperatingMode.READING) {
+ return emitNextNotReading(output);
+ }
+
+ InputStatus status;
+ do {
+ status = sourceReader.pollNext(currentMainOutput);
+ } while (status == InputStatus.MORE_AVAILABLE
+ && canEmitBatchOfRecords.check()
+ && !shouldWaitForAlignment());
+ return convertToInternalStatus(status);
+ }
+
+ private DataInputStatus emitNextNotReading(DataOutput output) throws Exception {
+ switch (operatingMode) {
+ case OUTPUT_NOT_INITIALIZED:
+ if (waitingForCheckpoint) {
+ return DataInputStatus.NOTHING_AVAILABLE;
+ }
+ if (watermarkAlignmentParams.isEnabled()) {
+ // Only wrap the output when watermark alignment is enabled, as otherwise this
+ // introduces a small performance regression (probably because of an extra
+ // virtual call)
+ processingTimeService.scheduleWithFixedDelay(
+ time -> sampleAndEmitLatestWatermark(),
+ watermarkAlignmentParams.getUpdateInterval(),
+ watermarkAlignmentParams.getUpdateInterval());
+ }
+ initializeMainOutput(output);
+ return convertToInternalStatus(sourceReader.pollNext(currentMainOutput));
+ case SOURCE_STOPPED:
+ setOperatingMode(OperatingMode.DATA_FINISHED);
+ sourceMetricGroup.idlingStarted();
+ return DataInputStatus.STOPPED;
+ case SOURCE_DRAINED:
+ setOperatingMode(OperatingMode.DATA_FINISHED);
+ sourceMetricGroup.idlingStarted();
+ return DataInputStatus.END_OF_DATA;
+ case DATA_FINISHED:
+ if (watermarkAlignmentParams.isEnabled()) {
+ if (currentMainOutput == null) {
+ // if the source operator was stopped while waiting for the first checkpoint
+ // then the output needs to be initialized so final watermark can be emitted
+ initializeMainOutput(output);
+ }
+ this.sampledLatestWatermark.addLatest(Watermark.MAX_WATERMARK.getTimestamp());
+ sampleAndEmitLatestWatermark();
+ }
+ sourceMetricGroup.idlingStarted();
+ return DataInputStatus.END_OF_INPUT;
+ case WAITING_FOR_ALIGNMENT:
+ checkState(!waitingForAlignmentFuture.isDone());
+ checkState(shouldWaitForAlignment());
+ return convertToInternalStatus(InputStatus.NOTHING_AVAILABLE);
+ case READING:
+ default:
+ throw new IllegalStateException("Unknown operating mode: " + operatingMode);
+ }
+ }
+
+ private void initializeMainOutput(DataOutput output) {
+ currentMainOutput = eventTimeLogic.createMainOutput(output, this);
+ initializeLatencyMarkerEmitter(output);
+ lastInvokedOutput = output;
+ // Create per-split output for pending splits added before main output is initialized
+ createOutputForSplits(splitsToInitializeOutput);
+ setOperatingMode(OperatingMode.READING);
+ }
+
+ private void initializeLatencyMarkerEmitter(DataOutput output) {
+ long latencyTrackingInterval =
+ getExecutionConfig().isLatencyTrackingConfigured()
+ ? getExecutionConfig().getLatencyTrackingInterval()
+ : getContainingTask()
+ .getEnvironment()
+ .getTaskManagerInfo()
+ .getConfiguration()
+ .get(MetricOptions.LATENCY_INTERVAL)
+ .toMillis();
+ if (latencyTrackingInterval > 0) {
+ latencyMarkerEmitter =
+ new LatencyMarkerEmitter<>(
+ getProcessingTimeService(),
+ output::emitLatencyMarker,
+ latencyTrackingInterval,
+ getOperatorID(),
+ getRuntimeContext().getTaskInfo().getIndexOfThisSubtask());
+ }
+ }
+
+ private DataInputStatus convertToInternalStatus(InputStatus inputStatus) {
+ switch (inputStatus) {
+ case MORE_AVAILABLE:
+ return DataInputStatus.MORE_AVAILABLE;
+ case NOTHING_AVAILABLE:
+ sourceMetricGroup.idlingStarted();
+ return DataInputStatus.NOTHING_AVAILABLE;
+ case END_OF_INPUT:
+ setOperatingMode(OperatingMode.DATA_FINISHED);
+ sourceMetricGroup.idlingStarted();
+ return DataInputStatus.END_OF_DATA;
+ default:
+ throw new IllegalArgumentException("Unknown input status: " + inputStatus);
+ }
+ }
+
+ private void sampleAndEmitLatestWatermark() {
+ sampleLatestWatermark();
+ emitLatestWatermark();
+ }
+
+ private void emitLatestWatermark() {
+ checkState(currentMainOutput != null);
+ long latestWatermark = sampledLatestWatermark.getLatest();
+ if (latestWatermark == Watermark.UNINITIALIZED.getTimestamp()) {
+ return;
+ }
+ operatorEventGateway.sendEventToCoordinator(
+ new ReportedWatermarkEvent(
+ idle ? Watermark.MAX_WATERMARK.getTimestamp() : latestWatermark));
+ }
+
+ private void sampleLatestWatermark() {
+ sampledSplitWatermarks.values().forEach(WatermarkSampler::sample);
+ sampledLatestWatermark.sample();
+
+ // as we updated sampled latest watermarks, we should check watermark alignment status
+ checkWatermarkAlignment();
+ checkSplitWatermarkAlignment();
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ if (waitingForCheckpoint) {
+ waitingForCheckpoint = false;
+ checkpointsStartedFuture.complete(null);
+ LOG.info("Source un-paused (checkpoint barrier received)");
+ }
+ long checkpointId = context.getCheckpointId();
+ LOG.debug("Taking a snapshot for checkpoint {}", checkpointId);
+ readerState.update(sourceReader.snapshotState(checkpointId));
+ }
+
+ @Override
+ public CompletableFuture getAvailableFuture() {
+ switch (operatingMode) {
+ case WAITING_FOR_ALIGNMENT:
+ return availabilityHelper.update(waitingForAlignmentFuture);
+ case OUTPUT_NOT_INITIALIZED:
+ return availabilityHelper.update(
+ waitingForCheckpoint
+ ? checkpointsStartedFuture
+ : sourceReader.isAvailable());
+ case READING:
+ return availabilityHelper.update(sourceReader.isAvailable());
+ case SOURCE_STOPPED:
+ case SOURCE_DRAINED:
+ case DATA_FINISHED:
+ return AvailabilityProvider.AVAILABLE;
+ default:
+ throw new IllegalStateException("Unknown operating mode: " + operatingMode);
+ }
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ super.initializeState(context);
+ final ListState rawState =
+ context.getOperatorStateStore().getListState(SPLITS_STATE_DESC);
+ readerState = new SimpleVersionedListState<>(rawState, splitSerializer);
+ if (waitingForCheckpoint && !context.isRestored()) {
+ LOG.debug("Not a recovery, won't wait for the checkpoint to emit records");
+ waitingForCheckpoint = false;
+ checkpointsStartedFuture.complete(null);
+ }
+ }
+
+ @Override
+ public void notifyCheckpointComplete(long checkpointId) throws Exception {
+ super.notifyCheckpointComplete(checkpointId);
+ sourceReader.notifyCheckpointComplete(checkpointId);
+ }
+
+ @Override
+ public void notifyCheckpointAborted(long checkpointId) throws Exception {
+ super.notifyCheckpointAborted(checkpointId);
+ sourceReader.notifyCheckpointAborted(checkpointId);
+ }
+
+ @SuppressWarnings("unchecked")
+ public void handleOperatorEvent(OperatorEvent event) {
+ if (event instanceof WatermarkAlignmentEvent) {
+ updateMaxDesiredWatermark((WatermarkAlignmentEvent) event);
+ checkWatermarkAlignment();
+ checkSplitWatermarkAlignment();
+ } else if (event instanceof AddSplitEvent) {
+ handleAddSplitsEvent(((AddSplitEvent) event));
+ } else if (event instanceof SourceEventWrapper) {
+ sourceReader.handleSourceEvents(((SourceEventWrapper) event).getSourceEvent());
+ } else if (event instanceof NoMoreSplitsEvent) {
+ sourceReader.notifyNoMoreSplits();
+ } else if (event instanceof IsProcessingBacklogEvent) {
+ if (eventTimeLogic != null) {
+ eventTimeLogic.emitImmediateWatermark(System.currentTimeMillis());
+ }
+ output.emitRecordAttributes(
+ new RecordAttributesBuilder(Collections.emptyList())
+ .setBacklog(((IsProcessingBacklogEvent) event).isProcessingBacklog())
+ .build());
+ } else {
+ throw new IllegalStateException("Received unexpected operator event " + event);
+ }
+ }
+
+ private void handleAddSplitsEvent(AddSplitEvent event) {
+ try {
+ List newSplits = event.splits(splitSerializer);
+ if (operatingMode == OperatingMode.OUTPUT_NOT_INITIALIZED) {
+ // For splits arrived before the main output is initialized, store them into the
+ // pending list. Outputs of these splits will be created once the main output is
+ // ready.
+ splitsToInitializeOutput.addAll(newSplits);
+ } else {
+ // Create output directly for new splits if the main output is already initialized.
+ createOutputForSplits(newSplits);
+ }
+ sourceReader.addSplits(newSplits);
+ createMetricGroupForSplits(newSplits);
+ } catch (IOException e) {
+ throw new FlinkRuntimeException("Failed to deserialize the splits.", e);
+ }
+ }
+
+ private void createOutputForSplits(List newSplits) {
+ for (SplitT split : newSplits) {
+ currentMainOutput.createOutputForSplit(split.splitId());
+ }
+ }
+
+ private void createMetricGroupForSplits(List newSplits) {
+ for (SplitT split : newSplits) {
+ getOrCreateSplitMetricGroup(split.splitId());
+ }
+ }
+
+ private void updateMaxDesiredWatermark(WatermarkAlignmentEvent event) {
+ currentMaxDesiredWatermark = event.getMaxWatermark();
+ sourceMetricGroup.updateMaxDesiredWatermark(currentMaxDesiredWatermark);
+ }
+
+ @Override
+ public void updateIdle(boolean isIdle) {
+ this.idle = isIdle;
+ }
+
+ @Override
+ public void updateCurrentEffectiveWatermark(long watermark) {
+ sampledLatestWatermark.addLatest(watermark);
+ checkWatermarkAlignment();
+ }
+
+ @Override
+ public void updateCurrentSplitWatermark(String splitId, long watermark) {
+ WatermarkSampler splitWatermarkSampler = checkNotNull(sampledSplitWatermarks.get(splitId));
+ splitWatermarkSampler.addLatest(watermark);
+ if (!currentlyIdleSplits.contains(splitId)) {
+ maybePauseSplit(splitId);
+ }
+ }
+
+ private void maybePauseSplit(String splitId) {
+ WatermarkSampler splitWatermarkSampler = checkNotNull(sampledSplitWatermarks.get(splitId));
+ long oldestSampledWatermark = splitWatermarkSampler.getOldestSample();
+ // oldestSampledWatermark can be only updated after adding new latest if sampling capacity
+ // is 0, but we still need to handle that
+ if (oldestSampledWatermark > currentMaxDesiredWatermark
+ && !currentlyPausedSplits.contains(splitId)) {
+ pauseOrResumeSplits(Collections.singletonList(splitId), Collections.emptyList());
+ currentlyPausedSplits.add(splitId);
+ }
+ }
+
+ @Override
+ public void updateCurrentSplitIdle(String splitId, boolean idle) {
+ final InternalSourceSplitMetricGroup splitMetricGroup =
+ this.getOrCreateSplitMetricGroup(splitId);
+ if (idle == currentlyIdleSplits.contains(splitId)) {
+ return;
+ }
+ if (idle) {
+ LOG.info("[{}] Marking split idle", splitId);
+ currentlyIdleSplits.add(splitId);
+ splitMetricGroup.markIdle();
+ } else {
+ LOG.info("[{}] Marking split not idle", splitId);
+ currentlyIdleSplits.remove(splitId);
+ splitMetricGroup.markNotIdle();
+ // Since we skipped alignment check
+ // for this split while it was idle:
+ maybePauseSplit(splitId);
+ }
+ }
+
+ @Override
+ public void splitFinished(String splitId) {
+ getOrCreateSplitMetricGroup(splitId).onSplitFinished();
+ this.splitMetricGroups.remove(splitId);
+ sampledSplitWatermarks.remove(splitId);
+ currentlyIdleSplits.remove(splitId);
+ }
+
+ /**
+ * Finds the splits that are beyond the current max watermark and pauses them. At the same time,
+ * splits that have been paused and where the global watermark caught up are resumed.
+ *
+ * Note: This takes effect only if there are multiple splits, otherwise it does nothing.
+ */
+ private void checkSplitWatermarkAlignment() {
+ Collection splitsToPause = new ArrayList<>();
+ Collection splitsToResume = new ArrayList<>();
+ sampledSplitWatermarks.forEach(
+ (splitId, splitWatermarks) -> {
+ if (splitWatermarks.getOldestSample() > currentMaxDesiredWatermark) {
+ // Skipping pause for idle splits so we won't clear their idleness
+ if (currentlyIdleSplits.contains(splitId)) {
+ LOG.info("[{}] Skipping pause for idle split", splitId);
+ return;
+ }
+ splitsToPause.add(splitId);
+ } else if (currentlyPausedSplits.contains(splitId)) {
+ // Resuming possibly-idle splits without clearing their idleness state
+ // (the next record to arrive will do it naturally)
+ splitsToResume.add(splitId);
+ }
+ });
+ splitsToPause.removeAll(currentlyPausedSplits);
+ if (!splitsToPause.isEmpty() || !splitsToResume.isEmpty()) {
+ pauseOrResumeSplits(splitsToPause, splitsToResume);
+ currentlyPausedSplits.addAll(splitsToPause);
+ splitsToResume.forEach(currentlyPausedSplits::remove);
+ }
+ }
+
+ private void pauseOrResumeSplits(
+ Collection splitsToPause, Collection splitsToResume) {
+ try {
+ LOG.info(
+ "pauseOrResumeSplits [splitsToPause={}][splitsToResume={}][idleSplits={}]"
+ + "[currentMaxDesiredWatermark={}][latestWatermark={}][oldestWatermark={}]",
+ splitsToPause,
+ splitsToResume,
+ currentlyIdleSplits,
+ currentMaxDesiredWatermark,
+ sampledLatestWatermark.getLatest(),
+ sampledLatestWatermark.getOldestSample());
+ sourceReader.pauseOrResumeSplits(splitsToPause, splitsToResume);
+ eventTimeLogic.pauseOrResumeSplits(splitsToPause, splitsToResume);
+ reportPausedOrResumed(splitsToPause, splitsToResume);
+ } catch (UnsupportedOperationException e) {
+ if (!allowUnalignedSourceSplits) {
+ throw e;
+ }
+ }
+ }
+
+ private void reportPausedOrResumed(
+ Collection splitsToPause, Collection splitsToResume) {
+ for (String splitId : splitsToResume) {
+ getOrCreateSplitMetricGroup(splitId).markNotPaused();
+ }
+ for (String splitId : splitsToPause) {
+ getOrCreateSplitMetricGroup(splitId).markPaused();
+ }
+ }
+
+ private void checkWatermarkAlignment() {
+ if (operatingMode == OperatingMode.READING) {
+ checkState(waitingForAlignmentFuture.isDone());
+ if (shouldWaitForAlignment()) {
+ setOperatingMode(OperatingMode.WAITING_FOR_ALIGNMENT);
+ waitingForAlignmentFuture = new CompletableFuture<>();
+ mainInputActivityClock.pause();
+ }
+ } else if (operatingMode == OperatingMode.WAITING_FOR_ALIGNMENT) {
+ checkState(!waitingForAlignmentFuture.isDone());
+ if (!shouldWaitForAlignment()) {
+ setOperatingMode(OperatingMode.READING);
+ waitingForAlignmentFuture.complete(null);
+ mainInputActivityClock.unPause();
+ }
+ }
+ }
+
+ private boolean shouldWaitForAlignment() {
+ return currentMaxDesiredWatermark < sampledLatestWatermark.getOldestSample();
+ }
+
+ private void registerReader(List splits) throws Exception {
+ operatorEventGateway.sendEventToCoordinator(
+ ReaderRegistrationEvent.createReaderRegistrationEvent(
+ getRuntimeContext().getTaskInfo().getIndexOfThisSubtask(),
+ localHostname,
+ splits,
+ splitSerializer));
+ }
+
+ // --------------- methods for unit tests ------------
+
+ @VisibleForTesting
+ public SourceReader getSourceReader() {
+ return sourceReader;
+ }
+
+ @VisibleForTesting
+ ListState getReaderState() {
+ return readerState;
+ }
+
+ private static class SourceOperatorAvailabilityHelper {
+ private final CompletableFuture forcedStopFuture = new CompletableFuture<>();
+ private final MultipleFuturesAvailabilityHelper availabilityHelper;
+
+ private SourceOperatorAvailabilityHelper() {
+ availabilityHelper = new MultipleFuturesAvailabilityHelper(2);
+ availabilityHelper.anyOf(0, forcedStopFuture);
+ }
+
+ public CompletableFuture update(CompletableFuture sourceReaderFuture) {
+ if (sourceReaderFuture == AvailabilityProvider.AVAILABLE
+ || sourceReaderFuture.isDone()) {
+ return AvailabilityProvider.AVAILABLE;
+ }
+ availabilityHelper.resetToUnAvailable();
+ availabilityHelper.anyOf(0, forcedStopFuture);
+ availabilityHelper.anyOf(1, sourceReaderFuture);
+ return availabilityHelper.getAvailableFuture();
+ }
+
+ public void forceStop() {
+ forcedStopFuture.complete(null);
+ }
+ }
+
+ private void setOperatingMode(OperatingMode newMode) {
+ final long now = System.currentTimeMillis();
+ LOG.info(
+ "Switch mode from {} to {} after {} ms, currentMaxDesiredWatermark={}, latestWatermark={}, oldestWatermark={}",
+ operatingMode,
+ newMode,
+ now - operatingModeChangeTs,
+ currentMaxDesiredWatermark,
+ sampledLatestWatermark.getLatest(),
+ sampledLatestWatermark.getOldestSample());
+ operatingMode = newMode;
+ operatingModeChangeTs = now;
+ }
+}
diff --git a/flink-sql-runner/src/main/java/org/apache/flink/streaming/api/operators/package-info.java b/flink-sql-runner/src/main/java/org/apache/flink/streaming/api/operators/package-info.java
new file mode 100644
index 00000000..11faca6a
--- /dev/null
+++ b/flink-sql-runner/src/main/java/org/apache/flink/streaming/api/operators/package-info.java
@@ -0,0 +1,24 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * Backporting FLINK-40093 bugfix until Flink 2.3.1 released.
+ *
+ * @see FLINK-40093
+ */
+package org.apache.flink.streaming.api.operators;