Taking savepoint also for LAST-STATE, removed the last checkpoint usage.

Author: schongloo

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

@@ -47,11 +47,9 @@
 import static org.apache.flink.kubernetes.operator.controller.bluegreen.BlueGreenKubernetesService.suspendFlinkDeployment;
 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;
@@ -224,7 +222,7 @@ private Savepoint configureInitialSavepoint(
                 context.getCtxFactory()
                         .getResourceContext(currentFlinkDeployment, context.getJosdkContext());
 
-        // If a savepoint is required we fetch it, should be ready by this point
+        // For now UpgradeMode != STATELESS will use a savepoint, originally only SAVEPOINT
         if (isSavepointRequired(context)) {
             String savepointTriggerId = context.getDeploymentStatus().getSavepointTriggerId();
             var savepointFetchResult = fetchSavepointInfo(ctx, savepointTriggerId);
@@ -242,13 +240,15 @@ private Savepoint configureInitialSavepoint(
                     savepointFormatType);
         }
 
-        // Else we start looking for the last checkpoint if needed
+        // The logic below looked for the last checkpoint in case upgradeMode = LAST_STATE
+        // We don't want to rely on last checkpoint for now.
+        return null;
 
-        if (!lookForCheckpoint(context)) {
-            return null;
-        }
-
-        return getLastCheckpoint(ctx);
+        //        if (!lookForCheckpoint(context)) {
+        //            return null;
+        //        }
+        //
+        //        return getLastCheckpoint(ctx);
     }
 
     private boolean handleSavepoint(

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/bluegreen/BlueGreenSpecUtils.java

@@ -106,13 +106,16 @@ public static BlueGreenDiffType getSpecDiff(BlueGreenContext context) {
     }
 
     public static boolean isSavepointRequired(BlueGreenContext context) {
-        return UpgradeMode.SAVEPOINT
-                == context.getBgDeployment()
+        UpgradeMode upgradeMode =
+                context.getBgDeployment()
                         .getSpec()
                         .getTemplate()
                         .getSpec()
                         .getJob()
                         .getUpgradeMode();
+        //        return UpgradeMode.SAVEPOINT == upgradeMode;
+        // For now we're taking savepoints in STATELESS or LAST-STATE
+        return UpgradeMode.STATELESS != upgradeMode;
     }
 
     public static boolean lookForCheckpoint(BlueGreenContext context) {

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

@@ -131,10 +131,9 @@ public void verifyBasicTransition(
         simulateChangeInSpec(
                 rs.deployment, customValue, ALT_DELETION_DELAY_VALUE, initialSavepointPath);
 
-        var expectedSavepointPath =
-                upgradeMode == UpgradeMode.LAST_STATE ? TEST_CHECKPOINT_PATH : initialSavepointPath;
+        var expectedSavepointPath = initialSavepointPath;
 
-        if (upgradeMode == UpgradeMode.SAVEPOINT) {
+        if (upgradeMode != UpgradeMode.STATELESS) {
             // In this case there will ALWAYS be a savepoint generated with this value,
             // regardless of the initialSavepointPath
             expectedSavepointPath = "savepoint_1";