Stateless Flink State Snapshot

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

@@ -89,6 +89,12 @@ public UpdateControl<FlinkStateSnapshot> reconcile(
     @Override
     public DeleteControl cleanup(
             FlinkStateSnapshot flinkStateSnapshot, Context<FlinkStateSnapshot> josdkContext) {
+        if (flinkStateSnapshot.getStatus() == null) {
+            LOG.info(
+                    "Snapshot {} has no status, was never reconciled. Removing finalizer.",
+                    flinkStateSnapshot.getMetadata().getName());
+            return DeleteControl.defaultDelete();
+        }
         var ctx = ctxFactory.getFlinkStateSnapshotContext(flinkStateSnapshot, josdkContext);
         try {
             metricManager.onRemove(flinkStateSnapshot);

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";

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

@@ -704,6 +704,20 @@ public void testMetrics() {
         assertSnapshotMetrics(listener, TestUtils.TEST_NAMESPACE, Map.of(), Map.of());
     }
 
+    @Test
+    public void testCleanupWithNullStatus() {
+        var deployment = createDeployment();
+        context = TestUtils.createSnapshotContext(client, deployment);
+
+        var savepoint = createSavepoint(deployment);
+        savepoint.setStatus(null);
+        assertDeleteControl(controller.cleanup(savepoint, context), true, null);
+
+        var checkpoint = createCheckpoint(deployment, CheckpointType.FULL, 0);
+        checkpoint.setStatus(null);
+        assertDeleteControl(controller.cleanup(checkpoint, context), true, null);
+    }
+
     private FlinkStateSnapshot createSavepoint(FlinkDeployment deployment) {
         return createSavepoint(deployment, false, 7);
     }