From 6dcba9677d43a311944fe382ecc8c21d9f77e04f Mon Sep 17 00:00:00 2001 From: Efrat Levitan <41479945+Efrat19@users.noreply.github.com> Date: Tue, 7 Jul 2026 19:22:25 +0300 Subject: [PATCH] [FLINK-40093][Runtime] Resume idle splits by alignment check if previously paused Usually pauseOrResumeSplits pauses idleness timer for the split so it isn't marked idle while paused. However with low idleness timeout (observed with 1s) + low allowed WM drift, a race condition could cause paused splits to never resume though they have records: 1. A split becomes paused due to too advanced records. 2. pauseOrResumeSplits pauses the split. 3. pauseOrResumeSplits reaches to pause the split idleness clock but is {idlenessTimeout} too late, and the split becomes idle. 4. The watermark advances but the split is excluded from the watermark alignment check due to its idleness. 5. More records arrive but the split is paused at the connector level so they aren't processed, nor seen by watermarkGenerator so it still considers the split idle The PR aims to fix it by preserving the part where idle splits are excluded from alignment pause (to not override their idle status) while allowing alignment check to resume splits even if they are currently idle. They are considered idle until they emit the next record. --- .../InternalSourceSplitMetricGroup.java | 5 +- .../api/operators/SourceOperator.java | 10 ++-- ...ceOperatorSplitWatermarkAlignmentTest.java | 52 +++++++++++++++++++ 3 files changed, 61 insertions(+), 6 deletions(-) diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/InternalSourceSplitMetricGroup.java b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/InternalSourceSplitMetricGroup.java index bc522d228b3f5a..a379a434764940 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/InternalSourceSplitMetricGroup.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/InternalSourceSplitMetricGroup.java @@ -121,10 +121,9 @@ public void markPaused() { public void markIdle() { maybeMarkSplitStart(); if (isPaused()) { - // If a split is marked idle, it has no records to emit. - // hence it shouldn't be considered paused anymore markNotPaused(); - LOG.warn("[{}] Split marked idle while still paused", splitId); + // This is benign: idleness takes over paused state if they race + LOG.info("[{}] Split marked idle while still paused", splitId); } this.idleTimePerSecond.markStart(); } 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 f153617b2c4feb..c487aba5255e94 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 @@ -848,12 +848,16 @@ private void checkSplitWatermarkAlignment() { Collection splitsToResume = new ArrayList<>(); sampledSplitWatermarks.forEach( (splitId, splitWatermarks) -> { - if (currentlyIdleSplits.contains(splitId)) { - return; - } 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); } }); 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 b14639a9fcc31f..a635064d949d9b 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 @@ -569,6 +569,58 @@ void testAlignmentCheckIsDeferredForIdleSplits() throws Exception { 0L, operator.getSplitMetricGroup(split0.splitId()).getAccumulatedPausedTime()); } + @Test + void testPausedIdleSplitsCanBeResumedByAlignmentCheck() throws Exception { + final long idleTimeout = 100; + final MockSourceReader sourceReader = + new MockSourceReader(WaitingForSplits.DO_NOT_WAIT_FOR_SPLITS, true, true); + final TestProcessingTimeService processingTimeService = new TestProcessingTimeService(); + final SourceOperator operator = + createAndOpenSourceOperatorWithIdleness( + sourceReader, processingTimeService, idleTimeout); + + final MockSourceSplit split0 = new MockSourceSplit(0, 0, 10); + final int allowedWatermark4 = 4; + final int allowedWatermark7 = 7; + split0.addRecord(4); + split0.addRecord(5); + split0.addRecord(6); + split0.addRecord(7); + split0.addRecord(8); + operator.handleOperatorEvent( + new AddSplitEvent<>(Arrays.asList(split0), new MockSourceSplitSerializer())); + final CollectingDataOutput actualOutput = new CollectingDataOutput<>(); + + // Emit enough records to fill the sampler buffer + for (int i = 0; i < WATERMARK_ALIGNMENT_BUFFER_SIZE.defaultValue(); i++) { + operator.emitNext(actualOutput); + processingTimeService.advance(idleTimeout - 1); + } + sampleAllWatermarks(processingTimeService); + assertOutput(actualOutput, Arrays.asList(4, 5, 6)); + + // Alignment check fires and pauses the split + operator.handleOperatorEvent(new WatermarkAlignmentEvent(allowedWatermark4)); + assertThat(operator.getSplitMetricGroup(split0.splitId()).isPaused()).isTrue(); + assertThat(sourceReader.getPausedSplits()).containsExactly("0"); + assertOutput(actualOutput, Arrays.asList(4, 5, 6)); + + // Normally idlenessTimer can't elapse while the split is paused + // So calling it manually to simulate a race condition + operator.updateCurrentSplitIdle(split0.splitId(), true); + assertThat(operator.getSplitMetricGroup(split0.splitId()).isIdle()).isTrue(); + + // Watermark advances and resumes the split (Though it is still considered idle) + operator.handleOperatorEvent(new WatermarkAlignmentEvent(allowedWatermark7)); + assertThat(operator.getSplitMetricGroup(split0.splitId()).isIdle()).isTrue(); + assertThat(sourceReader.getPausedSplits()).isEmpty(); + + // The split emits a record and breaks out of idleness + operator.emitNext(actualOutput); // 7 + assertThat(operator.getSplitMetricGroup(split0.splitId()).isActive()).isTrue(); + assertOutput(actualOutput, Arrays.asList(4, 5, 6, 7)); + } + private void sampleAllWatermarks(TestProcessingTimeService timeService) throws Exception { sampleWatermarks(timeService, WATERMARK_ALIGNMENT_BUFFER_SIZE.defaultValue()); }