Addressing PR comments. Adding unit test to cover for Spec changes mid-transition.

Author: schongloo

flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/bluegreen/DeploymentType.java

@@ -19,9 +19,15 @@
 
 import org.apache.flink.kubernetes.operator.api.FlinkDeployment;
 
-/** Enumeration of the two possible Flink Blue/Green deployment types. */
+/**
+ * Enumeration of the two possible Flink Blue/Green deployment types. Only one of each type will be
+ * present at all times for a particular job.
+ */
 public enum DeploymentType {
+    /** Identifier for the first or "Blue" deployment type. */
     BLUE,
+
+    /** Identifier for the second or "Green" deployment type. */
     GREEN;
 
     public static DeploymentType fromDeployment(FlinkDeployment flinkDeployment) {

flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/status/FlinkBlueGreenDeploymentState.java

@@ -19,8 +19,15 @@
 
 /** Enumeration of the possible states of the blue/green transition. */
 public enum FlinkBlueGreenDeploymentState {
+    /** Identifies the system is running normally with a "Blue" deployment type. */
     ACTIVE_BLUE,
+
+    /** Identifies the system is running normally with a "Green" deployment type. */
     ACTIVE_GREEN,
+
+    /** Identifies the system is transitioning from "Green" to "Blue". */
     TRANSITIONING_TO_BLUE,
+
+    /** Identifies the system is transitioning from "Blue" to "Green". */
     TRANSITIONING_TO_GREEN,
 }

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

@@ -101,8 +101,7 @@ public UpdateControl<FlinkBlueGreenDeployment> reconcile(
 
         if (deploymentStatus == null) {
             deploymentStatus = new FlinkBlueGreenDeploymentStatus();
-            deploymentStatus.setLastReconciledSpec(
-                    SpecUtils.serializeObject(flinkBlueGreenDeployment.getSpec(), "spec"));
+            setLastReconciledSpec(flinkBlueGreenDeployment, deploymentStatus);
             return initiateDeployment(
                     flinkBlueGreenDeployment,
                     deploymentStatus,
@@ -115,12 +114,6 @@ public UpdateControl<FlinkBlueGreenDeployment> reconcile(
             FlinkBlueGreenDeployments deployments =
                     FlinkBlueGreenDeployments.fromSecondaryResources(josdkContext);
 
-            // TODO: if a new deployment request comes while in the middle of a transition it's
-            //  currently ignored, but the new spec remains changed, should we roll it back?
-            // TODO: if we choose to leave a previously failed deployment 'running' for debug
-            // purposes,
-            //  we should flag it somehow as 'ROLLED_BACK' to signal that it can be overriden by a
-            // new deployment attempt.
             switch (deploymentStatus.getBlueGreenState()) {
                 case ACTIVE_BLUE:
                     return checkAndInitiateDeployment(
@@ -161,7 +154,21 @@ private UpdateControl<FlinkBlueGreenDeployment> monitorTransition(
             FlinkBlueGreenDeployments deployments,
             FlinkBlueGreenDeploymentStatus deploymentStatus,
             DeploymentType currentDeploymentType,
-            Context<FlinkBlueGreenDeployment> josdkContext) {
+            Context<FlinkBlueGreenDeployment> josdkContext)
+            throws JsonProcessingException {
+
+        if (hasSpecChanged(bgDeployment.getSpec(), deploymentStatus, currentDeploymentType)) {
+            // this means the spec was changed during transition,
+            //  ignore the new change, revert the spec and log as warning
+            bgDeployment.setSpec(
+                    SpecUtils.deserializeObject(
+                            deploymentStatus.getLastReconciledSpec(),
+                            "spec",
+                            FlinkBlueGreenDeploymentSpec.class));
+            josdkContext.getClient().resource(bgDeployment).replace();
+            LOG.warn(
+                    "Blue/Green Spec change detected during transition, ignored and reverted to the last reconciled spec");
+        }
 
         var nextState = FlinkBlueGreenDeploymentState.ACTIVE_BLUE;
         FlinkDeployment currentDeployment;
@@ -185,18 +192,8 @@ private UpdateControl<FlinkBlueGreenDeployment> monitorTransition(
 
         if (isDeploymentReady(nextDeployment, josdkContext, deploymentStatus)) {
             return deleteAndFinalize(
-                    bgDeployment,
-                    deploymentStatus,
-                    currentDeploymentType,
-                    josdkContext,
-                    currentDeployment,
-                    nextState);
+                    bgDeployment, deploymentStatus, josdkContext, currentDeployment, nextState);
         } else {
-            // This phase requires rescheduling the reconciliation because the pod initialization
-            // could get stuck
-            //  (e.g. waiting for resources)
-            // TODO: figure out the course of action for error/failure cases
-
             int maxNumRetries = bgDeployment.getSpec().getTemplate().getMaxNumRetries();
             if (maxNumRetries <= 0) {
                 maxNumRetries = DEFAULT_MAX_NUM_RETRIES;
@@ -242,7 +239,6 @@ private static int getReconciliationReschedInterval(FlinkBlueGreenDeployment bgD
     private UpdateControl<FlinkBlueGreenDeployment> deleteAndFinalize(
             FlinkBlueGreenDeployment bgDeployment,
             FlinkBlueGreenDeploymentStatus deploymentStatus,
-            DeploymentType currentDeploymentType,
             Context<FlinkBlueGreenDeployment> josdkContext,
             FlinkDeployment currentDeployment,
             FlinkBlueGreenDeploymentState nextState) {
@@ -251,11 +247,6 @@ private UpdateControl<FlinkBlueGreenDeployment> deleteAndFinalize(
             deleteDeployment(currentDeployment, josdkContext);
             return UpdateControl.noUpdate();
         } else {
-            deploymentStatus.setLastReconciledSpec(
-                    SpecUtils.serializeObject(bgDeployment.getSpec(), "spec"));
-
-            // TODO: Set the new child job STATUS to RUNNING too
-
             return patchStatusUpdateControl(
                     bgDeployment, deploymentStatus, nextState, JobStatus.RUNNING, false);
         }
@@ -272,6 +263,9 @@ private UpdateControl<FlinkBlueGreenDeployment> checkAndInitiateDeployment(
         if (hasSpecChanged(
                 flinkBlueGreenDeployment.getSpec(), deploymentStatus, currentDeploymentType)) {
 
+            // Ack the change in the spec (setLastReconciledSpec)
+            setLastReconciledSpec(flinkBlueGreenDeployment, deploymentStatus);
+
             FlinkDeployment currentFlinkDeployment =
                     DeploymentType.BLUE == currentDeploymentType
                             ? deployments.getFlinkDeploymentBlue()
@@ -286,10 +280,6 @@ private UpdateControl<FlinkBlueGreenDeployment> checkAndInitiateDeployment(
                 FlinkResourceContext<FlinkDeployment> resourceContext =
                         ctxFactory.getResourceContext(currentFlinkDeployment, josdkContext);
 
-                // TODO: this operation is already done by hasSpecChanged() above, dedup later
-                String serializedSpec =
-                        SpecUtils.serializeObject(flinkBlueGreenDeployment.getSpec(), "spec");
-
                 // Updating status
                 if (DeploymentType.BLUE == currentDeploymentType) {
                     nextState = FlinkBlueGreenDeploymentState.TRANSITIONING_TO_GREEN;
@@ -323,6 +313,14 @@ private UpdateControl<FlinkBlueGreenDeployment> checkAndInitiateDeployment(
         return UpdateControl.noUpdate();
     }
 
+    private static void setLastReconciledSpec(
+            FlinkBlueGreenDeployment flinkBlueGreenDeployment,
+            FlinkBlueGreenDeploymentStatus deploymentStatus) {
+        deploymentStatus.setLastReconciledSpec(
+                SpecUtils.serializeObject(flinkBlueGreenDeployment.getSpec(), "spec"));
+        deploymentStatus.setLastReconciledTimestamp(System.currentTimeMillis());
+    }
+
     public void logPotentialWarnings(
             FlinkDeployment flinkDeployment,
             Context<FlinkBlueGreenDeployment> josdkContext,
@@ -345,7 +343,7 @@ public void logPotentialWarnings(
             LOG.warn("Deployment not healthy, some Pods have the following status: " + podPhases);
         }
 
-        List<Event> badEvents =
+        List<Event> abnormalEvents =
                 josdkContext
                         .getClient()
                         .v1()
@@ -373,8 +371,8 @@ public void logPotentialWarnings(
                                                                         .contains(p)))
                         .collect(Collectors.toList());
 
-        if (!badEvents.isEmpty()) {
-            LOG.warn("Bad events detected: " + badEvents);
+        if (!abnormalEvents.isEmpty()) {
+            LOG.warn("Abnormal events detected: " + abnormalEvents);
         }
     }
 
@@ -393,9 +391,7 @@ private static Savepoint configureSavepoint(
                                                 .getJobId()),
                                 resourceContext.getObserveConfig());
 
-        // TODO 1: check the last CP age with the logic from
-        // AbstractJobReconciler.changeLastStateIfCheckpointTooOld
-        // TODO 2: if no checkpoint is available, take a savepoint? throw error?
+        // TODO: alternative action if no checkpoint is available?
         if (lastCheckpoint.isEmpty()) {
             throw new IllegalStateException(
                     "Last Checkpoint for Job "
@@ -422,8 +418,6 @@ private UpdateControl<FlinkBlueGreenDeployment> initiateDeployment(
                 josdkContext,
                 isFirstDeployment);
 
-        // TODO: set child job status to JobStatus.INITIALIZING
-
         return patchStatusUpdateControl(
                         flinkBlueGreenDeployment,
                         deploymentStatus,
@@ -439,8 +433,7 @@ private boolean isDeploymentReady(
             FlinkBlueGreenDeploymentStatus deploymentStatus) {
         if (ResourceLifecycleState.STABLE == deployment.getStatus().getLifecycleState()
                 && JobStatus.RUNNING == deployment.getStatus().getJobStatus().getState()) {
-            // TODO: verify, e.g. will pods be "pending" after the FlinkDeployment is RUNNING and
-            // STABLE?
+            // TODO: checking for running pods seems to be redundant, check if this can be removed
             int notRunningPods =
                     (int)
                             getDeploymentPods(josdkContext, deployment)
@@ -501,7 +494,6 @@ private UpdateControl<FlinkBlueGreenDeployment> patchStatusUpdateControl(
             deploymentStatus.getJobStatus().setState(jobState);
         }
 
-        deploymentStatus.setLastReconciledTimestamp(System.currentTimeMillis());
         flinkBlueGreenDeployment.setStatus(deploymentStatus);
         return UpdateControl.patchStatus(flinkBlueGreenDeployment);
     }
@@ -553,9 +545,6 @@ private void deploy(
 
     private static void deleteDeployment(
             FlinkDeployment currentDeployment, Context<FlinkBlueGreenDeployment> josdkContext) {
-        // TODO: This gets called multiple times, check to see if it's already in a TERMINATING
-        // state
-        // (or only execute if RUNNING)
         List<StatusDetails> deletedStatus =
                 josdkContext
                         .getClient()

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

@@ -107,12 +107,13 @@ public void verifyBasicDeployment(FlinkVersion flinkVersion) throws Exception {
         var blueGreenDeployment =
                 buildSessionCluster(TEST_DEPLOYMENT_NAME, TEST_NAMESPACE, flinkVersion);
 
-        // 1. Initiate the Green deployment
+        // 1. Initiate the Blue deployment
         var bgSpecBefore = blueGreenDeployment.getSpec();
         Long minReconciliationTs = System.currentTimeMillis() - 1;
         var rs = reconcile(blueGreenDeployment);
 
-        assertSpec(rs, minReconciliationTs);
+        assertTrue(rs.updateControl.isPatchStatus());
+        assertTrue(minReconciliationTs < rs.reconciledStatus.getLastReconciledTimestamp());
 
         // check the status (reconciled spec, reconciled ts, a/b state)
         assertEquals(
@@ -129,12 +130,10 @@ public void verifyBasicDeployment(FlinkVersion flinkVersion) throws Exception {
 
         simulateSubmitAndSuccessfulJobStart(deploymentA);
 
-        // 2. Finalize the Green deployment
-        minReconciliationTs = System.currentTimeMillis() - 1;
+        // 2. Finalize the Blue deployment
         rs = reconcile(rs.deployment);
 
-        assertSpec(rs, minReconciliationTs);
-
+        assertTrue(rs.updateControl.isPatchStatus());
         assertEquals(
                 SpecUtils.serializeObject(bgSpecBefore, "spec"),
                 rs.reconciledStatus.getLastReconciledSpec());
@@ -182,7 +181,8 @@ private void testTransitionToGreen(
         var flinkDeployments = getFlinkDeployments();
         var greenDeploymentName = flinkDeployments.get(1).getMetadata().getName();
 
-        assertSpec(rs, minReconciliationTs);
+        assertTrue(rs.updateControl.isPatchStatus());
+        assertTrue(minReconciliationTs < rs.reconciledStatus.getLastReconciledTimestamp());
 
         assertEquals(2, flinkDeployments.size());
         assertNull(flinkDeployments.get(0).getSpec().getJob().getInitialSavepointPath());
@@ -211,12 +211,9 @@ private void testTransitionToGreen(
         assertEquals(1, flinkDeployments.size());
         assertEquals(greenDeploymentName, flinkDeployments.get(0).getMetadata().getName());
 
-        minReconciliationTs = System.currentTimeMillis() - 1;
         rs = reconcile(rs.deployment);
 
-        // Spec should still be the new one
-        assertSpec(rs, minReconciliationTs);
-
+        assertTrue(rs.updateControl.isPatchStatus());
         assertNotNull(rs.reconciledStatus.getLastReconciledSpec());
         assertEquals(
                 SpecUtils.serializeObject(bgUpdatedSpec, "spec"),
@@ -252,7 +249,8 @@ public void verifyFailureBeforeTransition(FlinkVersion flinkVersion) throws Exce
         var minReconciliationTs = System.currentTimeMillis() - 1;
         rs = reconcile(rs.deployment);
 
-        assertSpec(rs, minReconciliationTs);
+        assertTrue(rs.updateControl.isPatchStatus());
+        assertTrue(minReconciliationTs < rs.reconciledStatus.getLastReconciledTimestamp());
 
         // Assert job status/state is left the way it is and that the Blue job never got submitted
         assertEquals(JobStatus.FAILING, rs.reconciledStatus.getJobStatus().getState());
@@ -315,7 +313,6 @@ public void verifyFailureDuringTransition(FlinkVersion flinkVersion) throws Exce
 
         // 4a. Simulating the Blue deployment doesn't start correctly (status will remain the same)
         //  Asserting the status retry count is incremented by 1
-        long lastTs = System.currentTimeMillis();
         for (int i = 1; i <= maxNumRetries; i++) {
             Thread.sleep(1);
             rs = reconcile(rs.deployment);
@@ -326,16 +323,13 @@ public void verifyFailureDuringTransition(FlinkVersion flinkVersion) throws Exce
                     reconciliationReschedulingIntervalMs,
                     rs.updateControl.getScheduleDelay().get());
             assertEquals(i, rs.reconciledStatus.getNumRetries());
-            assertTrue(rs.reconciledStatus.getLastReconciledTimestamp() > lastTs);
-            lastTs = rs.reconciledStatus.getLastReconciledTimestamp();
             System.out.println();
         }
 
         // 4b. After the retries are exhausted
-        var minReconciliationTs = System.currentTimeMillis() - 1;
         rs = reconcile(rs.deployment);
 
-        assertSpec(rs, minReconciliationTs);
+        assertTrue(rs.updateControl.isPatchStatus());
 
         // The first job should be RUNNING, the second should be SUSPENDED
         assertEquals(JobStatus.FAILING, rs.reconciledStatus.getJobStatus().getState());
@@ -361,17 +355,33 @@ public void verifyFailureDuringTransition(FlinkVersion flinkVersion) throws Exce
 
         // 6. Initiate the redeployment
         var bgUpdatedSpec = rs.deployment.getSpec();
-        minReconciliationTs = System.currentTimeMillis() - 1;
+        var minReconciliationTs = System.currentTimeMillis() - 1;
         rs = reconcile(rs.deployment);
 
         testTransitionToGreen(rs, minReconciliationTs, customValue, bgUpdatedSpec);
     }
 
-    private static void assertSpec(
-            TestingFlinkBlueGreenDeploymentController.BlueGreenReconciliationResult rs,
-            long minReconciliationTs) {
-        assertTrue(rs.updateControl.isPatchStatus());
-        assertTrue(minReconciliationTs < rs.reconciledStatus.getLastReconciledTimestamp());
+    @ParameterizedTest
+    @MethodSource("org.apache.flink.kubernetes.operator.TestUtils#flinkVersions")
+    public void verifySpecChangeDuringTransition(FlinkVersion flinkVersion) throws Exception {
+        var blueGreenDeployment =
+                buildSessionCluster(TEST_DEPLOYMENT_NAME, TEST_NAMESPACE, flinkVersion);
+
+        // 1. Initiate the Blue deployment
+        var originalSpec = blueGreenDeployment.getSpec();
+        var rs = reconcile(blueGreenDeployment);
+
+        // 2. Job starting...
+        simulateSubmitAndSuccessfulJobStart(getFlinkDeployments().get(0));
+
+        // 3. Simulate a spec change before the transition is complete
+        simulateChangeInSpec(rs.deployment, "MODIFIED_VALUE");
+        rs = reconcile(rs.deployment);
+
+        // The spec should have been reverted
+        assertEquals(
+                SpecUtils.serializeObject(originalSpec, "spec"),
+                SpecUtils.serializeObject(rs.deployment.getSpec(), "spec"));
     }
 
     private void simulateChangeInSpec(