[FLINK-40076][tests] Fix race in RescaleTimelineITCase.testRecordNonTerminatedRescaleMergingWithNewRecoverableFailureTriggerCause

[MartijnVisser]

What is the purpose of the change

RescaleTimelineITCase.testRecordNonTerminatedRescaleMergingWithNewRecoverableFailureTriggerCause is flaky (Azure nightly build 76751, leg test_cron_adaptive_scheduler core): it fails with expected: RECOVERABLE_FAILOVER but was: UPDATE_REQUIREMENT.

After killing a TaskManager to trigger a recoverable failover, the test waited via waitForVertexParallelismReachedAndJobRunning(..., PARALLELISM). Because waitUntilCondition checks its predicate before sleeping, that condition is satisfied immediately by the stale pre-failover state (the job is still observed RUNNING at PARALLELISM), so the wait does not synchronize with the failover at all. After the failover the job returns at a lower parallelism (a TaskManager is gone), so PARALLELISM is never seen again, confirming the wait can only match the stale state. The ExecutionGraphInfo snapshot could therefore be taken before the scheduler merged the failover into the in-progress UPDATE_REQUIREMENT rescale and re-stamped its trigger cause, so the test observed UPDATE_REQUIREMENT. This is a test-only race; the scheduler behavior is correct.

Distinct from the earlier races fixed in this class (FLINK-39902, FLINK-39903, FLINK-40009, FLINK-40010) and from FLINK-40067 (#28633), which all target other methods.

Brief change log

• Replace the ineffective parallelism/running wait after the TaskManager termination with a wait until the recoverable failover has been merged into the in-progress rescale (history size 2, latest trigger cause RECOVERABLE_FAILOVER), guarded on the enabled-history parameter (the disabled-history leg records nothing and still asserts an empty history, so it needs no wait).
• The assertion is unchanged; only the synchronization before snapshotting is corrected.

Verifying this change

This change is already covered by the existing test RescaleTimelineITCase#testRecordNonTerminatedRescaleMergingWithNewRecoverableFailureTriggerCause. Verified by running it 25 times (50 executions across both parameterizations); all green. spotless:check passes.

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?

• Yes (please specify the tool below)

Generated-by: Claude Opus 4.8 (via Claude Code)

[flinkbot]

CI report:

• 9edb1bf Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

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

[spuru9]

LGTM
Structure is similar to the hasRescaleHistoryMetCondition with a difference the 3rd argument.

[XComp]

Thanks @MartijnVisser for addressing this flakiness. I added two comments. PTAL

[XComp]

Suggested change

	return rescaleHistory.size() == 2
	&& rescaleHistory.get(0).getTriggerCause()
	== TriggerCause.RECOVERABLE_FAILOVER;
	return rescaleHistory.size() == 2

We shouldn't check for the TriggerCause here. If the trigger cause assertion is wrong for some reason, we would wait for ever here (i.e., till the CI timeout is triggered). We're expecting two rescales here. So, checking the size() is good enough for the wait. The actual assertion will happen afterwards.

That means, w/o the TriggerCause check in the wait loop, we would still cover any bug were the order didn't end up as expected through the assertion later on but fail immediately when two rescales happen (rather than waiting for the CI timeout).

[MartijnVisser]

Thanks for taking a look! I agree with the general principle, but in this specific case the size check alone cannot synchronize with the failover, because the merge does not append a history entry:

• The history is already at size 2 before the TaskManager is terminated: Created#onEnter records the initial rescale, and AdaptiveScheduler#recordRescaleForNewResourceRequirements (called synchronously while processing updateJobResourceRequirements) terminates it and adds the in-progress UPDATE_REQUIREMENT rescale via newRescale(true). That is also what testRecordInProgressRescale asserts, with no failover involved.
• The failover then takes the !rescaleTimeline.isIdling() branch in AdaptiveScheduler#recordRescaleForJobRestarting, which only calls updateRescale(...). In DefaultRescaleTimeline, updateRescale mutates currentRescale, the same object already held by the history queue; newRescale is the only method that appends. So the merge re-stamps the trigger cause of the existing entry and the size stays 2 throughout.

The original CI failure shows exactly that intermediate state: hasSize(2) passed while the trigger cause was still UPDATE_REQUIREMENT. So a size-only wait is satisfied immediately in the pre-merge state and we would re-introduce the race this PR fixes. The trigger-cause re-stamp is the only externally observable signal that the merge happened.

On the "wait forever" concern: together with your second suggestion (very long timeout), a genuine regression in the trigger cause would stall in this wait until the CI timeout, which then produces the thread dump you mentioned, and the logs would be missing the scheduler's "Merge the current non-terminated rescale..." INFO line, which makes the failure diagnosable. I think that is an acceptable trade-off given the merge is an in-place mutation with no other signal to wait on. WDYT?

[XComp]

Good point - looking at it once more, I noticed that the fix would be way easier:

waitForVertexParallelismReachedAndJobRunning(jobGraph, JOB_VERTEX_ID, NUMBER_SLOTS_PER_TASK_MANAGER);

The initial test implementation waited for 4 vertices running which is the case before any scaling or failover happens. The wait conditioning is, therefore, immediately met. The flakiness in the assert happens depending on whether the TaskManager shutdown happens quick enough or not.

Anyway... instead, we want to wait for the TM failover goes through before checking the rescale history. That happens when the job is up and running again with parallelism 2 (i.e. one TM with 2 slots is available).

WDYT?

[MartijnVisser]

Good point, that's much cleaner, I've updated it, the net change against master is now just PARALLELISM → NUMBER_SLOTS_PER_TASK_MANAGER on the existing wait.

[XComp]

nit: we're using different timeouts in different places in this ITCase. It's not a big deal as all it's only about waiting long enough. But we might want to move this into a constant for the test and make it extra long like Duration.ofDays(1).toMillis() to make the test stall long enough for the CI timeout to kick in.

The benefit here would be that we would generate a thread dump that might give us hints into why the test didn't continue. Having a test local timeout prevents this thread dump from being generated. We would just see a test-local timeout making the unit test fail.

[MartijnVisser]

Good call, I'll fix that. I'll wait until we've come to a conclusion on the top comment though

[MartijnVisser]

Since the wait is now dropped, a real stall would run into the CI timeout and produce the thread dump you're after.