[FLINK-38867] [Kubernetes Operator] Fix restartSavepointNonce to respect initialSavepointPath in BlueGreenDeployments

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/api/bluegreen/BlueGreenDiffType.java

@@ -26,5 +26,11 @@ public enum BlueGreenDiffType {
     TRANSITION,
 
     /** Changes that only affect the child FlinkDeploymentSpec. */
-    PATCH_CHILD
+    PATCH_CHILD,
+
+    /**
+     * Full redeploy from user-specified savepoint. Triggered when savepointRedeployNonce changes.
+     * Uses the initialSavepointPath from spec instead of taking a new savepoint.
+     */
+    SAVEPOINT_REDEPLOY
 }

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

@@ -42,6 +42,7 @@
 import org.slf4j.LoggerFactory;
 
 import java.time.Instant;
+import java.util.Objects;
 
 import static org.apache.flink.kubernetes.operator.controller.bluegreen.BlueGreenKubernetesService.deleteFlinkDeployment;
 import static org.apache.flink.kubernetes.operator.controller.bluegreen.BlueGreenKubernetesService.deployCluster;
@@ -143,6 +144,25 @@ public UpdateControl<FlinkBlueGreenDeployment> checkAndInitiateDeployment(
                         context.getDeploymentStatus().setSavepointTriggerId(null);
                         return markDeploymentFailing(context, error);
                     }
