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

[Efrat19]

What is the purpose of the change

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.

Brief change log

• Allow idle splits to be resumed by alignment check

Verifying this change

This change added tests and can be verified as follows:

• SourceOperatorSplitWatermarkAlignmentTest#testPausedIdleSplitsCanBeResumedByAlignmentCheck

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): no
• The public API, i.e., is any changed class annotated with @Public(Evolving): no
• The serializers: no
• The runtime per-record code paths (performance sensitive): no
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
• The S3 file system connector: no

Documentation

• Does this pull request introduce a new feature? no
• If yes, how is the feature documented? not applicable

---

Was generative AI tooling used to co-author this PR?

I used claude-sonnet-5 for the unit test

[flinkbot]

CI report:

• 6dcba96 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[SteveStevenpoor]

@Efrat19 Thank you for the PR. I have a few comments and suggestions.

[SteveStevenpoor]

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]

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

[SteveStevenpoor]

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]

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.