If a spec change comes in mid transition, we apply it right away.

Author: schongloo

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

@@ -29,7 +29,6 @@
 import org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions;
 import org.apache.flink.kubernetes.operator.controller.FlinkBlueGreenDeployments;
 import org.apache.flink.kubernetes.operator.controller.FlinkResourceContext;
-import org.apache.flink.kubernetes.operator.utils.bluegreen.BlueGreenSpecUtils;
 import org.apache.flink.kubernetes.operator.utils.bluegreen.BlueGreenUtils;
 import org.apache.flink.util.Preconditions;
 
@@ -49,13 +48,17 @@
 import static org.apache.flink.kubernetes.operator.controller.bluegreen.BlueGreenKubernetesService.updateFlinkDeployment;
 import static org.apache.flink.kubernetes.operator.utils.bluegreen.BlueGreenSpecUtils.fetchSavepointInfo;
 import static org.apache.flink.kubernetes.operator.utils.bluegreen.BlueGreenSpecUtils.getLastCheckpoint;
+import static org.apache.flink.kubernetes.operator.utils.bluegreen.BlueGreenSpecUtils.getSpecDiff;
+import static org.apache.flink.kubernetes.operator.utils.bluegreen.BlueGreenSpecUtils.hasSpecChanged;
 import static org.apache.flink.kubernetes.operator.utils.bluegreen.BlueGreenSpecUtils.isSavepointRequired;
 import static org.apache.flink.kubernetes.operator.utils.bluegreen.BlueGreenSpecUtils.lookForCheckpoint;
 import static org.apache.flink.kubernetes.operator.utils.bluegreen.BlueGreenSpecUtils.prepareFlinkDeployment;
+import static org.apache.flink.kubernetes.operator.utils.bluegreen.BlueGreenSpecUtils.setLastReconciledSpec;
 import static org.apache.flink.kubernetes.operator.utils.bluegreen.BlueGreenSpecUtils.triggerSavepoint;
 import static org.apache.flink.kubernetes.operator.utils.bluegreen.BlueGreenUtils.getReconciliationReschedInterval;
 import static org.apache.flink.kubernetes.operator.utils.bluegreen.BlueGreenUtils.instantStrToMillis;
 import static org.apache.flink.kubernetes.operator.utils.bluegreen.BlueGreenUtils.millisToInstantStr;
