From 501bd707c5993394c8c6cafefb49eb4f4b04c28e 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 | 50 +++++++++++++++++++ 3 files changed, 59 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 05b279b66c17df..6e0331c3d83ac6 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 @@ -780,12 +780,16 @@ private void checkSplitWatermarkAlignment() { Collection splitsToResume = new ArrayList<>(); splitCurrentWatermarks.forEach( (splitId, splitWatermark) -> { - if (currentlyIdleSplits.contains(splitId)) { - return; - } if (splitWatermark > 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 f184c3fac14fb0..21c00e44b5d3b9 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 @@ -547,6 +547,56 @@ 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<>(); + + for (int i = 0; i < 3; i++) { + operator.emitNext(actualOutput); + processingTimeService.advance(idleTimeout - 1); + } + 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 assertOutput( CollectingDataOutput actualOutput, List expectedOutput) { assertThat(