This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch release-2.3
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-2.3 by this push:
new 2a2d35906ce [FLINK-40093][Runtime] Resume idle splits by alignment
check if previously paused
2a2d35906ce is described below
commit 2a2d35906ceb27e1aae2ca715da2aa24c9189bb7
Author: Efrat Levitan <[email protected]>
AuthorDate: Tue Jul 7 19:22:25 2026 +0300
[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.
---
.../groups/InternalSourceSplitMetricGroup.java | 5 +-
.../streaming/api/operators/SourceOperator.java | 10 ++--
.../SourceOperatorSplitWatermarkAlignmentTest.java | 60 ++++++++++++++++++++++
3 files changed, 69 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 bc522d228b3..a379a434764 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 class InternalSourceSplitMetricGroup extends
ProxyMetricGroup<MetricGroup
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 f153617b2c4..c487aba5255 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 @@ public class SourceOperator<OUT, SplitT extends
SourceSplit> extends AbstractStr
Collection<String> 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 b14639a9fcc..36fda4eee68 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,66 @@ class SourceOperatorSplitWatermarkAlignmentTest {
0L,
operator.getSplitMetricGroup(split0.splitId()).getAccumulatedPausedTime());
}
+ @Test
+ void testPausedIdleSplitsCanBeResumedByAlignmentCheck() throws Exception {
+ final long idleTimeout = 1000;
+ final long autoWatermarkIntervalMillis = 100;
+ final long marginMillis = autoWatermarkIntervalMillis / 2;
+ final MockSourceReader sourceReader =
+ new MockSourceReader(WaitingForSplits.DO_NOT_WAIT_FOR_SPLITS,
true, true);
+ final TestProcessingTimeService processingTimeService = new
TestProcessingTimeService();
+ processingTimeService.setCurrentTime(0);
+ final SourceOperator<Integer, MockSourceSplit> operator =
+ createAndOpenSourceOperatorWithIdleness(
+ sourceReader, processingTimeService, idleTimeout);
+
+ final MockSourceSplit split0 = new MockSourceSplit(0, 0, 10);
+ final int allowedWatermark4 = 4;
+ final int allowedWatermark7 = 7;
+ split0.addRecord(5);
+ split0.addRecord(6);
+ operator.handleOperatorEvent(
+ new AddSplitEvent<>(
+ Collections.singletonList(split0), new
MockSourceSplitSerializer()));
+ final CollectingDataOutput<Integer> actualOutput = new
CollectingDataOutput<>();
+
+ operator.emitNext(actualOutput); // 5
+ sampleAllWatermarks(
+ processingTimeService); // fill the alignment ring buffer with
watermark=5
+ final long timeElapsed =
+ WATERMARK_ALIGNMENT_BUFFER_SIZE.defaultValue() *
updateIntervalMillis;
+
+ // Advancing in small steps (TestProcessingTimeService limitation)
+ processingTimeService.advance(2 * autoWatermarkIntervalMillis -
timeElapsed); // t=200
+ processingTimeService.advance(idleTimeout); // t=1200
+ processingTimeService.advance(marginMillis); // t=1250
+
+ // Firing alignment check at t=1250.
+ // Idleness clock will be now paused 250ms past the timeout (1000),
+ // which means it will fire on the next idleness periodic check
+ operator.handleOperatorEvent(new
WatermarkAlignmentEvent(allowedWatermark4));
+
assertThat(operator.getSplitMetricGroup(split0.splitId()).isPaused()).isTrue();
+ assertThat(sourceReader.getPausedSplits()).containsExactly("0");
+ assertOutput(actualOutput, Arrays.asList(5));
+
+ // Triggerring the next idleness periodic check
+ // which marks the split idle while paused.
+ processingTimeService.advance(autoWatermarkIntervalMillis +
marginMillis); // t=1400
+
assertThat(operator.getSplitMetricGroup(split0.splitId()).isIdle()).isTrue();
+ assertThat(sourceReader.getPausedSplits()).containsExactly("0");
+
+ // Watermark advances and resumes the split
+ // (Though it is still considered idle - this is what this commit is
about)
+ 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); // 6
+
assertThat(operator.getSplitMetricGroup(split0.splitId()).isActive()).isTrue();
+ assertOutput(actualOutput, Arrays.asList(5, 6));
+ }
+
private void sampleAllWatermarks(TestProcessingTimeService timeService)
throws Exception {
sampleWatermarks(timeService,
WATERMARK_ALIGNMENT_BUFFER_SIZE.defaultValue());
}
