[FLINK-40092] BlueGreen: finalize forward when a prior finalize didn't complete

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

@@ -476,6 +476,20 @@ public UpdateControl<FlinkBlueGreenDeployment> monitorTransition(
         TransitionState transitionState =
                 determineTransitionState(context, currentBlueGreenDeploymentType);
 
+        CutoverSchedule schedule =
+                new CutoverSchedule(
+                        instantStrToMillis(
+                                context.getDeploymentStatus().getDeploymentReadyTimestamp()),
+                        BlueGreenUtils.getDeploymentDeletionDelay(context));
+
+        // If an earlier BlueGreenDeployment deleted the previous deployment but failed before
+        // finalizing the transition, there is nothing to roll back to so we must finalize forward
+        // instead of aborting to a deleted target.
+        if (isGoneOrTerminating(transitionState.currentDeployment)
+                && schedule.isPastDeletionDeadline()) {
+            return finalizeBlueGreenDeployment(context, transitionState.nextState);
+        }
+
         if (isChildSuspended(transitionState.nextDeployment)) {
             if (transitionState.nextDeployment.getStatus().getLifecycleState()
                     == ResourceLifecycleState.SUSPENDED) {
@@ -491,13 +505,18 @@ public UpdateControl<FlinkBlueGreenDeployment> monitorTransition(
                     context,
                     transitionState.currentDeployment,
                     transitionState.nextDeployment,
-                    transitionState.nextState);
+                    transitionState.nextState,
+                    schedule);
         } else {
             return shouldWeAbort(
                     context, transitionState.nextDeployment, transitionState.nextState);
         }
     }
 
