Resolving conflicts after rebasing from main (FlinkConfiguration is now a ConfigObjectNode) and updating B/G e2e tests.

Author: schongloo

e2e-tests/test_bluegreen_laststate.sh

@@ -42,7 +42,7 @@ TIMEOUT=300
 #echo "APPLICATION_IDENTIFIER " $APPLICATION_IDENTIFIER
 #echo "BLUE_APPLICATION_IDENTIFIER " $BLUE_APPLICATION_IDENTIFIER
 
-#on_exit cleanup_and_exit "$APPLICATION_YAML" $TIMEOUT $BG_CLUSTER_ID
+on_exit cleanup_and_exit "$APPLICATION_YAML" $TIMEOUT $BG_CLUSTER_ID
 
 retry_times 5 30 "kubectl apply -f $APPLICATION_YAML" || exit 1
 
@@ -52,11 +52,11 @@ wait_for_status $BLUE_APPLICATION_IDENTIFIER '.status.lifecycleState' STABLE ${T
 wait_for_status $APPLICATION_IDENTIFIER '.status.jobStatus.state' RUNNING ${TIMEOUT} || exit 1
 wait_for_status $APPLICATION_IDENTIFIER '.status.blueGreenState' ACTIVE_BLUE ${TIMEOUT} || exit 1
 
-blue_job_id=$(kubectl get -oyaml flinkdep/basic-bluegreen-example-blue | yq '.status.jobStatus.jobId')
+#blue_job_id=$(kubectl get -oyaml flinkdep/basic-bluegreen-example-blue | yq '.status.jobStatus.jobId')
 
-echo "Giving a chance for checkpoints to be generated..."
-sleep 5
 kubectl patch flinkbgdep ${BG_CLUSTER_ID} --type merge --patch '{"spec":{"template":{"spec":{"flinkConfiguration":{"rest.port":"8082","state.checkpoints.num-retained":"51"}}}}}'
+echo "Resource patched, giving a chance for the savepoint to be taken..."
+sleep 10
 
 wait_for_status $GREEN_APPLICATION_IDENTIFIER '.status.lifecycleState' STABLE ${TIMEOUT} || exit 1
 kubectl wait --for=delete deployment --timeout=${TIMEOUT}s --selector="app=${BLUE_CLUSTER_ID}"
@@ -65,7 +65,7 @@ wait_for_status $APPLICATION_IDENTIFIER '.status.blueGreenState' ACTIVE_GREEN ${
 
 green_initialSavepointPath=$(kubectl get -oyaml $GREEN_APPLICATION_IDENTIFIER | yq '.spec.job.initialSavepointPath')
 
-if [[ $green_initialSavepointPath == '/flink-data/checkpoints/'$blue_job_id* ]]; then
+if [[ $green_initialSavepointPath == '/flink-data/savepoints/savepoint-'* ]]; then
   echo 'Green deployment started from the expected initialSavepointPath: ' $green_initialSavepointPath
 else
   echo 'Unexpected initialSavepointPath: ' $green_initialSavepointPath

e2e-tests/test_bluegreen_stateless.sh

@@ -47,8 +47,6 @@ wait_for_status $BLUE_APPLICATION_IDENTIFIER '.status.lifecycleState' STABLE ${T
 wait_for_status $APPLICATION_IDENTIFIER '.status.jobStatus.state' RUNNING ${TIMEOUT} || exit 1
 wait_for_status $APPLICATION_IDENTIFIER '.status.blueGreenState' ACTIVE_BLUE ${TIMEOUT} || exit 1
 
-blue_job_id=$(kubectl get -oyaml flinkdep/basic-bluegreen-example-blue | yq '.status.jobStatus.jobId')
-
 echo "PATCHING B/G deployment..."
 kubectl patch flinkbgdep ${BG_CLUSTER_ID} --type merge --patch '{"spec":{"template":{"spec":{"flinkConfiguration":{"rest.port":"8082","taskmanager.numberOfTaskSlots":"2"}}}}}'
 

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

@@ -25,6 +25,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.spec.ConfigObjectNode;
 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;
@@ -270,12 +271,8 @@ public void verifyFailureDuringTransition(FlinkVersion flinkVersion) throws Exce
                 rs.reconciledStatus.getBlueGreenState());
         assertEquals(
                 customValue,
-                rs.deployment
-                        .getSpec()
-                        .getTemplate()
-                        .getSpec()
-                        .getFlinkConfiguration()
-                        .get(CUSTOM_CONFIG_FIELD));
+                getFlinkConfigurationValue(
+                        rs.deployment.getSpec().getTemplate().getSpec(), CUSTOM_CONFIG_FIELD));
 
         // Simulating the Blue deployment doesn't start correctly (status will remain the same)
         Long reschedDelayMs = 0L;
@@ -328,6 +325,11 @@ public void verifyFailureDuringTransition(FlinkVersion flinkVersion) throws Exce
         testTransitionToGreen(rs, customValue, null);
     }
 
