From 475625a3960a0d8f6a58ea8d69764103764e07bb Mon Sep 17 00:00:00 2001 From: Sylwester Lachiewicz Date: Thu, 10 Sep 2026 14:45:52 +0200 Subject: [PATCH] [FLINK-40234][runtime] Exclude synchronous downstream processing from source idleness accounting Everything a source emits is processed by the chained operators on the task thread before the reader is polled again. That time counted towards per-split idleness, so a slow chained operator or a window firing that outlasted the idle timeout marked splits with fetched-but-unpolled records idle; the combined watermark then advanced without them and their records were dropped as late. The source's DataOutput is now wrapped so the input activity clock is paused while downstream processing runs, mirroring the backpressure pause of FLIP-471. Per-split clocks are layered on the main clock so one pause covers all splits. The wrapper is only installed when a watermark strategy asked for the input activity clock (withIdleness); other jobs keep the unwrapped output. Generated-by: Claude Fable 5.1 --- .../api/operators/SourceOperator.java | 4 +- .../ActivityClockPausingDataOutput.java | 122 ++++++++++ .../ProgressiveTimestampsAndWatermarks.java | 61 +++-- .../source/TimestampsAndWatermarks.java | 31 ++- .../TimestampsAndWatermarksContext.java | 16 ++ .../operators/util/PausableRelativeClock.java | 11 +- ...ceOperatorSplitWatermarkAlignmentTest.java | 209 ++++++++++++++++++ .../ActivityClockPausingDataOutputTest.java | 157 +++++++++++++ .../util/PausableRelativeClockTest.java | 21 ++ .../tasks/TestProcessingTimeService.java | 9 + 10 files changed, 591 insertions(+), 50 deletions(-) create mode 100644 flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/source/ActivityClockPausingDataOutput.java create mode 100644 flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/source/ActivityClockPausingDataOutputTest.java diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java index c487aba5255e94..bf32a2242d06eb 100644 --- a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java +++ b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java @@ -431,9 +431,7 @@ public void open() throws Exception { sourceMetricGroup, getProcessingTimeService(), getExecutionConfig().getAutoWatermarkInterval(), - mainInputActivityClock, - getProcessingTimeService().getClock(), - taskIOMetricGroup); + mainInputActivityClock); } else { eventTimeLogic = TimestampsAndWatermarks.createNoOpEventTimeLogic( diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/source/ActivityClockPausingDataOutput.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/source/ActivityClockPausingDataOutput.java new file mode 100644 index 00000000000000..c09c28df19377a --- /dev/null +++ b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/source/ActivityClockPausingDataOutput.java @@ -0,0 +1,122 @@ +/* + * 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.source; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.runtime.event.WatermarkEvent; +import org.apache.flink.streaming.api.operators.util.PausableRelativeClock; +import org.apache.flink.streaming.api.watermark.Watermark; +import org.apache.flink.streaming.runtime.io.PushingAsyncDataInput; +import org.apache.flink.streaming.runtime.streamrecord.LatencyMarker; +import org.apache.flink.streaming.runtime.streamrecord.RecordAttributes; +import org.apache.flink.streaming.runtime.streamrecord.StreamRecord; +import org.apache.flink.streaming.runtime.watermarkstatus.WatermarkStatus; + +import static org.apache.flink.util.Preconditions.checkNotNull; + +/** + * A {@link PushingAsyncDataInput.DataOutput} decorator that pauses the source's input activity + * clock while the downstream operators process what was emitted. + * + *

Everything emitted by a source is processed synchronously by the chained operators on the task + * thread before control returns to the source reader. During that time no split can be polled, so + * the elapsed time says nothing about whether a split has records. Without this decorator a slow + * chained operator, or a window firing triggered by a watermark, can exceed the idleness timeout + * and get a split with pending records marked idle, after which those records arrive behind an + * already advanced watermark and are dropped as late. + * + *

