[FLINK-26977] Unify SavepointITCase#testStopWithSavepointFailingAfter…

…SnapshotCreation for both schedulers This closes apache#19439
JasonLeeCoding · May 27, 2022 · 3ecd0a1 · 3ecd0a1
1 parent 29f41a2
commit 3ecd0a1
Showing 1 changed file with 6 additions and 40 deletions.
diff --git a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/SavepointITCase.java b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/SavepointITCase.java
@@ -1026,22 +1026,12 @@ public void testStopWithSavepointFailingAfterSnapshotCreation() throws Exception
                 // 2. job failover triggered by SchedulerBase.stopWithSavepoint
                 0,
                 (jobId, actualException) -> {
-                    if (ClusterOptions.isAdaptiveSchedulerEnabled(new Configuration())) {
-                        return actualException
-                                .getMessage()
-                                .contains("Stop with savepoint operation could not be completed");
-                    } else {
-                        Optional<StopWithSavepointStoppingException> actualFlinkException =
-                                findThrowable(
-                                        actualException, StopWithSavepointStoppingException.class);
-                        return actualFlinkException
-                                .map(
-                                        e ->
-                                                e.getMessage()
-                                                        .startsWith(
-                                                                "A savepoint has been created at:"))
-                                .orElse(false);
-                    }
+                    Optional<StopWithSavepointStoppingException> actualFlinkException =
+                            findThrowable(
+                                    actualException, StopWithSavepointStoppingException.class);
+                    return actualFlinkException
+                            .map(e -> e.getMessage().startsWith("A savepoint has been created at:"))
+                            .orElse(false);
                 },
                 false);
     }
@@ -1096,30 +1086,6 @@ public void testStopWithSavepointWithDrainGlobalFailoverIfSavepointAborted() thr
         }
     }
 
-    private static BiFunction<JobID, ExecutionException, Boolean>
-            assertAfterSnapshotCreationFailure() {
-        return (jobId, actualException) -> {
-            if (ClusterOptions.isAdaptiveSchedulerEnabled(new Configuration())) {
-                return actualException
-                        .getMessage()
-                        .contains("Stop with savepoint operation could not be completed");
-            } else {
-                Optional<FlinkException> actualFlinkException =
-                        findThrowable(actualException, FlinkException.class);
-                if (!actualFlinkException.isPresent()) {
-                    return false;
-                }
-                return actualFlinkException
-                        .get()
-                        .getMessage()
-                        .contains(
-                                String.format(
-                                        "A global fail-over is triggered to recover the job %s.",
-                                        jobId));
-            }
-        };
-    }
-
     private static BiFunction<JobID, ExecutionException, Boolean>
             assertInSnapshotCreationFailure() {
         return (ignored, actualException) -> {