Defensive code vs. non-BASIC TransitionMode (only for FLIP-503)

Author: schongloo

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

@@ -22,6 +22,7 @@
 import org.apache.flink.kubernetes.operator.api.FlinkBlueGreenDeployment;
 import org.apache.flink.kubernetes.operator.api.FlinkDeployment;
 import org.apache.flink.kubernetes.operator.api.bluegreen.DeploymentType;
+import org.apache.flink.kubernetes.operator.api.bluegreen.TransitionMode;
 import org.apache.flink.kubernetes.operator.api.lifecycle.ResourceLifecycleState;
 import org.apache.flink.kubernetes.operator.api.spec.FlinkBlueGreenDeploymentSpec;
 import org.apache.flink.kubernetes.operator.api.spec.JobState;
@@ -54,6 +55,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import javax.naming.OperationNotSupportedException;
+
 import java.time.Instant;
 import java.util.List;
 import java.util.Map;
@@ -98,6 +101,12 @@ public UpdateControl<FlinkBlueGreenDeployment> reconcile(
             FlinkBlueGreenDeployment bgDeployment, Context<FlinkBlueGreenDeployment> josdkContext)
             throws Exception {
 
+        // TODO: this verification is only for FLIP-503, remove later.
+        if (bgDeployment.getSpec().getTemplate().getTransitionMode() != TransitionMode.BASIC) {
+            throw new OperationNotSupportedException(
+                    "Only TransitionMode == BASIC is currently supported");
+        }
+
         FlinkBlueGreenDeploymentStatus deploymentStatus = bgDeployment.getStatus();
 
         if (deploymentStatus == null) {
@@ -568,6 +577,9 @@ private boolean hasSpecChanged(
 
         String lastReconciledSpec = deploymentStatus.getLastReconciledSpec();
         String newSpecSerialized = SpecUtils.serializeObject(newSpec, "spec");
+
+        // TODO: in FLIP-504 check here the TransitionMode has not been changed
+
         return !lastReconciledSpec.equals(newSpecSerialized);
     }
 

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

@@ -24,6 +24,7 @@
 import org.apache.flink.kubernetes.operator.TestingFlinkService;
 import org.apache.flink.kubernetes.operator.api.FlinkBlueGreenDeployment;
 import org.apache.flink.kubernetes.operator.api.FlinkDeployment;
+import org.apache.flink.kubernetes.operator.api.bluegreen.TransitionMode;
 import org.apache.flink.kubernetes.operator.api.spec.FlinkBlueGreenDeploymentSpec;
 import org.apache.flink.kubernetes.operator.api.spec.FlinkDeploymentSpec;
 import org.apache.flink.kubernetes.operator.api.spec.FlinkDeploymentTemplateSpec;
@@ -53,6 +54,8 @@
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.MethodSource;
 
+import javax.naming.OperationNotSupportedException;
+
 import java.time.Instant;
 import java.util.HashMap;
 import java.util.List;
@@ -66,6 +69,7 @@
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 /** {@link FlinkBlueGreenDeploymentController} tests. */
@@ -280,7 +284,6 @@ public void verifyFailureBeforeFirstDeployment(FlinkVersion flinkVersion) throws
                 buildSessionCluster(TEST_DEPLOYMENT_NAME, TEST_NAMESPACE, flinkVersion);
 
         // Initiate the Blue deployment
-        var originalSpec = blueGreenDeployment.getSpec();
         var rs = initialPhaseBasicDeployment(blueGreenDeployment, false);
 
         // Simulating the job did not start correctly before the AbortGracePeriodMs
@@ -319,6 +322,20 @@ public void verifyFailureBeforeFirstDeployment(FlinkVersion flinkVersion) throws
                 rs.reconciledStatus.getBlueGreenState());
     }
 
+    // TODO: this test is only for FLIP-503, remove later.
+    @ParameterizedTest
+    @MethodSource("org.apache.flink.kubernetes.operator.TestUtils#flinkVersions")
+    public void verifyOnlyBasicTransitionIsAllowed(FlinkVersion flinkVersion) throws Exception {
+        var blueGreenDeployment =
+                buildSessionCluster(TEST_DEPLOYMENT_NAME, TEST_NAMESPACE, flinkVersion);
+
+        blueGreenDeployment.getSpec().getTemplate().setTransitionMode(null);
+        assertThrows(OperationNotSupportedException.class, () -> reconcile(blueGreenDeployment));
+
+        blueGreenDeployment.getSpec().getTemplate().setTransitionMode(TransitionMode.ADVANCED);
+        assertThrows(OperationNotSupportedException.class, () -> reconcile(blueGreenDeployment));
+    }
+
     private TestingFlinkBlueGreenDeploymentController.BlueGreenReconciliationResult
             executeBasicDeployment(
                     FlinkVersion flinkVersion,
@@ -372,6 +389,12 @@ public void verifyFailureBeforeFirstDeployment(FlinkVersion flinkVersion) throws
             initialPhaseBasicDeployment(
                     FlinkBlueGreenDeployment blueGreenDeployment, boolean execAssertions)
                     throws Exception {
+        if (blueGreenDeployment.getSpec().getTemplate().getTransitionMode()
+                != TransitionMode.BASIC) {
+            throw new OperationNotSupportedException(
+                    "Only TransitionMode == BASIC is currently supported");
+        }
+
         Long minReconciliationTs = System.currentTimeMillis() - 1;
 
         // 1a. Initializing deploymentStatus with this call
@@ -594,6 +617,7 @@ private static FlinkBlueGreenDeployment buildSessionCluster(
                                 .state(JobState.RUNNING)
                                 .build());
 
+        bgDeploymentSpec.getTemplate().setTransitionMode(TransitionMode.BASIC);
         deployment.setSpec(bgDeploymentSpec);
         return deployment;
     }