+import static org.apache.flink.kubernetes.operator.utils.bluegreen.BlueGreenUtils.setAbortTimestamp;
 
 /** Consolidated service for all Blue/Green deployment operations. */
 public class BlueGreenDeploymentService {
@@ -108,7 +111,7 @@ public UpdateControl<FlinkBlueGreenDeployment> initiateDeployment(
      */
     public UpdateControl<FlinkBlueGreenDeployment> checkAndInitiateDeployment(
             BlueGreenContext context, DeploymentType currentDeploymentType) {
-        BlueGreenDiffType specDiff = BlueGreenSpecUtils.getSpecDiff(context);
+        BlueGreenDiffType specDiff = getSpecDiff(context);
 
         if (specDiff != BlueGreenDiffType.IGNORE) {
             FlinkDeployment currentFlinkDeployment =
@@ -124,15 +127,15 @@ public UpdateControl<FlinkBlueGreenDeployment> checkAndInitiateDeployment(
                                 .rescheduleAfter(getReconciliationReschedInterval(context));
                     }
 
-                    BlueGreenSpecUtils.setLastReconciledSpec(context);
+                    setLastReconciledSpec(context);
                     return startTransition(context, currentDeploymentType, currentFlinkDeployment);
                 } else {
-                    BlueGreenSpecUtils.setLastReconciledSpec(context);
+                    setLastReconciledSpec(context);
                     return patchFlinkDeployment(context, currentDeploymentType, specDiff);
                 }
             } else {
                 if (context.getDeploymentStatus().getJobStatus().getState() != JobStatus.FAILING) {
-                    BlueGreenSpecUtils.setLastReconciledSpec(context);
+                    setLastReconciledSpec(context);
                     return patchStatusUpdateControl(context, null, JobStatus.FAILING);
                 }
             }
@@ -292,10 +295,10 @@ private FlinkBlueGreenDeploymentState calculateSavepointingState(DeploymentType
     public UpdateControl<FlinkBlueGreenDeployment> monitorTransition(
             BlueGreenContext context, DeploymentType currentDeploymentType) {
 
-        handleSpecChangesDuringTransition(context);
-
         TransitionState transitionState = determineTransitionState(context, currentDeploymentType);
 
+        handleSpecChangesDuringTransition(context, transitionState);
+
         if (isFlinkDeploymentReady(transitionState.nextDeployment)) {
             return shouldWeDelete(
                     context,
@@ -308,11 +311,27 @@ public UpdateControl<FlinkBlueGreenDeployment> monitorTransition(
         }
     }
 
-    private void handleSpecChangesDuringTransition(BlueGreenContext context) {
-        if (BlueGreenSpecUtils.hasSpecChanged(context)) {
-            BlueGreenSpecUtils.revertToLastSpec(context);
-            LOG.warn(
-                    "Blue/Green Spec change detected during transition, ignored and reverted to the last reconciled spec");
+    private void handleSpecChangesDuringTransition(
+            BlueGreenContext context, TransitionState transitionState) {
+        if (hasSpecChanged(context)) {
+            BlueGreenDiffType diffType = getSpecDiff(context);
+
+            FlinkDeployment incomingFlinkDeployment = transitionState.nextDeployment;
+
+            if (diffType != BlueGreenDiffType.IGNORE) {
+                // Apply the new spec changes to the currently active deployment
+                incomingFlinkDeployment.setSpec(
+                        context.getBgDeployment().getSpec().getTemplate().getSpec());
+                // Resetting the abort grace period
+                setAbortTimestamp(context);
+                // Ack'ing the spec change
+                setLastReconciledSpec(context);
+                updateFlinkDeployment(incomingFlinkDeployment, context);
+
+                LOG.info(
+                        "Blue/Green Spec change detected during transition, patching incoming '{}' deployment with new changes",
+                        incomingFlinkDeployment.getMetadata().getName());
+            }
         }
     }
 

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/diff/FlinkBlueGreenDeploymentSpecDiff.java

@@ -65,19 +65,15 @@ public BlueGreenDiffType compare() {
 
         BlueGreenDiffType childDiffType = getChildSpecDiffType();
 
-        // If nested spec has SCALE or UPGRADE differences, return TRANSITION regardless
+        // If nested spec has SCALE or UPGRADE differences, return TRANSITION
         if (childDiffType == BlueGreenDiffType.TRANSITION) {
             return BlueGreenDiffType.TRANSITION;
         }
 
-        // Determine result based on where differences are found
-        boolean hasChildDiff = childDiffType != BlueGreenDiffType.IGNORE;
-
-        if (hasChildDiff) {
+        if (childDiffType != BlueGreenDiffType.IGNORE) {
             // Child spec changes take precedence, return the child diff type
             return BlueGreenDiffType.PATCH_CHILD;
-        }
-        else {
+        } else {
             return BlueGreenDiffType.IGNORE;
         }
     }

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

@@ -68,6 +68,7 @@
 import static org.apache.flink.kubernetes.operator.utils.bluegreen.BlueGreenUtils.instantStrToMillis;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -344,12 +345,17 @@ public void verifySpecChangeDuringTransition(FlinkVersion flinkVersion) throws E
 
         // Simulate a spec change before the transition is complete
         simulateChangeInSpec(rs.deployment, "MODIFIED_VALUE", 0, null);
+        var moddedSpec = rs.deployment.getSpec();
         rs = reconcile(rs.deployment);
 
-        // The spec should have been reverted
-        assertEquals(
+        // The spec change should have been preserved
+        assertNotEquals(
                 SpecUtils.writeSpecAsJSON(originalSpec, "spec"),
                 SpecUtils.writeSpecAsJSON(rs.deployment.getSpec(), "spec"));
+
+        assertEquals(
+                SpecUtils.writeSpecAsJSON(moddedSpec, "spec"),
+                rs.deployment.getStatus().getLastReconciledSpec(), "spec");
     }
 
     @ParameterizedTest