This is the same idea as pausing the clock during backpressure (FLIP-471): the idleness + * timeout only counts time during which the source was actually able to make progress on its input. + * + * @param The type of the emitted records. + */ +@Internal +public final class ActivityClockPausingDataOutput + implements PushingAsyncDataInput.DataOutput { + + private final PushingAsyncDataInput.DataOutput delegate; + private final PausableRelativeClock inputActivityClock; + + public ActivityClockPausingDataOutput( + PushingAsyncDataInput.DataOutput delegate, + PausableRelativeClock inputActivityClock) { + this.delegate = checkNotNull(delegate); + this.inputActivityClock = checkNotNull(inputActivityClock); + } + + @Override + public void emitRecord(StreamRecord streamRecord) throws Exception { + inputActivityClock.pause(); + try { + delegate.emitRecord(streamRecord); + } finally { + inputActivityClock.unPause(); + } + } + + @Override + public void emitWatermark(Watermark watermark) throws Exception { + inputActivityClock.pause(); + try { + delegate.emitWatermark(watermark); + } finally { + inputActivityClock.unPause(); + } + } + + @Override + public void emitWatermarkStatus(WatermarkStatus watermarkStatus) throws Exception { + inputActivityClock.pause(); + try { + delegate.emitWatermarkStatus(watermarkStatus); + } finally { + inputActivityClock.unPause(); + } + } + + @Override + public void emitLatencyMarker(LatencyMarker latencyMarker) throws Exception { + inputActivityClock.pause(); + try { + delegate.emitLatencyMarker(latencyMarker); + } finally { + inputActivityClock.unPause(); + } + } + + @Override + public void emitRecordAttributes(RecordAttributes recordAttributes) throws Exception { + inputActivityClock.pause(); + try { + delegate.emitRecordAttributes(recordAttributes); + } finally { + inputActivityClock.unPause(); + } + } + + @Override + public void emitWatermark(WatermarkEvent watermark) throws Exception { + inputActivityClock.pause(); + try { + delegate.emitWatermark(watermark); + } finally { + inputActivityClock.unPause(); + } + } +} diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/source/ProgressiveTimestampsAndWatermarks.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/source/ProgressiveTimestampsAndWatermarks.java index 99ffd6b05db543..169b8f6e31b8fd 100644 --- a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/source/ProgressiveTimestampsAndWatermarks.java +++ b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/source/ProgressiveTimestampsAndWatermarks.java @@ -27,12 +27,9 @@ import org.apache.flink.api.common.eventtime.WatermarkOutputMultiplexer; import org.apache.flink.api.connector.source.ReaderOutput; import org.apache.flink.api.connector.source.SourceOutput; -import org.apache.flink.runtime.metrics.groups.TaskIOMetricGroup; import org.apache.flink.streaming.api.operators.util.PausableRelativeClock; import org.apache.flink.streaming.runtime.io.PushingAsyncDataInput; import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService; -import org.apache.flink.util.clock.Clock; -import org.apache.flink.util.clock.RelativeClock; import javax.annotation.Nullable; @@ -66,11 +63,7 @@ public class ProgressiveTimestampsAndWatermarks implements TimestampsAndWater private final long periodicWatermarkInterval; - private final RelativeClock mainInputActivityClock; - - private final Clock clock; - - private final TaskIOMetricGroup taskIOMetricGroup; + private final PausableRelativeClock mainInputActivityClock; @Nullable private SplitLocalOutputs currentPerSplitOutputs; @@ -84,17 +77,13 @@ public ProgressiveTimestampsAndWatermarks( TimestampsAndWatermarksContextProvider watermarksContextProvider, ProcessingTimeService timeService, Duration periodicWatermarkInterval, - RelativeClock mainInputActivityClock, - Clock clock, - TaskIOMetricGroup taskIOMetricGroup) { + PausableRelativeClock mainInputActivityClock) { this.timestampAssigner = timestampAssigner; this.watermarksFactory = watermarksFactory; this.watermarksContextProvider = watermarksContextProvider; this.timeService = timeService; this.mainInputActivityClock = mainInputActivityClock; - this.clock = clock; - this.taskIOMetricGroup = taskIOMetricGroup; long periodicWatermarkIntervalMillis; try { @@ -120,28 +109,36 @@ public ReaderOutput createMainOutput( currentMainOutput == null && currentPerSplitOutputs == null, "already created a main output"); - final WatermarkOutput watermarkOutput = - new WatermarkToDataOutput(output, watermarkUpdateListener); - IdlenessManager idlenessManager = new IdlenessManager(watermarkOutput); - final WatermarkGenerator watermarkGenerator = watermarksFactory.createWatermarkGenerator( watermarksContextProvider.create(mainInputActivityClock)); + // Downstream operators process every emitted element synchronously on the task thread, + // during which no split can be polled. When something measures input activity time (i.e. + // idleness detection is configured), hide that time from the activity clocks; otherwise + // keep the plain output to leave the hot path untouched. + final PushingAsyncDataInput.DataOutput recordOutput = + watermarksContextProvider.isInputActivityClockRequested() + ? new ActivityClockPausingDataOutput<>(output, mainInputActivityClock) + : output; + + final WatermarkOutput watermarkOutput = + new WatermarkToDataOutput(recordOutput, watermarkUpdateListener); + IdlenessManager idlenessManager = new IdlenessManager(watermarkOutput); + currentPerSplitOutputs = new SplitLocalOutputs<>( - output, + recordOutput, idlenessManager.getSplitLocalOutput(), watermarkUpdateListener, timestampAssigner, watermarksFactory, watermarksContextProvider, - clock, - taskIOMetricGroup); + mainInputActivityClock); currentMainOutput = new StreamingReaderOutput<>( - output, + recordOutput, idlenessManager.getMainOutput(), timestampAssigner, watermarkGenerator, @@ -237,8 +234,7 @@ private static final class SplitLocalOutputs { private final WatermarkGeneratorSupplier watermarksFactory; private final TimestampsAndWatermarksContextProvider watermarksContextProvider; private final WatermarkUpdateListener watermarkUpdateListener; - private final Clock clock; - private final TaskIOMetricGroup taskIOMetricGroup; + private final PausableRelativeClock mainInputActivityClock; private SplitLocalOutputs( PushingAsyncDataInput.DataOutput recordOutput, @@ -247,16 +243,14 @@ private SplitLocalOutputs( TimestampAssigner timestampAssigner, WatermarkGeneratorSupplier watermarksFactory, TimestampsAndWatermarksContextProvider watermarksContextProvider, - Clock clock, - TaskIOMetricGroup taskIOMetricGroup) { + PausableRelativeClock mainInputActivityClock) { this.recordOutput = recordOutput; this.timestampAssigner = timestampAssigner; this.watermarksFactory = watermarksFactory; this.watermarksContextProvider = watermarksContextProvider; this.watermarkUpdateListener = watermarkUpdateListener; - this.clock = clock; - this.taskIOMetricGroup = taskIOMetricGroup; + this.mainInputActivityClock = mainInputActivityClock; this.watermarkMultiplexer = new WatermarkOutputMultiplexer(watermarkOutput); this.localOutputs = @@ -303,11 +297,12 @@ public void onIdleUpdate(boolean idle) { } private PausableRelativeClock createInputActivityClock(String splitId) { - // Dedicated inputActivityClock for a particular split. It will be paused both in case - // of back pressure and when split is paused due to watermark alignment. - PausableRelativeClock inputActivityClock = new PausableRelativeClock(clock); + // Dedicated inputActivityClock for a particular split, layered on the main input + // activity clock: it is paused whenever the main clock is (backpressure, downstream + // processing) and additionally when this split is paused due to watermark alignment. + PausableRelativeClock inputActivityClock = + new PausableRelativeClock(mainInputActivityClock); inputActivityClocks.put(splitId, inputActivityClock); - taskIOMetricGroup.registerBackPressureListener(inputActivityClock); return inputActivityClock; } @@ -315,9 +310,7 @@ void releaseOutputForSplit(String splitId) { watermarkUpdateListener.splitFinished(splitId); localOutputs.remove(splitId); watermarkMultiplexer.unregisterOutput(splitId); - PausableRelativeClock inputActivityClock = - requireNonNull(inputActivityClocks.remove(splitId)); - taskIOMetricGroup.unregisterBackPressureListener(inputActivityClock); + requireNonNull(inputActivityClocks.remove(splitId)); } void emitPeriodicWatermark() { diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/source/TimestampsAndWatermarks.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/source/TimestampsAndWatermarks.java index 9da5a9274d18cc..6d99afce224abe 100644 --- a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/source/TimestampsAndWatermarks.java +++ b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/source/TimestampsAndWatermarks.java @@ -24,10 +24,9 @@ import org.apache.flink.api.common.eventtime.WatermarkStrategy; import org.apache.flink.api.connector.source.ReaderOutput; import org.apache.flink.metrics.MetricGroup; -import org.apache.flink.runtime.metrics.groups.TaskIOMetricGroup; +import org.apache.flink.streaming.api.operators.util.PausableRelativeClock; import org.apache.flink.streaming.runtime.io.PushingAsyncDataInput; import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService; -import org.apache.flink.util.clock.Clock; import org.apache.flink.util.clock.RelativeClock; import java.time.Duration; @@ -104,9 +103,7 @@ static TimestampsAndWatermarks createProgressiveEventTimeLogic( MetricGroup metrics, ProcessingTimeService timeService, long periodicWatermarkIntervalMillis, - RelativeClock mainInputActivityClock, - Clock clock, - TaskIOMetricGroup taskIOMetricGroup) { + PausableRelativeClock mainInputActivityClock) { TimestampsAndWatermarksContextProvider contextProvider = new TimestampsAndWatermarksContextProvider(metrics); @@ -120,9 +117,7 @@ static TimestampsAndWatermarks createProgressiveEventTimeLogic( contextProvider, timeService, Duration.ofMillis(periodicWatermarkIntervalMillis), - mainInputActivityClock, - clock, - taskIOMetricGroup); + mainInputActivityClock); } static TimestampsAndWatermarks createNoOpEventTimeLogic( @@ -142,12 +137,30 @@ static TimestampsAndWatermarks createNoOpEventTimeLogic( class TimestampsAndWatermarksContextProvider { private final MetricGroup metrics; + /** Whether any created context handed out the input activity clock. */ + private boolean inputActivityClockRequested; + public TimestampsAndWatermarksContextProvider(MetricGroup metrics) { this.metrics = metrics; } public TimestampsAndWatermarksContext create(RelativeClock inputActivityClock) { - return new TimestampsAndWatermarksContext(metrics, inputActivityClock); + return new TimestampsAndWatermarksContext( + metrics, inputActivityClock, () -> inputActivityClockRequested = true); + } + + /** + * Returns true if a timestamp assigner or watermark generator created through this provider + * asked for the input activity clock, which is the case with {@link + * org.apache.flink.api.common.eventtime.WatermarkStrategy#withIdleness(java.time.Duration)}. + * Only then is it worth paying for keeping that clock accurate. + * + *

The decision to hide downstream processing time from the clock is taken once, when the + * main output is created. A supplier that asks for the clock lazily, after its first {@code + * createWatermarkGenerator} call returned, therefore keeps the previous behaviour. + */ + public boolean isInputActivityClockRequested() { + return inputActivityClockRequested; } } } diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/source/TimestampsAndWatermarksContext.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/source/TimestampsAndWatermarksContext.java index 6f543e0392c901..c10fa29ab3f1d2 100644 --- a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/source/TimestampsAndWatermarksContext.java +++ b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/source/TimestampsAndWatermarksContext.java @@ -35,12 +35,27 @@ public final class TimestampsAndWatermarksContext implements TimestampAssignerSupplier.Context, WatermarkGeneratorSupplier.Context { private final MetricGroup metricGroup; + private final RelativeClock inputActivityClock; + private final Runnable onInputActivityClockRequested; + public TimestampsAndWatermarksContext( MetricGroup metricGroup, RelativeClock inputActivityClock) { + this(metricGroup, inputActivityClock, () -> {}); + } + + /** + * @param onInputActivityClockRequested invoked every time {@link #getInputActivityClock()} is + * called, so that the owner can tell whether anything measures input activity time. + */ + public TimestampsAndWatermarksContext( + MetricGroup metricGroup, + RelativeClock inputActivityClock, + Runnable onInputActivityClockRequested) { this.metricGroup = checkNotNull(metricGroup); this.inputActivityClock = inputActivityClock; + this.onInputActivityClockRequested = checkNotNull(onInputActivityClockRequested); } @Override @@ -50,6 +65,7 @@ public MetricGroup getMetricGroup() { @Override public RelativeClock getInputActivityClock() { + onInputActivityClockRequested.run(); return inputActivityClock; } } diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/util/PausableRelativeClock.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/util/PausableRelativeClock.java index b49e52cf6cab04..e162615c7890fd 100644 --- a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/util/PausableRelativeClock.java +++ b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/util/PausableRelativeClock.java @@ -20,7 +20,6 @@ import org.apache.flink.annotation.Internal; import org.apache.flink.runtime.metrics.TimerGauge; -import org.apache.flink.util.clock.Clock; import org.apache.flink.util.clock.RelativeClock; import javax.annotation.concurrent.ThreadSafe; @@ -28,14 +27,14 @@ import static org.apache.flink.util.Preconditions.checkState; /** - * A {@link RelativeClock} whose time progress with respect to the wall clock can be paused and + * A {@link RelativeClock} whose time progress with respect to its base clock can be paused and * un-paused. It can be paused multiple times. If it is paused N times, it has to be un-paused also * N times to resume progress. */ @Internal @ThreadSafe public class PausableRelativeClock implements RelativeClock, TimerGauge.StartStopListener { - private final Clock baseClock; + private final RelativeClock baseClock; private long accumulativeBlockedNanoTime; private long currentBlockedNanoTimeStart; @@ -43,7 +42,11 @@ public class PausableRelativeClock implements RelativeClock, TimerGauge.StartSto /** How many times this clock has been paused. */ private long pausedCounter; - public PausableRelativeClock(Clock baseClock) { + /** + * @param baseClock the clock this one derives its time from. It can itself be a {@link + * PausableRelativeClock}, in which case pausing the base clock also pauses this one. + */ + public PausableRelativeClock(RelativeClock baseClock) { this.baseClock = baseClock; } diff --git a/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/SourceOperatorSplitWatermarkAlignmentTest.java b/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/SourceOperatorSplitWatermarkAlignmentTest.java index 36fda4eee681d3..83a040d4737236 100644 --- a/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/SourceOperatorSplitWatermarkAlignmentTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/SourceOperatorSplitWatermarkAlignmentTest.java @@ -31,6 +31,7 @@ Licensed to the Apache Software Foundation (ASF) under one import org.apache.flink.configuration.Configuration; import org.apache.flink.metrics.Metric; import org.apache.flink.runtime.clusterframework.types.ResourceID; +import org.apache.flink.runtime.event.WatermarkEvent; import org.apache.flink.runtime.execution.Environment; import org.apache.flink.runtime.metrics.MetricNames; import org.apache.flink.runtime.metrics.NoOpMetricRegistry; @@ -47,6 +48,9 @@ Licensed to the Apache Software Foundation (ASF) under one import org.apache.flink.streaming.api.operators.source.CollectingDataOutput; import org.apache.flink.streaming.api.operators.source.TestingSourceOperator; import org.apache.flink.streaming.runtime.io.DataInputStatus; +import org.apache.flink.streaming.runtime.io.PushingAsyncDataInput; +import org.apache.flink.streaming.runtime.streamrecord.LatencyMarker; +import org.apache.flink.streaming.runtime.streamrecord.RecordAttributes; import org.apache.flink.streaming.runtime.streamrecord.StreamRecord; import org.apache.flink.streaming.runtime.tasks.SourceOperatorStreamTask; import org.apache.flink.streaming.runtime.tasks.StreamMockEnvironment; @@ -629,6 +633,142 @@ void testPausedIdleSplitsCanBeResumedByAlignmentCheck() throws Exception { assertOutput(actualOutput, Arrays.asList(5, 6)); } + /** + * FLINK-40234: a split whose records have been fetched but not yet polled must not be marked + * idle because a slow chained operator kept the task thread busy for longer than the idle + * timeout. Otherwise the watermark advances on the other splits alone and the pending records + * arrive late. + */ + @Test + void testSlowDownstreamRecordProcessingDoesNotMarkUnpolledSplitIdle() throws Exception { + final long idleTimeout = 1000; + final MockSourceReader sourceReader = + new MockSourceReader(WaitingForSplits.DO_NOT_WAIT_FOR_SPLITS, false, true); + final TestProcessingTimeService processingTimeService = new TestProcessingTimeService(); + processingTimeService.setCurrentTime(0); + final SourceOperator operator = + createAndOpenSourceOperatorWithIdleness( + sourceReader, processingTimeService, idleTimeout); + + final MockSourceSplit split0 = new MockSourceSplit(0, 0, 10).addRecord(5).addRecord(6); + final MockSourceSplit split1 = new MockSourceSplit(1, 10, 20).addRecord(2); + operator.handleOperatorEvent( + new AddSplitEvent<>( + Arrays.asList(split0, split1), new MockSourceSplitSerializer())); + final CollectingDataOutput collected = new CollectingDataOutput<>(); + final BusyDataOutput dataOutput = + new BusyDataOutput<>(collected, processingTimeService, false); + + // Rule out watermark alignment pauses for this test. + operator.handleOperatorEvent(new WatermarkAlignmentEvent(Long.MAX_VALUE)); + + operator.emitNext(dataOutput); // split0 emits 5, periodic watermark timer is armed + + // A few periodic probes pass while split1 is quiet, well within the idle timeout. + for (int i = 0; i < 4; i++) { + processingTimeService.advance(100); + } + assertThat(operator.getSplitMetricGroup(split1.splitId()).isIdle()).isFalse(); + + // The chained operator now takes ten idle timeouts to process the next record. No timer + // can fire in the meantime because the task thread is busy. + dataOutput.setBusyMillis(10 * idleTimeout); + operator.emitNext(dataOutput); // split0 emits 6 + dataOutput.setBusyMillis(0); + + // Thread frees up, the overdue periodic probe fires. + processingTimeService.advance(50); + + // split1 has a pending record that nothing asked for yet; the busy time was not idle time. + assertThat(operator.getSplitMetricGroup(split1.splitId()).isIdle()).isFalse(); + + operator.emitNext(dataOutput); // split1 emits 2 + final List events = collected.getEvents(); + final int recordIndex = indexOfRecordWithValue(events, 2); + final int watermarkIndex = indexOfWatermarkAtLeast(events, 5); + assertThat(recordIndex).isNotNegative(); + assertThat(watermarkIndex == -1 || recordIndex < watermarkIndex) + .as("record with timestamp 2 must not arrive behind a watermark of 5 or more") + .isTrue(); + } + + /** + * FLINK-40234, second scenario: the busy work is triggered by a watermark (e.g. a HOP window + * firing) rather than by a record. The quiet split with a pending record must not be marked + * idle for the duration of that work either. + */ + @Test + void testSlowDownstreamWatermarkProcessingDoesNotMarkUnpolledSplitIdle() throws Exception { + final long idleTimeout = 1000; + final MockSourceReader sourceReader = + new MockSourceReader(WaitingForSplits.DO_NOT_WAIT_FOR_SPLITS, false, true); + final TestProcessingTimeService processingTimeService = new TestProcessingTimeService(); + processingTimeService.setCurrentTime(0); + final SourceOperator operator = + createAndOpenSourceOperatorWithIdleness( + sourceReader, processingTimeService, idleTimeout); + + // split0 is ahead in event time and goes quiet; split1 keeps advancing the combined + // watermark, each advance firing a slow downstream computation. + final MockSourceSplit split0 = new MockSourceSplit(0, 0, 10).addRecord(8); + final MockSourceSplit split1 = new MockSourceSplit(1, 10, 20).addRecord(5).addRecord(6); + operator.handleOperatorEvent( + new AddSplitEvent<>( + Arrays.asList(split0, split1), new MockSourceSplitSerializer())); + final CollectingDataOutput collected = new CollectingDataOutput<>(); + final BusyDataOutput dataOutput = + new BusyDataOutput<>(collected, processingTimeService, true); + + operator.handleOperatorEvent(new WatermarkAlignmentEvent(Long.MAX_VALUE)); + + operator.emitNext(dataOutput); // split0 emits 8 + operator.emitNext(dataOutput); // split1 emits 5, combined watermark 5 goes downstream + + for (int i = 0; i < 4; i++) { + processingTimeService.advance(100); + } + assertThat(operator.getSplitMetricGroup(split0.splitId()).isIdle()).isFalse(); + + // split1's next record advances the combined watermark to 6, and the downstream reaction + // to that watermark takes ten idle timeouts. + dataOutput.setBusyMillis(10 * idleTimeout); + operator.emitNext(dataOutput); // split1 emits 6 + dataOutput.setBusyMillis(0); + assertThat(collected.getEvents()) + .contains(new org.apache.flink.streaming.api.watermark.Watermark(6)); + + // Meanwhile more data arrived for split0 in the reader, but it has not been polled yet. + // (Added after the emit above because the mock reader always polls split0 first.) + split0.addRecord(9); + + processingTimeService.advance(50); + + assertThat(operator.getSplitMetricGroup(split0.splitId()).isIdle()).isFalse(); + } + + private static int indexOfWatermarkAtLeast(List events, long minTimestamp) { + for (int i = 0; i < events.size(); i++) { + Object event = events.get(i); + if (event instanceof org.apache.flink.streaming.api.watermark.Watermark + && ((org.apache.flink.streaming.api.watermark.Watermark) event).getTimestamp() + >= minTimestamp) { + return i; + } + } + return -1; + } + + private static int indexOfRecordWithValue(List events, int value) { + for (int i = 0; i < events.size(); i++) { + Object event = events.get(i); + if (event instanceof StreamRecord + && ((StreamRecord) event).getValue().equals(value)) { + return i; + } + } + return -1; + } + private void sampleAllWatermarks(TestProcessingTimeService timeService) throws Exception { sampleWatermarks(timeService, WATERMARK_ALIGNMENT_BUFFER_SIZE.defaultValue()); } @@ -758,6 +898,75 @@ public void onPeriodicEmit(WatermarkOutput output) { } } + /** + * Output that models a slow chained operator: every record (or every watermark, if {@code + * busyOnWatermarks}) burns {@code busyMillis} of wall-clock time on the task thread, during + * which no timer can fire. + */ + private static final class BusyDataOutput implements PushingAsyncDataInput.DataOutput { + + private final CollectingDataOutput delegate; + private final TestProcessingTimeService timeService; + private final boolean busyOnWatermarks; + private long busyMillis; + + BusyDataOutput( + CollectingDataOutput delegate, + TestProcessingTimeService timeService, + boolean busyOnWatermarks) { + this.delegate = delegate; + this.timeService = timeService; + this.busyOnWatermarks = busyOnWatermarks; + } + + void setBusyMillis(long busyMillis) { + this.busyMillis = busyMillis; + } + + private void burn() { + if (busyMillis > 0) { + timeService.advanceClockWithoutFiringTimers(busyMillis); + } + } + + @Override + public void emitRecord(StreamRecord streamRecord) throws Exception { + if (!busyOnWatermarks) { + burn(); + } + delegate.emitRecord(streamRecord); + } + + @Override + public void emitWatermark(org.apache.flink.streaming.api.watermark.Watermark watermark) + throws Exception { + if (busyOnWatermarks) { + burn(); + } + delegate.emitWatermark(watermark); + } + + @Override + public void emitWatermarkStatus(WatermarkStatus watermarkStatus) throws Exception { + delegate.emitWatermarkStatus(watermarkStatus); + } + + @Override + public void emitLatencyMarker(LatencyMarker latencyMarker) throws Exception { + delegate.emitLatencyMarker(latencyMarker); + } + + @Override + public void emitRecordAttributes(RecordAttributes recordAttributes) throws Exception { + delegate.emitRecordAttributes(recordAttributes); + } + + @Override + public void emitWatermark(WatermarkEvent watermark) throws Exception { + delegate.emitWatermark(watermark); + } + } + /** Condition checking if there is no watermark above a certain value among StreamElements. */ public static class WatermarkAbove extends Condition { public WatermarkAbove(int maxEmittedWatermark) { diff --git a/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/source/ActivityClockPausingDataOutputTest.java b/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/source/ActivityClockPausingDataOutputTest.java new file mode 100644 index 00000000000000..13f3bf7437b98f --- /dev/null +++ b/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/source/ActivityClockPausingDataOutputTest.java @@ -0,0 +1,157 @@ +/* + * 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.source; + +import org.apache.flink.runtime.event.WatermarkEvent; +import org.apache.flink.streaming.api.operators.util.PausableRelativeClock; +import org.apache.flink.streaming.api.watermark.Watermark; +import org.apache.flink.streaming.runtime.io.PushingAsyncDataInput; +import org.apache.flink.streaming.runtime.streamrecord.LatencyMarker; +import org.apache.flink.streaming.runtime.streamrecord.RecordAttributes; +import org.apache.flink.streaming.runtime.streamrecord.RecordAttributesBuilder; +import org.apache.flink.streaming.runtime.streamrecord.StreamRecord; +import org.apache.flink.streaming.runtime.watermarkstatus.WatermarkStatus; +import org.apache.flink.util.clock.ManualClock; + +import org.junit.jupiter.api.Test; + +import java.time.Duration; +import java.util.Collections; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Unit tests for {@link ActivityClockPausingDataOutput}. */ +class ActivityClockPausingDataOutputTest { + + private static final long BUSY_MILLIS = 250; + + @Test + void downstreamTimeDoesNotAdvanceActivityClock() throws Exception { + final ManualClock baseClock = new ManualClock(); + final PausableRelativeClock activityClock = new PausableRelativeClock(baseClock); + final CollectingDataOutput collected = new CollectingDataOutput<>(); + final BusyDelegate busyDelegate = new BusyDelegate<>(collected, baseClock); + final ActivityClockPausingDataOutput output = + new ActivityClockPausingDataOutput<>(busyDelegate, activityClock); + + final long start = activityClock.relativeTimeMillis(); + + output.emitRecord(new StreamRecord<>(1, 1L)); + output.emitWatermark(new Watermark(1L)); + output.emitWatermarkStatus(WatermarkStatus.IDLE); + output.emitLatencyMarker(new LatencyMarker(1L, null, 0)); + output.emitRecordAttributes(new RecordAttributesBuilder(Collections.emptyList()).build()); + output.emitWatermark(new WatermarkEvent(null, false)); + + assertThat(activityClock.relativeTimeMillis() - start).isZero(); + assertThat(baseClock.relativeTimeMillis()).isEqualTo(6 * BUSY_MILLIS); + assertThat(collected.getEvents()).hasSize(6); + } + + @Test + void clockResumesBetweenEmits() throws Exception { + final ManualClock baseClock = new ManualClock(); + final PausableRelativeClock activityClock = new PausableRelativeClock(baseClock); + final ActivityClockPausingDataOutput output = + new ActivityClockPausingDataOutput<>( + new BusyDelegate<>(new CollectingDataOutput<>(), baseClock), activityClock); + + final long start = activityClock.relativeTimeMillis(); + output.emitRecord(new StreamRecord<>(1, 1L)); + baseClock.advanceTime(Duration.ofMillis(40)); // reader time between records: counts + output.emitRecord(new StreamRecord<>(2, 2L)); + + assertThat(activityClock.relativeTimeMillis() - start).isEqualTo(40); + } + + @Test + void clockResumesWhenDelegateThrows() { + final ManualClock baseClock = new ManualClock(); + final PausableRelativeClock activityClock = new PausableRelativeClock(baseClock); + final ActivityClockPausingDataOutput output = + new ActivityClockPausingDataOutput<>( + new BusyDelegate(new CollectingDataOutput<>(), baseClock) { + @Override + public void emitRecord(StreamRecord streamRecord) + throws Exception { + throw new Exception("downstream failure"); + } + }, + activityClock); + + assertThatThrownBy(() -> output.emitRecord(new StreamRecord<>(1, 1L))) + .hasMessage("downstream failure"); + + final long afterFailure = activityClock.relativeTimeMillis(); + baseClock.advanceTime(Duration.ofMillis(10)); + assertThat(activityClock.relativeTimeMillis() - afterFailure).isEqualTo(10); + } + + /** A downstream output that burns {@link #BUSY_MILLIS} of wall-clock time on every call. */ + private static class BusyDelegate implements PushingAsyncDataInput.DataOutput { + private final PushingAsyncDataInput.DataOutput delegate; + private final ManualClock clock; + + BusyDelegate(PushingAsyncDataInput.DataOutput delegate, ManualClock clock) { + this.delegate = delegate; + this.clock = clock; + } + + private void burn() { + clock.advanceTime(Duration.ofMillis(BUSY_MILLIS)); + } + + @Override + public void emitRecord(StreamRecord streamRecord) throws Exception { + burn(); + delegate.emitRecord(streamRecord); + } + + @Override + public void emitWatermark(Watermark watermark) throws Exception { + burn(); + delegate.emitWatermark(watermark); + } + + @Override + public void emitWatermarkStatus(WatermarkStatus watermarkStatus) throws Exception { + burn(); + delegate.emitWatermarkStatus(watermarkStatus); + } + + @Override + public void emitLatencyMarker(LatencyMarker latencyMarker) throws Exception { + burn(); + delegate.emitLatencyMarker(latencyMarker); + } + + @Override + public void emitRecordAttributes(RecordAttributes recordAttributes) throws Exception { + burn(); + delegate.emitRecordAttributes(recordAttributes); + } + + @Override + public void emitWatermark(WatermarkEvent watermark) throws Exception { + burn(); + delegate.emitWatermark(watermark); + } + } +} diff --git a/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/util/PausableRelativeClockTest.java b/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/util/PausableRelativeClockTest.java index a7ff6fef8223fd..aac1d588291c3d 100644 --- a/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/util/PausableRelativeClockTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/util/PausableRelativeClockTest.java @@ -77,4 +77,25 @@ void pausedTest() throws Exception { assertThat((durationNanos) / 1_000_000).isEqualTo(TIME_STEP * 2); assertThat(durationMillis).isEqualTo(TIME_STEP * 2); } + + @Test + void layeredClockFollowsBaseClockPauses() { + ManualClock baseClock = new ManualClock(); + PausableRelativeClock parent = new PausableRelativeClock(baseClock); + PausableRelativeClock child = new PausableRelativeClock(parent); + + long startNanos = child.relativeTimeNanos(); + + baseClock.advanceTime(Duration.ofMillis(TIME_STEP)); // counts + parent.pause(); + baseClock.advanceTime(Duration.ofMillis(TIME_STEP)); // parent paused: doesn't count + child.pause(); + parent.unPause(); + baseClock.advanceTime(Duration.ofMillis(TIME_STEP)); // child paused: doesn't count + child.unPause(); + baseClock.advanceTime(Duration.ofMillis(TIME_STEP)); // counts + + long durationNanos = child.relativeTimeNanos() - startNanos; + assertThat(durationNanos / 1_000_000).isEqualTo(TIME_STEP * 2); + } } diff --git a/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/tasks/TestProcessingTimeService.java b/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/tasks/TestProcessingTimeService.java index 2b54390d986293..44c9e30a84c1ad 100644 --- a/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/tasks/TestProcessingTimeService.java +++ b/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/tasks/TestProcessingTimeService.java @@ -72,6 +72,15 @@ public void setCurrentTime(long timestamp) throws Exception { maybeFireTimers(); } + /** + * Advances the clock without firing any timers. This models wall-clock time passing while the + * task thread is busy and therefore unable to service timers, for example inside a slow chained + * operator. + */ + public void advanceClockWithoutFiringTimers(long delta) { + clock.advanceTime(Duration.ofMillis(delta)); + } + private void maybeFireTimers() throws Exception { if (!isQuiesced) { while (!priorityQueue.isEmpty()