[FLINK-38403][tests] Fix the unexpected test that the second job does not restore from checkpoint

Author: 1996fanrui

flink-tests/src/test/java/org/apache/flink/test/checkpointing/UnalignedCheckpointRescaleITCase.java

@@ -54,13 +54,13 @@
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
 
-import java.io.File;
 import java.util.Arrays;
 import java.util.BitSet;
 import java.util.Collections;
 
 import static org.apache.flink.api.common.eventtime.WatermarkStrategy.noWatermarks;
 import static org.apache.flink.util.Preconditions.checkState;
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.hamcrest.Matchers.equalTo;
 
 /** Integration test for performing rescale of unaligned checkpoint. */
@@ -619,8 +619,10 @@ public void shouldRescaleUnalignedCheckpoint() throws Exception {
                         .setExpectedFailures(1)
                         .setSourceSleepMs(sourceSleepMs);
         prescaleSettings.setGenerateCheckpoint(true);
-        final File checkpointDir = super.execute(prescaleSettings);
-
+        final String checkpointDir = super.execute(prescaleSettings);
+        assertThat(checkpointDir)
+                .as("First job must generate a checkpoint for rescale test to be valid.")
+                .isNotNull();
         // resume
         final UnalignedSettings postscaleSettings =
                 new UnalignedSettings(topology)

flink-tests/src/test/java/org/apache/flink/test/checkpointing/UnalignedCheckpointTestBase.java

@@ -73,6 +73,7 @@
 
 import org.apache.flink.shaded.guava33.com.google.common.collect.Iterables;
 
+import org.assertj.core.api.Fail;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
 import org.junit.Rule;
@@ -150,7 +151,7 @@ public static void afterAll() {
     }
 
     @Nullable
-    protected File execute(UnalignedSettings settings) throws Exception {
+    protected String execute(UnalignedSettings settings) throws Exception {
         final File checkpointDir = temp.newFolder();
         Configuration conf = settings.getConfiguration(checkpointDir);
 
@@ -179,14 +180,15 @@ protected File execute(UnalignedSettings settings) throws Exception {
         final StreamExecutionEnvironment env =
                 StreamExecutionEnvironment.getExecutionEnvironment(conf);
         settings.configure(env);
+        JobID jobID = null;
         try {
             // print the test parameters to help debugging when the case is stuck
             System.out.println(
                     "Starting " + getClass().getCanonicalName() + "#" + name.getMethodName() + ".");
             final CompletableFuture<JobSubmissionResult> result =
                     miniCluster.getMiniCluster().submitJob(streamGraph.getJobGraph());
 
-            final JobID jobID = result.get().getJobID();
+            jobID = result.get().getJobID();
             checkCounters(
                     miniCluster
                             .getMiniCluster()
@@ -198,13 +200,17 @@ protected File execute(UnalignedSettings settings) throws Exception {
             if (settings.generateCheckpoint) {
                 return CommonTestUtils.getLatestCompletedCheckpointPath(
                                 jobID, miniCluster.getMiniCluster())
-                        .map(File::new)
-                        .orElseThrow(() -> new AssertionError("Could not generate checkpoint"));
+                        .orElseThrow(() -> Fail.fail("Could not generate checkpoint"));
             }
         } catch (Exception e) {
             if (ExceptionUtils.findThrowable(e, TestException.class).isEmpty()) {
                 throw e;
             }
+            if (settings.generateCheckpoint) {
+                return CommonTestUtils.getLatestCompletedCheckpointPath(
+                                jobID, miniCluster.getMiniCluster())
+                        .orElseThrow(() -> Fail.fail("Could not generate checkpoint"));
+            }
         } finally {
             miniCluster.after();
         }
@@ -680,7 +686,7 @@ public String toString() {
     protected static class UnalignedSettings {
         private int parallelism;
         private final int minCheckpoints = 10;
-        @Nullable private File restoreCheckpoint;
+        @Nullable private String restoreCheckpoint;
         private boolean generateCheckpoint = false;
         int expectedFailures = 0;
         int tolerableCheckpointFailures = 0;
@@ -701,7 +707,7 @@ public UnalignedSettings setParallelism(int parallelism) {
             return this;
         }
 
-        public UnalignedSettings setRestoreCheckpoint(File restoreCheckpoint) {
+        public UnalignedSettings setRestoreCheckpoint(String restoreCheckpoint) {
             this.restoreCheckpoint = restoreCheckpoint;
             return this;
         }
@@ -772,7 +778,7 @@ public Configuration getConfiguration(File checkpointDir) {
             conf.set(StateBackendOptions.STATE_BACKEND, "hashmap");
             conf.set(CheckpointingOptions.CHECKPOINTS_DIRECTORY, checkpointDir.toURI().toString());
             if (restoreCheckpoint != null) {
-                conf.set(StateRecoveryOptions.SAVEPOINT_PATH, restoreCheckpoint.toURI().toString());
+                conf.set(StateRecoveryOptions.SAVEPOINT_PATH, restoreCheckpoint);
             }
 
             conf.set(