Addressing PR comments. Corrected abort/delay logic. Added the e2e tests to ci.yml. Missing Licenses.

schongloo · schongloo · commit f7121252f4b9 · 2025-08-18T23:35:11.000-07:00
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
@@ -172,6 +172,8 @@ jobs:
           - test_flink_operator_ha.sh
           - test_snapshot.sh
           - test_batch_job.sh
+          - test_bluegreen_laststate.sh
+          - test_bluegreen_stateless.sh
         exclude:
           - flink-version: v1_16
             test: test_autoscaler.sh
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkBlueGreenDeploymentController.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkBlueGreenDeploymentController.java
@@ -42,7 +42,6 @@
 import java.util.ArrayList;
 import java.util.List;
 
-import static org.apache.flink.kubernetes.operator.api.spec.FlinkBlueGreenDeploymentConfigOptions.ABORT_GRACE_PERIOD;
 import static org.apache.flink.kubernetes.operator.api.status.FlinkBlueGreenDeploymentState.INITIALIZING_BLUE;
 
 /**
@@ -72,8 +71,6 @@ public class FlinkBlueGreenDeploymentController implements Reconciler<FlinkBlueG
     private final FlinkResourceContextFactory ctxFactory;
     private final BlueGreenStateHandlerRegistry handlerRegistry;
 
-    public static long minimumAbortGracePeriodMs = ABORT_GRACE_PERIOD.defaultValue().toMillis();
-
     public FlinkBlueGreenDeploymentController(FlinkResourceContextFactory ctxFactory) {
         this.ctxFactory = ctxFactory;
         this.handlerRegistry = new BlueGreenStateHandlerRegistry();
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/bluegreen/BlueGreenContext.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/bluegreen/BlueGreenContext.java
@@ -1,3 +1,20 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
 package org.apache.flink.kubernetes.operator.controller.bluegreen;
 
 import org.apache.flink.kubernetes.operator.api.FlinkBlueGreenDeployment;
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
@@ -426,14 +426,15 @@ private UpdateControl<FlinkBlueGreenDeployment> shouldWeAbort(
         if (abortTimestamp < System.currentTimeMillis()) {
             return abortDeployment(context, nextDeployment, nextState, deploymentName);
         } else {
-            return retryDeployment(context, abortTimestamp, deploymentName);
+            return retryDeployment(context, deploymentName);
         }
     }
 
     private UpdateControl<FlinkBlueGreenDeployment> retryDeployment(
-            BlueGreenContext context, long abortTimestamp, String deploymentName) {
+            BlueGreenContext context, String deploymentName) {
+
+        long delay = getReconciliationReschedInterval(context);
 
-        long delay = abortTimestamp - System.currentTimeMillis();
         LOG.info(
                 "FlinkDeployment '{}' not ready yet, retrying in {} seconds.",
                 deploymentName,
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/bluegreen/handlers/InitializingBlueStateHandler.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/bluegreen/handlers/InitializingBlueStateHandler.java
@@ -1,3 +1,20 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
 package org.apache.flink.kubernetes.operator.controller.bluegreen.handlers;
 
 import org.apache.flink.api.common.JobStatus;
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/bluegreen/handlers/SavepointingStateHandler.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/bluegreen/handlers/SavepointingStateHandler.java
@@ -1,3 +1,20 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
 package org.apache.flink.kubernetes.operator.controller.bluegreen.handlers;
 
 import org.apache.flink.kubernetes.operator.api.FlinkBlueGreenDeployment;
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/bluegreen/BlueGreenUtils.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/bluegreen/BlueGreenUtils.java
@@ -17,7 +17,6 @@
 
 package org.apache.flink.kubernetes.operator.utils.bluegreen;
 
-import org.apache.flink.kubernetes.operator.controller.FlinkBlueGreenDeploymentController;
 import org.apache.flink.kubernetes.operator.controller.bluegreen.BlueGreenContext;
 
 import java.time.Instant;
@@ -90,8 +89,10 @@ public static long getAbortGracePeriod(BlueGreenContext context) {
         long abortGracePeriod =
                 BlueGreenSpecUtils.getConfigOption(context.getBgDeployment(), ABORT_GRACE_PERIOD)
                         .toMillis();
-        return Math.max(
-                abortGracePeriod, FlinkBlueGreenDeploymentController.minimumAbortGracePeriodMs);
+        return abortGracePeriod;
+        //        return Math.max(
+        //                abortGracePeriod,
+        // FlinkBlueGreenDeploymentController.minimumAbortGracePeriodMs);
     }
 
     /**
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
@@ -84,6 +84,7 @@ public class FlinkBlueGreenDeploymentControllerTest {
     private static final String CUSTOM_CONFIG_FIELD = "custom-configuration-field";
     private static final int DEFAULT_DELETION_DELAY_VALUE = 500;
     private static final int ALT_DELETION_DELAY_VALUE = 1000;
+    private static final int MINIMUM_ABORT_GRACE_PERIOD = 1000;
     private static final String TEST_CHECKPOINT_PATH = "/tmp/checkpoints";
     private static final String TEST_INITIAL_SAVEPOINT_PATH = "/tmp/savepoints";
     private final FlinkConfigManager configManager = new FlinkConfigManager(new Configuration());
@@ -240,7 +241,7 @@ public void verifyFailureDuringTransition(FlinkVersion flinkVersion) throws Exce
 
         // Overriding the maxNumRetries and Reschedule Interval
         var abortGracePeriodMs = 1200;
-        var reconciliationReschedulingIntervalMs = 5000;
+        var reconciliationReschedulingIntervalMs = 3000;
         Map<String, String> configuration =
                 blueGreenDeployment.getSpec().getTemplate().getConfiguration();
         configuration.put(ABORT_GRACE_PERIOD.key(), String.valueOf(abortGracePeriodMs));
@@ -280,7 +281,7 @@ public void verifyFailureDuringTransition(FlinkVersion flinkVersion) throws Exce
             assertFalse(rs.updateControl.isPatchResource());
             assertTrue(rs.updateControl.getScheduleDelay().isPresent());
             reschedDelayMs = rs.updateControl.getScheduleDelay().get();
-            assertTrue(reschedDelayMs < abortGracePeriodMs && reschedDelayMs > 0);
+            assertTrue(reschedDelayMs == reconciliationReschedulingIntervalMs && reschedDelayMs > 0);
             assertTrue(
                     instantStrToMillis(rs.reconciledStatus.getAbortTimestamp())
                             > System.currentTimeMillis());
@@ -365,7 +366,7 @@ public void verifyFailureBeforeFirstDeployment(FlinkVersion flinkVersion) throws
         var rs = initialPhaseBasicDeployment(blueGreenDeployment, false);
 
         // Simulating the job did not start correctly before the AbortGracePeriodMs
-        Thread.sleep(FlinkBlueGreenDeploymentController.minimumAbortGracePeriodMs);
+        Thread.sleep(MINIMUM_ABORT_GRACE_PERIOD);
 
         rs = reconcile(rs.deployment);
 
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/TestingFlinkBlueGreenDeploymentController.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/TestingFlinkBlueGreenDeploymentController.java
@@ -51,7 +51,6 @@ public TestingFlinkBlueGreenDeploymentController(
                         null);
 
         flinkBlueGreenDeploymentController = new FlinkBlueGreenDeploymentController(contextFactory);
-        flinkBlueGreenDeploymentController.minimumAbortGracePeriodMs = 1000;
     }
 
     @Override

Original file line number	Diff line number	Diff line change
`@@ -51,7 +51,6 @@ public TestingFlinkBlueGreenDeploymentController(`
`51`	`51`	`null);`
`52`	`52`
`53`	`53`	`flinkBlueGreenDeploymentController = new FlinkBlueGreenDeploymentController(contextFactory);`
`54`		`- flinkBlueGreenDeploymentController.minimumAbortGracePeriodMs = 1000;`
`55`	`54`	`}`
`56`	`55`
`57`	`56`	`@Override`