+    private static boolean isGoneOrTerminating(FlinkDeployment deployment) {
+        return deployment == null || deployment.getMetadata().getDeletionTimestamp() != null;
+    }
+
     private UpdateControl<FlinkBlueGreenDeployment> finalizeSuspendedDeployment(
             BlueGreenContext context, FlinkBlueGreenDeploymentState nextState) {
 
@@ -581,7 +600,8 @@ private UpdateControl<FlinkBlueGreenDeployment> shouldWeDelete(
             BlueGreenContext context,
             FlinkDeployment currentDeployment,
             FlinkDeployment nextDeployment,
-            FlinkBlueGreenDeploymentState nextState) {
+            FlinkBlueGreenDeploymentState nextState,
+            CutoverSchedule schedule) {
 
         var deploymentStatus = context.getDeploymentStatus();
 
@@ -590,22 +610,18 @@ private UpdateControl<FlinkBlueGreenDeployment> shouldWeDelete(
             return finalizeBlueGreenDeployment(context, nextState);
         }
 
-        long deploymentDeletionDelayMs = BlueGreenUtils.getDeploymentDeletionDelay(context);
-        long deploymentReadyTimestamp =
-                instantStrToMillis(deploymentStatus.getDeploymentReadyTimestamp());
-
-        if (deploymentReadyTimestamp == 0) {
+        if (!schedule.readinessRecorded()) {
             LOG.info(
                     "FlinkDeployment '{}' marked ready, rescheduling reconciliation in {} seconds.",
                     nextDeployment.getMetadata().getName(),
-                    deploymentDeletionDelayMs / 1000);
+                    schedule.getDeletionDelayMs() / 1000);
 
             deploymentStatus.setDeploymentReadyTimestamp(Instant.now().toString());
             return patchStatusUpdateControl(context, null, null, null)
-                    .rescheduleAfter(deploymentDeletionDelayMs);
+                    .rescheduleAfter(schedule.getDeletionDelayMs());
         }
 
-        long deletionTimestamp = deploymentReadyTimestamp + deploymentDeletionDelayMs;
+        long deletionTimestamp = schedule.deletionDeadlineMs();
 
         if (deletionTimestamp < System.currentTimeMillis()) {
             return deleteDeployment(currentDeployment, context);
@@ -861,4 +877,23 @@ private static class TransitionState {
         final FlinkDeployment nextDeployment;
         final FlinkBlueGreenDeploymentState nextState;
     }
+
+    @Getter
+    @AllArgsConstructor
+    private static class CutoverSchedule {
+        final long deploymentReadyTimestampMs;
+        final long deletionDelayMs;
+
+        boolean readinessRecorded() {
+            return deploymentReadyTimestampMs != 0;
+        }
+
+        long deletionDeadlineMs() {
+            return deploymentReadyTimestampMs + deletionDelayMs;
+        }
+
+        boolean isPastDeletionDeadline() {
+            return readinessRecorded() && deletionDeadlineMs() < System.currentTimeMillis();
+        }
+    }
 }

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

@@ -44,6 +44,7 @@
 import org.apache.flink.kubernetes.operator.utils.IngressUtils;
 import org.apache.flink.runtime.jobgraph.SavepointConfigOptions;
 
+import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.fabric8.kubernetes.api.model.ObjectMetaBuilder;
 import io.fabric8.kubernetes.api.model.networking.v1.Ingress;
 import io.fabric8.kubernetes.client.KubernetesClient;
@@ -57,6 +58,7 @@
 import org.junit.jupiter.params.provider.MethodSource;
 
 import java.io.IOException;
+import java.time.Duration;
 import java.time.Instant;
 import java.util.HashMap;
 import java.util.List;
@@ -71,6 +73,7 @@
 import static org.apache.flink.kubernetes.operator.api.utils.BaseTestUtils.TEST_DEPLOYMENT_NAME;
 import static org.apache.flink.kubernetes.operator.api.utils.BaseTestUtils.TEST_NAMESPACE;
 import static org.apache.flink.kubernetes.operator.utils.bluegreen.BlueGreenUtils.instantStrToMillis;
+import static org.awaitility.Awaitility.await;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotEquals;
@@ -558,6 +561,111 @@ public void verifyFailureDuringTransition(FlinkVersion flinkVersion) throws Exce
         testTransitionToGreen(rs, customValue, null);
     }
 
+    @ParameterizedTest
+    @MethodSource("org.apache.flink.kubernetes.operator.TestUtils#flinkVersions")
+    public void verifyGreenFinalizesForwardWhenBlueDeletedButTransitionNotFinalized(
+            FlinkVersion flinkVersion) throws Exception {
+        var blueGreenDeployment =
+                buildSessionCluster(
+                        TEST_DEPLOYMENT_NAME,
+                        TEST_NAMESPACE,
+                        flinkVersion,
+                        null,
+                        UpgradeMode.STATELESS);
+        // Tiny abort grace so the abort deadline is already expired by the time Blue is gone.
+        var configuration = blueGreenDeployment.getSpec().getConfiguration();
+        configuration.put(ABORT_GRACE_PERIOD.key(), "1");
+        configuration.put(DEPLOYMENT_DELETION_DELAY.key(), "0");
+
+        blueGreenDeployment
+                .getSpec()
+                .setIngress(
+                        IngressSpec.builder()
+                                .template("{{name}}.{{namespace}}.example.com")
+                                .className("nginx")
+                                .build());
+
+        var rs = executeBasicDeployment(flinkVersion, blueGreenDeployment, false, null);
+        assertIngressPointsToService(BLUE_CLUSTER_ID + REST_SVC_NAME_SUFFIX);
+
+        // Trigger the transition to Green
+        simulateChangeInSpec(rs.deployment, UUID.randomUUID().toString(), 0, null);
+        rs = reconcile(rs.deployment);
+        assertEquals(
+                FlinkBlueGreenDeploymentState.TRANSITIONING_TO_GREEN,
+                rs.reconciledStatus.getBlueGreenState());
+        assertEquals(2, getFlinkDeployments().size());
+
+        // Green becomes ready -> marks the ready timestamp and schedules deletion of Blue. Blue is
+        // not deleted in this reconcile (deletion happens on the next one), so we stay in
+        // TRANSITIONING_TO_GREEN. This is the last parent status that was persisted before the
+        // (simulated) crash.
+        simulateSuccessfulJobStart(getFlinkDeploymentByName(GREEN_CLUSTER_ID));
+        rs = reconcile(rs.deployment);
+        assertEquals(
+                FlinkBlueGreenDeploymentState.TRANSITIONING_TO_GREEN,
+                rs.reconciledStatus.getBlueGreenState());
+        assertTrue(
+                instantStrToMillis(rs.reconciledStatus.getDeploymentReadyTimestamp()) > 0,
+                "Ready timestamp must be set once Green is ready (cutover reached)");
+        var preFinalize = rs.deployment;
+        // Ingress still points to Blue here (cutover hasn't happened). Capture it: the loop below
+        // runs the real finalize, which prematurely flips the ingress to Green. Restoring this
+        // pre-cutover ingress models the ingress write being part of the same lost finalize, so the
+        // recovery reconcile is what actually performs the Blue -> Green cutover.
+        var preFinalizeIngress = captureManagedIngress();
+
+        // Let the controller actually delete Blue at cutover via its real delete path. We keep
+        // reconciling from preFinalize and wait for Blue to be gone. The ACTIVE_GREEN status this
+        // finalize would have persisted is intentionally discarded below, modelling a crash after
+        // the delete was accepted but before the finalizing status patch persisted.
+        await().atMost(Duration.ofSeconds(10))
+                .until(
+                        () -> {
+                            reconcile(preFinalize);
+                            return getFlinkDeployments().size() == 1;
+                        });
+
+        // Restore the pre-cutover (Blue) ingress so the recovery reconcile must perform the flip.
+        restoreIngress(preFinalizeIngress);
+        assertIngressPointsToService(BLUE_CLUSTER_ID + REST_SVC_NAME_SUFFIX);
+
+        // Crash gap: recovery resumes from the last persisted parent status (preFinalize: still
+        // TRANSITIONING_TO_GREEN with the ready timestamp set) while Blue is already gone. Green is
+        // momentarily not-ready, the exact condition that previously caused a rollback to the
+        // already-deleted Blue.
+        simulateJobFailure(getFlinkDeploymentByName(GREEN_CLUSTER_ID));
+        rs = reconcile(preFinalize);
+        assertEquals(1, getFlinkDeployments().size(), "Only Green must remain after cutover");
+        assertEquals(
+                FlinkBlueGreenDeploymentState.ACTIVE_GREEN,
+                rs.reconciledStatus.getBlueGreenState(),
+                "Transition must commit forward to ACTIVE_GREEN once Blue is deleted");
+        assertEquals(
+                0,
+                instantStrToMillis(rs.reconciledStatus.getAbortTimestamp()),
+                "Abort timer must be cleared once finalized");
+        assertNotEquals(
+                JobState.SUSPENDED,
+                getFlinkDeploymentByName(GREEN_CLUSTER_ID).getSpec().getJob().getState(),
+                "Green must not be suspended by a not-ready blip after Blue is deleted");
+        // Cutover contract: ingress must point to Green.
+        assertIngressPointsToService(GREEN_CLUSTER_ID + REST_SVC_NAME_SUFFIX);
+
+        // Post-cutover restart: Green briefly goes not-ready again
+        simulateJobFailure(getFlinkDeploymentByName(GREEN_CLUSTER_ID));
+        rs = reconcile(rs.deployment);
+        assertEquals(1, getFlinkDeployments().size(), "Green must still exist");
+        assertNotEquals(
+                JobState.SUSPENDED,
+                getFlinkDeploymentByName(GREEN_CLUSTER_ID).getSpec().getJob().getState(),
+                "Green must not be suspended by a post-cutover not-ready blip");
+        assertEquals(
+                FlinkBlueGreenDeploymentState.ACTIVE_GREEN,
+                rs.reconciledStatus.getBlueGreenState(),
+                "Must remain ACTIVE_GREEN; must not roll back after cutover");
+    }
+
     private static String getFlinkConfigurationValue(
             FlinkDeploymentSpec flinkDeploymentSpec, String propertyName) {
         return flinkDeploymentSpec.getFlinkConfiguration().get(propertyName).asText();
@@ -1474,6 +1582,13 @@ private List<FlinkDeployment> getFlinkDeployments() {
                 .getItems();
     }
 
+    private FlinkDeployment getFlinkDeploymentByName(String name) {
+        return getFlinkDeployments().stream()
+                .filter(d -> name.equals(d.getMetadata().getName()))
+                .findFirst()
+                .orElseThrow(() -> new AssertionError("FlinkDeployment '" + name + "' not found"));
+    }
+
     private static FlinkBlueGreenDeployment buildSessionCluster(
             String name,
             String namespace,
@@ -1551,6 +1666,19 @@ private static FlinkBlueGreenDeploymentSpec getTestFlinkDeploymentSpec(FlinkVers
 
     // ==================== Ingress Helper Methods ====================
 
+    private HasMetadata captureManagedIngress() {
+        return IngressUtils.ingressInNetworkingV1(kubernetesClient)
+                ? getIngressV1(TEST_DEPLOYMENT_NAME, TEST_NAMESPACE)
+                : getIngressV1beta1(TEST_DEPLOYMENT_NAME, TEST_NAMESPACE);
+    }
+
+    private void restoreIngress(HasMetadata ingress) {
+        // Clear the resourceVersion so the replace isn't rejected by optimistic locking (the live
+        // object's version changed when the loop's finalize flipped it).
+        ingress.getMetadata().setResourceVersion(null);
+        kubernetesClient.resource(ingress).update();
+    }
+
     private void assertIngressPointsToService(String expectedServiceName) {
         if (IngressUtils.ingressInNetworkingV1(kubernetesClient)) {
             Ingress ingress = getIngressV1(TEST_DEPLOYMENT_NAME, TEST_NAMESPACE);