From 452a52e220fec3c42c6988a6febe4f23f6377ee5 Mon Sep 17 00:00:00 2001 From: Farooq Qaiser Date: Mon, 13 Jul 2026 14:57:09 -0400 Subject: [PATCH] [FLINK-40092] BlueGreen: finalize forward when a prior finalize didn't complete --- .../bluegreen/BlueGreenDeploymentService.java | 55 ++++++-- ...linkBlueGreenDeploymentControllerTest.java | 128 ++++++++++++++++++ 2 files changed, 173 insertions(+), 10 deletions(-) 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 931bf36e5c..2a01af4456 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 @@ -476,6 +476,20 @@ public UpdateControl 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 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 finalizeSuspendedDeployment( BlueGreenContext context, FlinkBlueGreenDeploymentState nextState) { @@ -581,7 +600,8 @@ private UpdateControl shouldWeDelete( BlueGreenContext context, FlinkDeployment currentDeployment, FlinkDeployment nextDeployment, - FlinkBlueGreenDeploymentState nextState) { + FlinkBlueGreenDeploymentState nextState, + CutoverSchedule schedule) { var deploymentStatus = context.getDeploymentStatus(); @@ -590,22 +610,18 @@ private UpdateControl 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(); + } + } } 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 b3b348ded9..d91e6be4ba 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 @@ -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 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);