+    private static String getFlinkConfigurationValue(
+            FlinkDeploymentSpec flinkDeploymentSpec, String propertyName) {
+        return flinkDeploymentSpec.getFlinkConfiguration().get(propertyName).asText();
+    }
+
     @ParameterizedTest
     @MethodSource("org.apache.flink.kubernetes.operator.TestUtils#flinkVersions")
     public void verifySpecChangeDuringTransition(FlinkVersion flinkVersion) throws Exception {
@@ -702,11 +704,9 @@ public void verifySpecificBehavior(
             assertEquals(1, deployments.size());
             assertEquals(
                     "100 SECONDS",
-                    deployments
-                            .get(0)
-                            .getSpec()
-                            .getFlinkConfiguration()
-                            .get("kubernetes.operator.reconcile.interval"));
+                    getFlinkConfigurationValue(
+                            deployments.get(0).getSpec(),
+                            "kubernetes.operator.reconcile.interval"));
         }
     }
 
@@ -763,11 +763,9 @@ public void verifySpecificBehavior(
             // Child spec change should be applied to FlinkDeployment
             assertEquals(
                     "100 SECONDS",
-                    deployments
-                            .get(0)
-                            .getSpec()
-                            .getFlinkConfiguration()
-                            .get("kubernetes.operator.reconcile.interval"));
+                    getFlinkConfigurationValue(
+                            deployments.get(0).getSpec(),
+                            "kubernetes.operator.reconcile.interval"));
 
             // Top-level changes should be preserved in reconciled spec
             assertNotNull(result.rs.reconciledStatus.getLastReconciledSpec());
@@ -1043,12 +1041,8 @@ private void testTransitionToGreen(
         assertEquals(0, instantStrToMillis(rs.reconciledStatus.getDeploymentReadyTimestamp()));
         assertEquals(
                 customValue,
-                rs.deployment
-                        .getSpec()
-                        .getTemplate()
-                        .getSpec()
-                        .getFlinkConfiguration()
-                        .get(CUSTOM_CONFIG_FIELD));
+                getFlinkConfigurationValue(
+                        rs.deployment.getSpec().getTemplate().getSpec(), CUSTOM_CONFIG_FIELD));
 
         // Initiate and mark the Green deployment ready
         simulateSuccessfulJobStart(getFlinkDeployments().get(1));
@@ -1213,12 +1207,14 @@ private static FlinkBlueGreenDeploymentSpec getTestFlinkDeploymentSpec(FlinkVers
                         .imagePullPolicy(IMAGE_POLICY)
                         .serviceAccount(SERVICE_ACCOUNT)
                         .flinkVersion(version)
-                        .flinkConfiguration(conf)
+                        .flinkConfiguration(new ConfigObjectNode())
                         .jobManager(new JobManagerSpec(new Resource(1.0, "2048m", "2G"), 1, null))
                         .taskManager(
                                 new TaskManagerSpec(new Resource(1.0, "2048m", "2G"), null, null))
                         .build();
 
+        flinkDeploymentSpec.setFlinkConfiguration(conf);
+
         Map<String, String> configuration = new HashMap<>();
         configuration.put(ABORT_GRACE_PERIOD.key(), "1");
         configuration.put(RECONCILIATION_RESCHEDULING_INTERVAL.key(), "500");