Move setLastReconciledSpec to after successful transition start

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

@@ -162,10 +162,18 @@ public UpdateControl<FlinkBlueGreenDeployment> checkAndInitiateDeployment(
                                 .rescheduleAfter(getReconciliationReschedInterval(context));
                     }
 
-                    setLastReconciledSpec(context);
                     try {
-                        return startTransition(
-                                context, currentBlueGreenDeploymentType, currentFlinkDeployment);
+                        var result =
+                                startTransition(
+                                        context,
+                                        currentBlueGreenDeploymentType,
+                                        currentFlinkDeployment);
+                        // Only stamp lastReconciledSpec after the transition
+                        // succeeds. If stamped before and the transition
+                        // fails/aborts, lastReconciledSpec drifts from the
+                        // active child's actual spec
+                        setLastReconciledSpec(context);
+                        return result;
                     } catch (Exception e) {
                         var error = "Could not start Transition. Details: " + e.getMessage();
                         context.getDeploymentStatus().setSavepointTriggerId(null);
@@ -180,10 +188,12 @@ public UpdateControl<FlinkBlueGreenDeployment> checkAndInitiateDeployment(
                             "Savepoint redeploy triggered for '{}', using initialSavepointPath: {}",
                             context.getBgDeployment().getMetadata().getName(),
                             Objects.toString(jobSpec.getInitialSavepointPath(), "<none>"));
-                    setLastReconciledSpec(context);
                     try {
-                        return startSavepointRedeployTransition(
-                                context, currentBlueGreenDeploymentType);
+                        var result =
+                                startSavepointRedeployTransition(
+                                        context, currentBlueGreenDeploymentType);
+                        setLastReconciledSpec(context);
+                        return result;
                     } catch (Exception e) {
                         var error =
                                 "Could not start Savepoint Redeploy Transition. Details: "
@@ -685,6 +695,7 @@ private UpdateControl<FlinkBlueGreenDeployment> abortDeployment(
         FlinkBlueGreenDeploymentState previousState =
                 getPreviousState(nextState, context.getDeployments());
         context.getDeploymentStatus().setBlueGreenState(previousState);
+        context.getDeploymentStatus().setSavepointTriggerId(null);
 
         var error =
                 String.format(

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

@@ -546,6 +546,9 @@ public void verifyFailureDuringTransition(FlinkVersion flinkVersion) throws Exce
                 ReconciliationState.UPGRADING,
                 flinkDeployments.get(1).getStatus().getReconciliationStatus().getState());
         assertTrue(instantStrToMillis(rs.reconciledStatus.getAbortTimestamp()) > 0);
+        // savepointTriggerId must be cleared on abort so the next transition
+        // triggers a fresh savepoint instead of reusing a stale triggerId
+        assertNull(rs.reconciledStatus.getSavepointTriggerId());
 
         // Simulate another change in the spec to trigger a redeployment
         customValue = UUID.randomUUID().toString();
@@ -767,6 +770,13 @@ public void verifySavepointFetchFailureRecovery(FlinkVersion flinkVersion) throw
         rs = reconcile(rs.deployment);
         assertFailingWithError(rs, "Could not start Transition", error);
 
+        // lastReconciledSpec must NOT contain the failed spec change — otherwise
+        // subsequent deploys see no diff and fall into PATCH_CHILD (in-place
+        // upgrade) instead of a proper B/G transition
+        assertFalse(
+                rs.reconciledStatus.getLastReconciledSpec().contains(customValue),
+                "lastReconciledSpec should not be stamped when transition fails");
+
         // Recovery: Clear the fetch error and try again with new spec change
         flinkService.clearSavepointFetchError();
         customValue = UUID.randomUUID().toString() + "_recovery";