Skip to content

[FLINK-40093][Runtime] Resume idle splits by alignment check if previously paused #28689

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Efrat19 wants to merge 1 commit into
base: master
Choose a base branch
from
+61 −6
Open

[FLINK-40093][Runtime] Resume idle splits by alignment check if previously paused #28689

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 2 additions & 3 deletions ...src/main/java/org/apache/flink/runtime/metrics/groups/InternalSourceSplitMetricGroup.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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();
}
Expand Down
10 changes: 7 additions & 3 deletions flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -848,12 +848,16 @@ private void checkSplitWatermarkAlignment() {
Collection<String> splitsToResume = new ArrayList<>();
sampledSplitWatermarks.forEach(
(splitId, splitWatermarks) -> {
if (currentlyIdleSplits.contains(splitId)) {
return;
}
if (splitWatermarks.getOldestSample() > currentMaxDesiredWatermark) {

@SteveStevenpoor SteveStevenpoor Jul 8, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I’m still concerned about the case where we mark an idle split as paused: markPaused() marks it as not idle, but the split remains in currentlyIdleSplits.

So we can get the following sequence:

  1. A split is paused.
  2. During the pause process, the split is marked idle.
  3. reportPausedOrResumed() is called and marks the split as not idle.
  4. However, the split still remains in currentlyIdleSplits.

As a result, before the next alignment, the split can be non-idle according to the metric group, but still present in both currentlyIdleSplits and currentlyPausedSplits.

I’m not sure whether this affects timers or checks like:

idle == currentlyIdleSplits.contains(splitId)

but this inconsistency still looks risky to me.

@Efrat19 Efrat19 Jul 12, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

we mark an idle split as paused: markPaused() marks it as not idle, but the split remains in currentlyIdleSplits.

Pause is skipped for idle splits.

// 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);
}
});
Expand Down
52 changes: 52 additions & 0 deletions ...a/org/apache/flink/streaming/api/operators/SourceOperatorSplitWatermarkAlignmentTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<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(4);
split0.addRecord(5);
split0.addRecord(6);
split0.addRecord(7);
split0.addRecord(8);
operator.handleOperatorEvent(
new AddSplitEvent<>(Arrays.asList(split0), new MockSourceSplitSerializer()));
final CollectingDataOutput<Integer> 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);

@SteveStevenpoor SteveStevenpoor Jul 8, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Here you mark the split as idle only after the full pauseOrResumeSplits process has finished. To simulate the issue being addressed, you can use something like this:

    private static class RaceInjectingMockSourceReader extends MockSourceReader {
        private Runnable afterNextPause = () -> {};

        RaceInjectingMockSourceReader(
                WaitingForSplits waitingForSplitsBehaviour,
                boolean markIdleOnNoSplits,
                boolean usePerSplitOutputs) {
            super(waitingForSplitsBehaviour, markIdleOnNoSplits, usePerSplitOutputs);
        }

        void runAfterNextPause(Runnable afterNextPause) {
            this.afterNextPause = afterNextPause;
        }

        @Override
        public void pauseOrResumeSplits(
                Collection<String> splitsToPause, Collection<String> splitsToResume) {
            super.pauseOrResumeSplits(splitsToPause, splitsToResume);

            if (!splitsToPause.isEmpty()) {
                Runnable callback = afterNextPause;
                afterNextPause = () -> {};
                callback.run();
            }
        }
    }

Then in the test, you can use:

  sourceReader.runAfterNextPause(
                () -> operator.updateCurrentSplitIdle(split0.splitId(), true));

This will mark the split idle right after the actual pause.

    private void pauseOrResumeSplits(
            Collection<String> splitsToPause, Collection<String> splitsToResume) {
        try {
            LOG.info(
                    "pauseOrResumeSplits [splitsToPause={}][splitsToResume={}][idleSplits={}]"
                            + "[currentMaxDesiredWatermark={}][latestWatermark={}][oldestWatermark={}]",
                    splitsToPause,
                    splitsToResume,
                    currentlyIdleSplits,
                    currentMaxDesiredWatermark,
                    sampledLatestWatermark.getLatest(),
                    sampledLatestWatermark.getOldestSample());
            sourceReader.pauseOrResumeSplits(splitsToPause, splitsToResume);
            // the split is marked idle
            eventTimeLogic.pauseOrResumeSplits(splitsToPause, splitsToResume);
            reportPausedOrResumed(splitsToPause, splitsToResume);
        } catch (UnsupportedOperationException e) {
            if (!allowUnalignedSourceSplits) {
                throw e;
            }
        }
    }

@Efrat19 Efrat19 Jul 12, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks!
This is really nice, though I don't think such complexity is required when we can simply simulate by calling updateCurrentSplitIdle

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());
}
Expand Down