XComp commented on code in PR #28645:
URL: https://github.com/apache/flink/pull/28645#discussion_r3534611700
##########
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;
+ },
+ 10000);
Review Comment:
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.
##########
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:
```suggestion
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).
--
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]
