This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch release-2.2
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-2.2 by this push:
new 342fb954921 [FLINK-40093][Runtime] Resume idle splits by alignment
check if previously paused
342fb954921 is described below
commit 342fb95492144f810239bd3f0b8d46389f466df9
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 | 56 ++++++++++++++++++++++
3 files changed, 65 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 05b279b66c1..6e0331c3d83 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 @@ public class SourceOperator<OUT, SplitT extends
SourceSplit> extends AbstractStr
Collection<String> 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 f184c3fac14..6a95af5b091 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,62 @@ 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
+
+ // Advancing in small steps (TestProcessingTimeService limitation)
+ processingTimeService.advance(2 * autoWatermarkIntervalMillis); //
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 assertOutput(
CollectingDataOutput<Integer> actualOutput, List<Integer>
expectedOutput) {
assertThat(
