Cleaning up and refactoring test code.

Author: schongloo

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/bluegreen/BlueGreenDeploymentService.java

@@ -565,7 +565,7 @@ private UpdateControl<FlinkBlueGreenDeployment> abortDeployment(
     }
 
     @NotNull
-    public static UpdateControl<FlinkBlueGreenDeployment> markDeploymentFailing(
+    private static UpdateControl<FlinkBlueGreenDeployment> markDeploymentFailing(
             BlueGreenContext context, String error) {
         LOG.error(error);
         return patchStatusUpdateControl(context, null, JobStatus.FAILING, error);

flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkBlueGreenDeploymentControllerTest.java

@@ -456,12 +456,11 @@ public void verifySavepointFailureRecovery(FlinkVersion flinkVersion) throws Exc
                 new IllegalStateException("Job not in valid state for savepoint"));
 
         String customValue = UUID.randomUUID().toString();
-        simulateChangeInSpec(rs.deployment, customValue, 0, null);
+        simulateSpecChange(rs.deployment, customValue);
 
         // Should fail with savepoint error
         rs = reconcile(rs.deployment);
-        assertFailingJobStatus(rs);
-        assertTrue(rs.reconciledStatus.getError().contains("Job not in valid state for savepoint"));
+        assertFailingWithError(rs, "Job not in valid state for savepoint");
 
         // Recovery: Clear the exception and try again with new spec change
         flinkService.clearSavepointTriggerException();
@@ -478,8 +477,9 @@ public void verifySavepointFailureRecovery(FlinkVersion flinkVersion) throws Exc
     @ParameterizedTest
     @MethodSource("savepointExceptionProvider")
     public void verifySavepointFailureWithDifferentExceptionTypes(
-            FlinkVersion flinkVersion, Exception savepointException, String expectedErrorFragment)
-            throws Exception {
+            FlinkVersion flinkVersion, Exception savepointException) throws Exception {
+
+        String expectedErrorFragment = savepointException.getMessage();
 
         var blueGreenDeployment =
                 buildSessionCluster(
@@ -511,6 +511,8 @@ public void verifySavepointFailureWithDifferentExceptionTypes(
     @ParameterizedTest
     @MethodSource("org.apache.flink.kubernetes.operator.TestUtils#flinkVersions")
     public void verifySavepointFetchFailureRecovery(FlinkVersion flinkVersion) throws Exception {
+        String error = "Savepoint corrupted or not found";
+
         var blueGreenDeployment =
                 buildSessionCluster(
                         TEST_DEPLOYMENT_NAME,
@@ -522,22 +524,17 @@ public void verifySavepointFetchFailureRecovery(FlinkVersion flinkVersion) throw
         var rs = executeBasicDeployment(flinkVersion, blueGreenDeployment, false, null);
 
         String customValue = UUID.randomUUID().toString();
-        simulateChangeInSpec(rs.deployment, customValue, 0, null);
+        simulateSpecChange(rs.deployment, customValue);
 
         // Trigger savepoint successfully and go through savepointing flow
         rs = handleSavepoint(rs);
 
-        // Now configure service to return fetch error - this will be detected in
-        // configureInitialSavepoint
-        flinkService.setSavepointFetchError("Savepoint corrupted or not found");
+        // Configure service to return fetch error
+        flinkService.setSavepointFetchError(error);
 
-        // The next reconciliation should transition back to ACTIVE_BLUE and then try
-        // startTransition
-        // which will fail in configureInitialSavepoint due to fetch error
+        // The next reconciliation will fail in configureInitialSavepoint due to fetch error
         rs = reconcile(rs.deployment);
-        assertFailingJobStatus(rs);
-        assertTrue(rs.reconciledStatus.getError().contains("Could not start Transition"));
-        assertTrue(rs.reconciledStatus.getError().contains("Savepoint corrupted or not found"));
+        assertFailingWithError(rs, "Could not start Transition", error);
 
         // Recovery: Clear the fetch error and try again with new spec change
         flinkService.clearSavepointFetchError();
@@ -552,10 +549,9 @@ public void verifySavepointFetchFailureRecovery(FlinkVersion flinkVersion) throw
     }
 
     @ParameterizedTest
-    @MethodSource("savepointFetchErrorProvider")
+    @MethodSource("savepointErrorProvider")
     public void verifySavepointFetchFailureWithDifferentErrors(
-            FlinkVersion flinkVersion, String fetchError, String expectedErrorFragment)
-            throws Exception {
+            FlinkVersion flinkVersion, String errorMessage, boolean isFetchError) throws Exception {
 
         var blueGreenDeployment =
                 buildSessionCluster(
@@ -566,23 +562,25 @@ public void verifySavepointFetchFailureWithDifferentErrors(
                         UpgradeMode.SAVEPOINT);
         var rs = executeBasicDeployment(flinkVersion, blueGreenDeployment, false, null);
 
-        simulateChangeInSpec(rs.deployment, UUID.randomUUID().toString(), 0, null);
+        simulateSpecChange(rs.deployment, UUID.randomUUID().toString());
 
-        // Trigger savepoint successfully and go through savepointing flow
-        rs = handleSavepoint(rs);
+        if (isFetchError) {
+            // Trigger savepoint successfully and go through savepointing flow
+            rs = handleSavepoint(rs);
 
-        // Configure service to return fetch error - this will be detected in
-        // configureInitialSavepoint
-        flinkService.setSavepointFetchError(fetchError);
+            // Configure service to return fetch error
+            flinkService.setSavepointFetchError(errorMessage);
 
-        // The next reconciliation should transition back to ACTIVE_BLUE and then try
-        // startTransition
-        // which will fail in configureInitialSavepoint due to fetch error
-        rs = reconcile(rs.deployment);
+            // The next reconciliation will fail in configureInitialSavepoint due to fetch error
+            rs = reconcile(rs.deployment);
+            assertFailingWithError(rs, "Could not start Transition", errorMessage);
+        } else {
+            // Configure service to throw trigger exception
+            flinkService.setSavepointTriggerException(new RuntimeException(errorMessage));
 
-        assertFailingJobStatus(rs);
-        assertTrue(rs.reconciledStatus.getError().contains("Could not start Transition"));
-        assertTrue(rs.reconciledStatus.getError().contains(expectedErrorFragment));
+            rs = reconcile(rs.deployment);
+            assertFailingWithError(rs, "Could not trigger Savepoint", errorMessage);
+        }
 
         // Should remain in ACTIVE_BLUE state after failure
         assertEquals(
@@ -595,28 +593,23 @@ public void verifySavepointFetchFailureWithDifferentErrors(
 
     // ==================== Parameterized Test Inputs ====================
 
-    static Stream<Arguments> savepointFetchErrorProvider() {
+    static Stream<Arguments> savepointErrorProvider() {
         return TestUtils.flinkVersions()
                 .flatMap(
                         flinkVersionArgs -> {
                             FlinkVersion version = (FlinkVersion) flinkVersionArgs.get()[0];
                             return Stream.of(
+                                    // Fetch errors
+                                    Arguments.of(version, "Savepoint file corrupted", true),
+                                    Arguments.of(version, "Storage system unavailable", true),
                                     Arguments.of(
-                                            version,
-                                            "Savepoint file corrupted",
-                                            "Savepoint file corrupted"),
-                                    Arguments.of(
-                                            version,
-                                            "Storage system unavailable",
-                                            "Storage system unavailable"),
-                                    Arguments.of(
-                                            version,
-                                            "Access denied to savepoint location",
-                                            "Access denied to savepoint location"),
-                                    Arguments.of(
-                                            version,
-                                            "Savepoint metadata missing",
-                                            "Savepoint metadata missing"));
+                                            version, "Access denied to savepoint location", true),
+                                    Arguments.of(version, "Savepoint metadata missing", true),
+                                    // Trigger exceptions
+                                    Arguments.of(version, "Network timeout", false),
+                                    Arguments.of(version, "Job not running", false),
+                                    Arguments.of(version, "Service unavailable", false),
+                                    Arguments.of(version, "Generic error", false));
                         });
     }
 
@@ -626,22 +619,12 @@ static Stream<Arguments> savepointExceptionProvider() {
                         flinkVersionArgs -> {
                             FlinkVersion version = (FlinkVersion) flinkVersionArgs.get()[0];
                             return Stream.of(
+                                    Arguments.of(version, new IOException("Network timeout")),
                                     Arguments.of(
-                                            version,
-                                            new IOException("Network timeout"),
-                                            "Network timeout"),
+                                            version, new IllegalStateException("Job not running")),
                                     Arguments.of(
-                                            version,
-                                            new IllegalStateException("Job not running"),
-                                            "Job not running"),
-                                    Arguments.of(
-                                            version,
-                                            new RuntimeException("Service unavailable"),
-                                            "Service unavailable"),
-                                    Arguments.of(
-                                            version,
-                                            new Exception("Generic error"),
-                                            "Generic error"));
+                                            version, new RuntimeException("Service unavailable")),
+                                    Arguments.of(version, new Exception("Generic error")));
                         });
     }
 
@@ -897,6 +880,15 @@ private static void assertFailingJobStatus(
         assertNotNull(rs.reconciledStatus.getError());
     }
 
+    private static void assertFailingWithError(
+            TestingFlinkBlueGreenDeploymentController.BlueGreenReconciliationResult rs,
+            String... expectedErrorFragments) {
+        assertFailingJobStatus(rs);
+        for (String fragment : expectedErrorFragments) {
+            assertTrue(rs.reconciledStatus.getError().contains(fragment));
+        }
+    }
+
     private void assertFinalized(
             long minReconciliationTs,
             TestingFlinkBlueGreenDeploymentController.BlueGreenReconciliationResult rs,
@@ -1102,6 +1094,10 @@ private void simulateChangeInSpec(
         kubernetesClient.resource(blueGreenDeployment).createOrReplace();
     }
 
+    private void simulateSpecChange(FlinkBlueGreenDeployment deployment, String customValue) {
+        simulateChangeInSpec(deployment, customValue, 0, null);
+    }
+
     /*
     Convenience function to reconcile and get the frequently used `BlueGreenReconciliationResult`
      */