diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkStateSnapshotController.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkStateSnapshotController.java index 8c36dc0efd..57626c7865 100644 --- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkStateSnapshotController.java +++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkStateSnapshotController.java @@ -89,6 +89,12 @@ public UpdateControl reconcile( @Override public DeleteControl cleanup( FlinkStateSnapshot flinkStateSnapshot, Context 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); diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/bluegreen/BlueGreenDeploymentService.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/bluegreen/BlueGreenDeploymentService.java index 9543fecabd..6274881df2 100644 --- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/bluegreen/BlueGreenDeploymentService.java +++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/bluegreen/BlueGreenDeploymentService.java @@ -162,10 +162,18 @@ public UpdateControl 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 checkAndInitiateDeployment( "Savepoint redeploy triggered for '{}', using initialSavepointPath: {}", context.getBgDeployment().getMetadata().getName(), Objects.toString(jobSpec.getInitialSavepointPath(), "")); - 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 abortDeployment( FlinkBlueGreenDeploymentState previousState = getPreviousState(nextState, context.getDeployments()); context.getDeploymentStatus().setBlueGreenState(previousState); + context.getDeploymentStatus().setSavepointTriggerId(null); var error = String.format( diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkBlueGreenDeploymentControllerTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkBlueGreenDeploymentControllerTest.java index 528b61727d..e657bb7227 100644 --- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkBlueGreenDeploymentControllerTest.java +++ b/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"; diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkStateSnapshotControllerTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkStateSnapshotControllerTest.java index cfe6947343..68b979625a 100644 --- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkStateSnapshotControllerTest.java +++ b/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); }