MartijnVisser commented on code in PR #28645:
URL: https://github.com/apache/flink/pull/28645#discussion_r3534924591
##########
flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/timeline/RescaleTimelineITCase.java:
##########
@@ -567,7 +567,21 @@ void
testRecordNonTerminatedRescaleMergingWithNewRecoverableFailureTriggerCause(
miniCluster.terminateTaskManager(0);
- waitForVertexParallelismReachedAndJobRunning(jobGraph, JOB_VERTEX_ID,
PARALLELISM);
+ // A parallelism/running wait cannot synchronize with the failover
here: it is satisfied
+ // immediately by the stale pre-failover state, and the post-failover
job runs at a lower
+ // parallelism. Instead, wait until the failover has been merged into
the in-progress
+ // rescale before snapshotting. Guarded by if (not assumeThat) because
the disabled-history
+ // leg records nothing and must still assert an empty history below.
+ if (enabledRescaleHistory(configuration)) {
+ waitUntilConditionWithTimeout(
+ () -> {
+ List<Rescale> rescaleHistory =
getRescaleHistory(miniCluster, jobGraph);
+ return rescaleHistory.size() == 2
+ && rescaleHistory.get(0).getTriggerCause()
+ == TriggerCause.RECOVERABLE_FAILOVER;
Review Comment:
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?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
