Re-adding test inadvertently removed.

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

@@ -52,6 +52,7 @@
 import org.junit.jupiter.params.provider.Arguments;
 import org.junit.jupiter.params.provider.MethodSource;
 
+import java.io.IOException;
 import java.time.Instant;
 import java.util.HashMap;
 import java.util.List;
@@ -473,6 +474,40 @@ public void verifySavepointFailureRecovery(FlinkVersion flinkVersion) throws Exc
         testTransitionToGreen(rs, customValue, "savepoint_1");
     }
 
+    @ParameterizedTest
+    @MethodSource("savepointExceptionProvider")
+    public void verifySavepointFailureWithDifferentExceptionTypes(
+            FlinkVersion flinkVersion, Exception savepointException) throws Exception {
+
+        String expectedErrorFragment = savepointException.getMessage();
+
+        var blueGreenDeployment =
+                buildSessionCluster(
+                        TEST_DEPLOYMENT_NAME,
+                        TEST_NAMESPACE,
+                        flinkVersion,
+                        null,
+                        UpgradeMode.SAVEPOINT);
+        var rs = executeBasicDeployment(flinkVersion, blueGreenDeployment, false, null);
+
+        flinkService.setSavepointTriggerException(savepointException);
+        simulateChangeInSpec(rs.deployment, UUID.randomUUID().toString(), 0, null);
+
+        rs = reconcile(rs.deployment);
+
+        assertFailingJobStatus(rs);
+        assertTrue(rs.reconciledStatus.getError().contains("Could not trigger Savepoint"));
+        assertTrue(rs.reconciledStatus.getError().contains(expectedErrorFragment));
+
+        // Should remain in ACTIVE_BLUE state (no transition started)
+        assertEquals(
+                FlinkBlueGreenDeploymentState.ACTIVE_BLUE, rs.reconciledStatus.getBlueGreenState());
+
+        // Verify only Blue deployment exists (Green was never created)
+        var flinkDeployments = getFlinkDeployments();
+        assertEquals(1, flinkDeployments.size());
+    }
+
     @ParameterizedTest
     @MethodSource("org.apache.flink.kubernetes.operator.TestUtils#flinkVersions")
     public void verifySavepointFetchFailureRecovery(FlinkVersion flinkVersion) throws Exception {
@@ -578,6 +613,21 @@ static Stream<Arguments> savepointErrorProvider() {
                         });
     }
 
+    static Stream<Arguments> savepointExceptionProvider() {
+        return TestUtils.flinkVersions()
+                .flatMap(
+                        flinkVersionArgs -> {
+                            FlinkVersion version = (FlinkVersion) flinkVersionArgs.get()[0];
+                            return Stream.of(
+                                    Arguments.of(version, new IOException("Network timeout")),
+                                    Arguments.of(
+                                            version, new IllegalStateException("Job not running")),
+                                    Arguments.of(
+                                            version, new RuntimeException("Service unavailable")),
+                                    Arguments.of(version, new Exception("Generic error")));
+                        });
+    }
+
     static Stream<Arguments> patchScenarioProvider() {
         // Extract FlinkVersions from TestUtils and combine with PatchTypes
         return TestUtils.flinkVersions()