[FLINK-38403][tests] Fix the unexpected test that the second job does not restore from checkpoint #27254
Conversation
| conf.set(StateRecoveryOptions.SAVEPOINT_PATH, restoreCheckpoint.toURI().toString()); | ||
| conf.set(StateRecoveryOptions.SAVEPOINT_PATH, restoreCheckpoint); |
There was a problem hiding this comment.
It adds the wrong prefix, so updated restoreCheckpoint to String.
AHeise
left a comment
AHeise
left a comment
There was a problem hiding this comment.
Thank you very much for fixing this test! Left two remarks that need to be addressed before approval.
| assertNotNull( | ||
| "First job must generate a checkpoint for rescale test to be valid.", | ||
| checkpointDir); |
There was a problem hiding this comment.
Please use assertj (assertThat(checkpointDir).as("First job must generate a checkpoint for rescale test to be valid.").isNotNull)
| return CommonTestUtils.getLatestCompletedCheckpointPath( | ||
| jobID, miniCluster.getMiniCluster()) | ||
| .map(File::new) | ||
| .orElseThrow(() -> new AssertionError("Could not generate checkpoint")); |
There was a problem hiding this comment.
Should we Fail.fail("Expected exception") here?
1996fanrui
force-pushed
the
38403/ITCase-rescale-not-restore
branch
from
80692f1 to
4e675d2
Compare
1996fanrui
left a comment
1996fanrui
left a comment
There was a problem hiding this comment.
Thanks @AHeise for the quick review, all comments are addressed.
1996fanrui
force-pushed
the
38403/ITCase-rescale-not-restore
branch
from
4e675d2 to
736ad29
Compare
|
fyi: to have green ci, rebase to the latest master |
Thanks @snuyanzin for the reminder, I am still changing the PR, and I will rebase master branch before next push. |
… not restore from checkpoint
…ished eventually Add comments to help understand the UnalignedCheckpointRescaleITCase
1996fanrui
force-pushed
the
38403/ITCase-rescale-not-restore
branch
from
736ad29 to
8938499
Compare
| * <p>Postscale phase: Job restores from checkpoint with different parallelism, failovers once, | ||
| * and finishes after source generates all records. |
There was a problem hiding this comment.
The second job expects failover once, and finishes after source generates all records. So removing @ThrowableAnnotation(ThrowableType.NonRecoverableError) for TestException.
Also, I introduced ExpectedFinalJobStatus in UnalignedSettings to check the final JobStatus.
| RestartStrategyUtils.configureFixedDelayRestartStrategy( | ||
| env, generateCheckpoint ? expectedFailures / 2 : expectedFailures, 100L); |
There was a problem hiding this comment.
What's the reasoning behind the calculated number of expected failures?
There was a problem hiding this comment.
It reverts the change in https://github.com/apache/flink/pull/27119/files#diff-ace775e80e66d4f4001bdaea6bcbaae1975bd5e9a5497532d8d7152e4090069aL752
The original intention is :
generateCheckpointcontrols the operational phase: true is for the job before rescaling, and false is for the new job after rescaling- The value
expectedFailures / 2acts as the failure threshold for the first job. This setup ensures that the first job fails after half of the expected exceptions are met, allowing the second job to automatically recover from the generated checkpoint and continue consumption.
5 participants
What is the purpose of the change
Find latest checkpointshould be placed in catch block as well,Get more details from https://github.com/apache/flink/pull/27119/files#r2542189639
@ThrowableAnnotation(ThrowableType.NonRecoverableError)for TestException.Also, I introduced ExpectedFinalJobStatus in UnalignedSettings to check the final JobStatus.
Get more details from #27254 (comment)
Brief change log
[FLINK-38403][tests] Fix the unexpected test that the second job does not restore from checkpoint