+
+                } else if (specDiff == BlueGreenDiffType.SAVEPOINT_REDEPLOY) {
+                    // Savepoint redeploy: skip taking a new savepoint, use initialSavepointPath
+                    var jobSpec =
+                            context.getBgDeployment().getSpec().getTemplate().getSpec().getJob();
+                    LOG.info(
+                            "Savepoint redeploy triggered for '{}', using initialSavepointPath: {}",
+                            context.getBgDeployment().getMetadata().getName(),
+                            Objects.toString(jobSpec.getInitialSavepointPath(), "<none>"));
+                    setLastReconciledSpec(context);
+                    try {
+                        return startSavepointRedeployTransition(
+                                context, currentBlueGreenDeploymentType);
+                    } catch (Exception e) {
+                        var error =
+                                "Could not start Savepoint Redeploy Transition. Details: "
+                                        + e.getMessage();
+                        return markDeploymentFailing(context, error);
+                    }
                 } else {
                     setLastReconciledSpec(context);
                     LOG.info(
@@ -170,6 +190,13 @@ public UpdateControl<FlinkBlueGreenDeployment> checkAndInitiateDeployment(
 
     private UpdateControl<FlinkBlueGreenDeployment> patchFlinkDeployment(
             BlueGreenContext context, BlueGreenDeploymentType blueGreenDeploymentTypeToPatch) {
+        return patchFlinkDeployment(context, blueGreenDeploymentTypeToPatch, true);
+    }
+
+    private UpdateControl<FlinkBlueGreenDeployment> patchFlinkDeployment(
+            BlueGreenContext context,
+            BlueGreenDeploymentType blueGreenDeploymentTypeToPatch,
+            boolean carryOverSavepointInPatch) {
 
         String childDeploymentName =
                 context.getBgDeployment().getMetadata().getName()
@@ -188,7 +215,10 @@ private UpdateControl<FlinkBlueGreenDeployment> patchFlinkDeployment(
         //  will it be used by this patching? otherwise this is unnecessary, keep lastSavepoint =
         // null.
         Savepoint lastSavepoint =
-                carryOverSavepoint(context, blueGreenDeploymentTypeToPatch, childDeploymentName);
+                carryOverSavepointInPatch
+                        ? carryOverSavepoint(
+                                context, blueGreenDeploymentTypeToPatch, childDeploymentName)
+                        : null;
 
         return initiateDeployment(
                 context,
@@ -246,6 +276,26 @@ private UpdateControl<FlinkBlueGreenDeployment> startTransition(
                 false);
     }
 
+    /**
+     * Starts a transition for savepoint redeploy scenario. Unlike normal transitions, this does not
+     * take a new savepoint - it uses the initialSavepointPath specified in the spec.
+     *
+     * @param context the transition context
+     * @param currentBlueGreenDeploymentType the current deployment type
+     * @return UpdateControl for the deployment
+     */
+    private UpdateControl<FlinkBlueGreenDeployment> startSavepointRedeployTransition(
+            BlueGreenContext context, BlueGreenDeploymentType currentBlueGreenDeploymentType) {
+        DeploymentTransition transition = calculateTransition(currentBlueGreenDeploymentType);
+
+        return initiateDeployment(
+                context,
+                transition.nextBlueGreenDeploymentType,
+                transition.nextState,
+                null, // Use initialSavepointPath from spec
+                false);
+    }
+
     private DeploymentTransition calculateTransition(BlueGreenDeploymentType currentType) {
         if (BlueGreenDeploymentType.BLUE == currentType) {
             return new DeploymentTransition(
@@ -411,7 +461,10 @@ private UpdateControl<FlinkBlueGreenDeployment> handleSpecChangesDuringTransitio
                         context.getDeploymentByType(oppositeDeploymentType)
                                 .getMetadata()
                                 .getName());
-                return patchFlinkDeployment(context, oppositeDeploymentType);
+                return patchFlinkDeployment(
+                        context,
+                        oppositeDeploymentType,
+                        diffType != BlueGreenDiffType.SAVEPOINT_REDEPLOY);
             }
         }
 

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/diff/FlinkBlueGreenDeploymentSpecDiff.java

@@ -59,24 +59,31 @@ public BlueGreenDiffType compare() {
         FlinkDeploymentSpec leftSpec = left.getTemplate().getSpec();
         FlinkDeploymentSpec rightSpec = right.getTemplate().getSpec();
 
-        // Used in Case 2 & 3: Delegate to ReflectiveDiffBuilder for nested spec comparison
+        // Used in Case 2, 3 & 4: Delegate to ReflectiveDiffBuilder for nested spec comparison
         // Calculate diffResult before comparison to apply in-place removal of ignored fields
         DiffResult<FlinkDeploymentSpec> diffResult =
                 new ReflectiveDiffBuilder<>(deploymentMode, leftSpec, rightSpec).build();
 
-        // Case 1: FlinkDeploymentSpecs are identical
+        // Extract the diff type from ReflectiveDiffBuilder result
+        DiffType diffType = diffResult.getType();
+
+        // Case 1: Check for savepoint redeploy first (savepointRedeployNonce changed)
+        // This takes precedence as it indicates user wants to redeploy from their
+        // initialSavepointPath
+        if (diffType == DiffType.SAVEPOINT_REDEPLOY) {
+            return BlueGreenDiffType.SAVEPOINT_REDEPLOY;
+        }
+
+        // Case 2: FlinkDeploymentSpecs are identical
         if (leftSpec.equals(rightSpec)) {
             return BlueGreenDiffType.IGNORE;
         }
 
-        // Extract the diff type from ReflectiveDiffBuilder result
-        DiffType diffType = diffResult.getType();
-
-        // Case 2: ReflectiveDiffBuilder returns IGNORE
+        // Case 3: ReflectiveDiffBuilder returns IGNORE
         if (diffType == DiffType.IGNORE) {
             return BlueGreenDiffType.PATCH_CHILD;
         } else {
-            // Case 3: ReflectiveDiffBuilder returns anything else map it to TRANSITION as well
+            // Case 4: ReflectiveDiffBuilder returns anything else map it to TRANSITION
             return BlueGreenDiffType.TRANSITION;
         }
     }

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/bluegreen/BlueGreenUtils.java

@@ -332,7 +332,12 @@ public static FlinkDeployment prepareFlinkDeployment(
                         "spec",
                         FlinkBlueGreenDeploymentSpec.class);
 
-        // The Blue/Green initialSavepointPath is only used for first-time deployments
+        // Determine which savepoint/checkpoint to restore from:
+        // 1. First-time deployments: use initialSavepointPath from spec (if set)
+        // 2. Normal transitions: use lastCheckpoint from previous deployment
+        // 3. Redeploy scenarios (lastCheckpoint is null): use initialSavepointPath from spec
+        //    - savepointRedeployNonce changed
+        //    - upgradeMode is STATELESS
         if (isFirstDeployment) {
             String initialSavepointPath =
                     spec.getTemplate().getSpec().getJob().getInitialSavepointPath();
@@ -346,6 +351,16 @@ public static FlinkDeployment prepareFlinkDeployment(
             String location = lastCheckpoint.getLocation().replace("file:", "");
             LOG.info("Using Blue/Green savepoint/checkpoint: " + location);
             spec.getTemplate().getSpec().getJob().setInitialSavepointPath(location);
+        } else {
+            String initialSavepointPath =
+                    spec.getTemplate().getSpec().getJob().getInitialSavepointPath();
+            if (initialSavepointPath != null && !initialSavepointPath.isEmpty()) {
+                LOG.info(
+                        "Using user-specified initialSavepointPath for redeploy: {}",
+                        initialSavepointPath);
+            } else {
+                LOG.info("Starting fresh with no savepoint restoration");
+            }
         }
 
         flinkDeployment.setSpec(spec.getTemplate().getSpec());

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

@@ -189,6 +189,68 @@ private TestingFlinkBlueGreenDeploymentController.BlueGreenReconciliationResult
         return rs;
     }
 
+    @ParameterizedTest
+    @MethodSource("org.apache.flink.kubernetes.operator.TestUtils#flinkVersions")
+    public void verifySavepointRedeployNonceTriggersTransitionWithInitialSavepointPath(
+            FlinkVersion flinkVersion) throws Exception {
+        // Start with SAVEPOINT upgrade mode so normally a savepoint would be taken
+        var blueGreenDeployment =
+                buildSessionCluster(
+                        TEST_DEPLOYMENT_NAME,
+                        TEST_NAMESPACE,
+                        flinkVersion,
+                        null,
+                        UpgradeMode.SAVEPOINT);
+        var rs = executeBasicDeployment(flinkVersion, blueGreenDeployment, false, null);
+
+        // Set initialSavepointPath and bump savepointRedeployNonce
+        String userSpecifiedSavepoint = "s3://bucket/my-specific-savepoint";
+        rs.deployment
+                .getSpec()
+                .getTemplate()
+                .getSpec()
+                .getJob()
+                .setInitialSavepointPath(userSpecifiedSavepoint);
+        rs.deployment.getSpec().getTemplate().getSpec().getJob().setSavepointRedeployNonce(12345L);
+        rs.deployment
+                .getSpec()
+                .getConfiguration()
+                .put(DEPLOYMENT_DELETION_DELAY.key(), String.valueOf(ALT_DELETION_DELAY_VALUE));
+        kubernetesClient.resource(rs.deployment).createOrReplace();
+
+        // Reconcile - should skip savepointing and go directly to transition
+        rs = reconcile(rs.deployment);
+
+        // Verify: Should be TRANSITIONING_TO_GREEN (not SAVEPOINTING_BLUE)
+        assertEquals(
+                FlinkBlueGreenDeploymentState.TRANSITIONING_TO_GREEN,
+                rs.reconciledStatus.getBlueGreenState());
+
+        // Verify: Green deployment should use the user-specified initialSavepointPath
+        var flinkDeployments = getFlinkDeployments();
+        assertEquals(2, flinkDeployments.size());
+        assertEquals(
+                userSpecifiedSavepoint,
+                flinkDeployments.get(1).getSpec().getJob().getInitialSavepointPath());
+
+        // Complete the transition
+        simulateSuccessfulJobStart(flinkDeployments.get(1));
+        rs = reconcile(rs.deployment);
+
+        // Wait for deletion delay
+        Thread.sleep(rs.updateControl.getScheduleDelay().get());
+        reconcile(rs.deployment);
+
+        // Verify Green is now active
+        flinkDeployments = getFlinkDeployments();
+        assertEquals(1, flinkDeployments.size());
+
+        rs = reconcile(rs.deployment);
+        assertEquals(
+                FlinkBlueGreenDeploymentState.ACTIVE_GREEN,
+                rs.reconciledStatus.getBlueGreenState());
+    }
+
     @ParameterizedTest
     @MethodSource("org.apache.flink.kubernetes.operator.TestUtils#flinkVersions")
     public void verifyFailureBeforeTransition(FlinkVersion flinkVersion) throws Exception {

flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/diff/FlinkBlueGreenDeploymentSpecDiffTest.java

@@ -201,13 +201,28 @@ public void testTransitionForNestedSpecDifference() {
         FlinkBlueGreenDeploymentSpec spec1 = createBasicSpec();
         FlinkBlueGreenDeploymentSpec spec2 = createBasicSpec();
 
-        // Change nested spec property - setSavepointRedeployNonce triggers TRANSITION
+        // Change nested spec property - setSavepointRedeployNonce now triggers SAVEPOINT_REDEPLOY
         spec2.getTemplate().getSpec().getJob().setSavepointRedeployNonce(12345L);
 
         FlinkBlueGreenDeploymentSpecDiff diff =
                 new FlinkBlueGreenDeploymentSpecDiff(DEPLOYMENT_MODE, spec1, spec2);
 
-        assertEquals(BlueGreenDiffType.TRANSITION, diff.compare());
+        assertEquals(BlueGreenDiffType.SAVEPOINT_REDEPLOY, diff.compare());
+    }
+
+    @Test
+    public void testSavepointRedeployForNonceChangeWithJarUpdate() {
+        FlinkBlueGreenDeploymentSpec spec1 = createBasicSpec();
+        FlinkBlueGreenDeploymentSpec spec2 = createBasicSpec();
+
+        // Nonce change with additional jarURI change - SAVEPOINT_REDEPLOY should still be detected
+        spec2.getTemplate().getSpec().getJob().setJarURI("local:///opt/flink/examples/other.jar");
+        spec2.getTemplate().getSpec().getJob().setSavepointRedeployNonce(12345L);
+
+        FlinkBlueGreenDeploymentSpecDiff diff =
+                new FlinkBlueGreenDeploymentSpecDiff(DEPLOYMENT_MODE, spec1, spec2);
+
+        assertEquals(BlueGreenDiffType.SAVEPOINT_REDEPLOY, diff.compare());
     }
 
     @Test
@@ -281,8 +296,8 @@ public void testTransitionForTopLevelAndNestedDifferences() {
         FlinkBlueGreenDeploymentSpec spec2 = createBasicSpec();
 
         // Change both top-level (configuration) and nested spec
-        // With new logic, only nested spec changes matter - setSavepointRedeployNonce triggers
-        // TRANSITION
+        // With new logic, only nested spec changes matter - setSavepointRedeployNonce now
+        // triggers SAVEPOINT_REDEPLOY
         Map<String, String> config = new HashMap<>();
         config.put("custom.config", "different-value");
         spec2.setConfiguration(config);
@@ -291,7 +306,39 @@ public void testTransitionForTopLevelAndNestedDifferences() {
         FlinkBlueGreenDeploymentSpecDiff diff =
                 new FlinkBlueGreenDeploymentSpecDiff(DEPLOYMENT_MODE, spec1, spec2);
 
-        assertEquals(BlueGreenDiffType.TRANSITION, diff.compare());
+        assertEquals(BlueGreenDiffType.SAVEPOINT_REDEPLOY, diff.compare());
+    }
+
+    @Test
+    public void testSavepointRedeployTakesPrecedenceOverScale() {
+        FlinkBlueGreenDeploymentSpec spec1 = createBasicSpec();
+        FlinkBlueGreenDeploymentSpec spec2 = createBasicSpec();
+
+        // Change both nonce (SAVEPOINT_REDEPLOY) AND parallelism (SCALE)
+        // SAVEPOINT_REDEPLOY should take precedence
+        spec2.getTemplate().getSpec().getJob().setSavepointRedeployNonce(12345L);
+        spec2.getTemplate().getSpec().getJob().setParallelism(10);
+
+        FlinkBlueGreenDeploymentSpecDiff diff =
+                new FlinkBlueGreenDeploymentSpecDiff(DEPLOYMENT_MODE, spec1, spec2);
+
+        assertEquals(BlueGreenDiffType.SAVEPOINT_REDEPLOY, diff.compare());
+    }
+
+    @Test
+    public void testSavepointRedeployTakesPrecedenceOverUpgrade() {
+        FlinkBlueGreenDeploymentSpec spec1 = createBasicSpec();
+        FlinkBlueGreenDeploymentSpec spec2 = createBasicSpec();
+
+        // Change both nonce (SAVEPOINT_REDEPLOY) AND Flink version (UPGRADE)
+        // SAVEPOINT_REDEPLOY should take precedence
+        spec2.getTemplate().getSpec().getJob().setSavepointRedeployNonce(12345L);
+        spec2.getTemplate().getSpec().setFlinkVersion(FlinkVersion.v1_17);
+
+        FlinkBlueGreenDeploymentSpecDiff diff =
+                new FlinkBlueGreenDeploymentSpecDiff(DEPLOYMENT_MODE, spec1, spec2);
+
+        assertEquals(BlueGreenDiffType.SAVEPOINT_REDEPLOY, diff.compare());
     }
 
     @Test