[FLINK-40092] Finalize BlueGreen transition after deleting previous deployment

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

@@ -630,7 +630,7 @@ private UpdateControl<FlinkBlueGreenDeployment> shouldWeDelete(
         long deletionTimestamp = deploymentReadyTimestamp + deploymentDeletionDelayMs;
 
         if (deletionTimestamp < System.currentTimeMillis()) {
-            return deleteDeployment(currentDeployment, context);
+            return deleteDeployment(currentDeployment, context, nextState);
         } else {
             return waitBeforeDeleting(currentDeployment, deletionTimestamp);
         }
@@ -649,17 +649,20 @@ private UpdateControl<FlinkBlueGreenDeployment> waitBeforeDeleting(
     }
 
     private UpdateControl<FlinkBlueGreenDeployment> deleteDeployment(
-            FlinkDeployment currentDeployment, BlueGreenContext context) {
+            FlinkDeployment currentDeployment,
+            BlueGreenContext context,
+            FlinkBlueGreenDeploymentState nextState) {
 
         boolean deleted = deleteFlinkDeployment(currentDeployment, context);
 
         if (!deleted) {
             LOG.info("FlinkDeployment '{}' not deleted, will retry", currentDeployment);
+            return UpdateControl.<FlinkBlueGreenDeployment>noUpdate()
+                    .rescheduleAfter(RETRY_DELAY_MS);
         } else {
             LOG.info("FlinkDeployment '{}' deleted!", currentDeployment);
+            return finalizeBlueGreenDeployment(context, nextState);
         }
-
-        return UpdateControl.<FlinkBlueGreenDeployment>noUpdate().rescheduleAfter(RETRY_DELAY_MS);
     }
 
     // ==================== Abort and Retry Methods ====================

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

@@ -57,11 +57,13 @@
 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;
 import java.util.Map;
 import java.util.UUID;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.stream.Stream;
 
 import static org.apache.flink.kubernetes.operator.api.spec.FlinkBlueGreenDeploymentConfigOptions.ABORT_GRACE_PERIOD;
@@ -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;
@@ -564,6 +567,82 @@ public void verifyFailureDuringTransition(FlinkVersion flinkVersion) throws Exce
         testTransitionToGreen(rs, customValue, null);
     }
 
+    @ParameterizedTest
+    @MethodSource("org.apache.flink.kubernetes.operator.TestUtils#flinkVersions")
+    public void verifyGreenNotAbortedWhenNotReadyAfterBlueDeleted(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 deleted.
+        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.
+        simulateSuccessfulJobStart(getFlinkDeploymentByName(GREEN_CLUSTER_ID));
+
+        // Reconcile until the (zero) deletion delay elapses: Blue is deleted at cutover and the
+        // transition finalizes forward to ACTIVE_GREEN.
+        var latestRs = new AtomicReference<>(rs);
+        await().atMost(Duration.ofSeconds(10))
+                .until(
+                        () -> {
+                            latestRs.set(reconcile(latestRs.get().deployment));
+                            return latestRs.get().reconciledStatus.getBlueGreenState()
+                                    == FlinkBlueGreenDeploymentState.ACTIVE_GREEN;
+                        });
+        rs = latestRs.get();
+
+        assertEquals(1, getFlinkDeployments().size(), "Blue should be deleted at cutover");
+        assertEquals(
+                FlinkBlueGreenDeploymentState.ACTIVE_GREEN,
+                rs.reconciledStatus.getBlueGreenState(),
+                "Transition must commit forward to ACTIVE_GREEN when Blue is deleted");
+        assertEquals(
+                0,
+                instantStrToMillis(rs.reconciledStatus.getAbortTimestamp()),
+                "Abort timer must be cleared once finalized");
+        assertIngressPointsToService(GREEN_CLUSTER_ID + REST_SVC_NAME_SUFFIX);
+
+        // Post-cutover restart: Green briefly goes not-ready
+        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();
@@ -1487,6 +1566,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,