Updated unit test to assert Savepointing. Checkstyle fixes

Author: schongloo

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

@@ -118,7 +118,7 @@ public UpdateControl<FlinkBlueGreenDeployment> checkAndInitiateDeployment(
                 if (specDiff == BlueGreenDiffType.TRANSITION) {
                     if (handleSavepoint(context, currentFlinkDeployment)) {
                         // This is the only portion where the last reconciled spec is not set,
-                        // so we can process TRANSITION after the savepoint is done
+                        // so we can reprocess TRANSITION after the savepoint is done
                         var savepointingState = calculateSavepointingState(currentDeploymentType);
                         return patchStatusUpdateControl(context, savepointingState, null)
                                 .rescheduleAfter(getReconciliationReschedInterval(context));

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

@@ -1,15 +1,19 @@
 package org.apache.flink.kubernetes.operator.controller.bluegreen.handlers;
 
-import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
 import org.apache.flink.kubernetes.operator.api.FlinkBlueGreenDeployment;
 import org.apache.flink.kubernetes.operator.api.bluegreen.DeploymentType;
 import org.apache.flink.kubernetes.operator.api.status.FlinkBlueGreenDeploymentState;
 import org.apache.flink.kubernetes.operator.controller.bluegreen.BlueGreenContext;
 import org.apache.flink.kubernetes.operator.controller.bluegreen.BlueGreenDeploymentService;
 
+import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
+
 import static org.apache.flink.kubernetes.operator.controller.bluegreen.BlueGreenDeploymentService.patchStatusUpdateControl;
 import static org.apache.flink.kubernetes.operator.utils.bluegreen.BlueGreenUtils.getReconciliationReschedInterval;
 
+/**
+ * State handler for managing Blue/Green deployment savepointing transitions.
+ */
 public class SavepointingStateHandler extends AbstractBlueGreenStateHandler {
 
     public SavepointingStateHandler(

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

@@ -17,7 +17,6 @@
 
 package org.apache.flink.kubernetes.operator.utils.bluegreen;
 
-import lombok.SneakyThrows;
 import org.apache.flink.api.common.JobID;
 import org.apache.flink.configuration.CheckpointingOptions;
 import org.apache.flink.configuration.ConfigOption;
@@ -40,6 +39,7 @@
 import org.apache.flink.util.Preconditions;
 
 import io.fabric8.kubernetes.api.model.ObjectMeta;
+import lombok.SneakyThrows;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 

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

@@ -142,12 +142,59 @@ public void verifyBasicTransition(
         simulateChangeInSpec(
                 rs.deployment, customValue, ALT_DELETION_DELAY_VALUE, initialSavepointPath);
 
-        // Transitioning to the Green deployment
         var expectedSavepointPath =
                 upgradeMode == UpgradeMode.LAST_STATE ? TEST_CHECKPOINT_PATH : initialSavepointPath;
+
+        if (upgradeMode == UpgradeMode.SAVEPOINT) {
+            // In this case there will ALWAYS be a savepoint generated with this value,
+            // regardless of the initialSavepointPath
+            expectedSavepointPath = "savepoint_1";
+            rs = handleSavepoint(rs);
+        }
+
+        // Transitioning to the Green deployment
+
         testTransitionToGreen(rs, customValue, expectedSavepointPath);
     }
 
+    @NotNull
+    private TestingFlinkBlueGreenDeploymentController.BlueGreenReconciliationResult handleSavepoint(
+            TestingFlinkBlueGreenDeploymentController.BlueGreenReconciliationResult rs) throws Exception {
+        var triggers = flinkService.getSavepointTriggers();
+        triggers.clear();
+
+        rs = reconcile(rs.deployment);
+
+        // Simulating a pending savepoint
+        triggers.put(rs.deployment.getStatus().getSavepointTriggerId(), false);
+
+        // Should be in SAVEPOINTING_BLUE state first
+        assertEquals(
+                FlinkBlueGreenDeploymentState.SAVEPOINTING_BLUE,
+                rs.reconciledStatus.getBlueGreenState());
+        assertTrue(rs.updateControl.isPatchStatus());
+        assertTrue(rs.updateControl.getScheduleDelay().isPresent());
+
+        // This next reconciliation should continue waiting on the pending savepoint
+        var rs2 = reconcile(rs.deployment);
+
+        assertTrue(rs2.updateControl.isNoUpdate());
+        assertTrue(rs2.updateControl.getScheduleDelay().isPresent());
+
+        // Completing the savepoint
+        triggers.put(rs.deployment.getStatus().getSavepointTriggerId(), true);
+
+        // This next reconciliation should move on to the next state
+        rs = reconcile(rs.deployment);
+
+        assertEquals(
+                FlinkBlueGreenDeploymentState.ACTIVE_BLUE,
+                rs.reconciledStatus.getBlueGreenState());
+        assertTrue(rs.updateControl.isPatchStatus());
+        assertTrue(rs.updateControl.getScheduleDelay().isPresent());
+        return rs;
+    }
+
     @ParameterizedTest
     @MethodSource("org.apache.flink.kubernetes.operator.TestUtils#flinkVersions")
     public void verifyFailureBeforeTransition(FlinkVersion flinkVersion) throws Exception {