Consolidated and organized the B/G utility methods.

Author: schongloo

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

@@ -46,17 +46,17 @@
 import static org.apache.flink.kubernetes.operator.controller.bluegreen.BlueGreenKubernetesService.isFlinkDeploymentReady;
 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.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.prepareFlinkDeployment;
-import static org.apache.flink.kubernetes.operator.utils.bluegreen.BlueGreenSpecUtils.setLastReconciledSpec;
-import static org.apache.flink.kubernetes.operator.utils.bluegreen.BlueGreenSpecUtils.triggerSavepoint;
+import static org.apache.flink.kubernetes.operator.utils.bluegreen.BlueGreenUtils.fetchSavepointInfo;
 import static org.apache.flink.kubernetes.operator.utils.bluegreen.BlueGreenUtils.getReconciliationReschedInterval;
+import static org.apache.flink.kubernetes.operator.utils.bluegreen.BlueGreenUtils.getSpecDiff;
+import static org.apache.flink.kubernetes.operator.utils.bluegreen.BlueGreenUtils.hasSpecChanged;
 import static org.apache.flink.kubernetes.operator.utils.bluegreen.BlueGreenUtils.instantStrToMillis;
+import static org.apache.flink.kubernetes.operator.utils.bluegreen.BlueGreenUtils.isSavepointRequired;
 import static org.apache.flink.kubernetes.operator.utils.bluegreen.BlueGreenUtils.millisToInstantStr;
+import static org.apache.flink.kubernetes.operator.utils.bluegreen.BlueGreenUtils.prepareFlinkDeployment;
 import static org.apache.flink.kubernetes.operator.utils.bluegreen.BlueGreenUtils.setAbortTimestamp;
+import static org.apache.flink.kubernetes.operator.utils.bluegreen.BlueGreenUtils.setLastReconciledSpec;
+import static org.apache.flink.kubernetes.operator.utils.bluegreen.BlueGreenUtils.triggerSavepoint;
 
 /** Consolidated service for all Blue/Green deployment operations. */
 public class BlueGreenDeploymentService {
@@ -86,11 +86,7 @@ public UpdateControl<FlinkBlueGreenDeployment> initiateDeployment(
 
         FlinkDeployment flinkDeployment =
                 prepareFlinkDeployment(
-                        context.getBgDeployment(),
-                        nextDeploymentType,
-                        lastCheckpoint,
-                        isFirstDeployment,
-                        bgMeta);
+                        context, nextDeploymentType, lastCheckpoint, isFirstDeployment, bgMeta);
 
         deployCluster(context, flinkDeployment);
 

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

@@ -24,10 +24,12 @@
 import org.apache.flink.kubernetes.operator.api.status.FlinkBlueGreenDeploymentStatus;
 import org.apache.flink.kubernetes.operator.controller.bluegreen.BlueGreenContext;
 import org.apache.flink.kubernetes.operator.controller.bluegreen.BlueGreenDeploymentService;
-import org.apache.flink.kubernetes.operator.utils.bluegreen.BlueGreenSpecUtils;
 
 import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
 
+import static org.apache.flink.kubernetes.operator.utils.bluegreen.BlueGreenUtils.hasSpecChanged;
+import static org.apache.flink.kubernetes.operator.utils.bluegreen.BlueGreenUtils.setLastReconciledSpec;
+
 /** State handler for the INITIALIZING_BLUE state. */
 public class InitializingBlueStateHandler extends AbstractBlueGreenStateHandler {
 
@@ -43,9 +45,9 @@ public UpdateControl<FlinkBlueGreenDeployment> handle(BlueGreenContext context)
         // or if we're recovering (failing status) and the spec has changed
         if (deploymentStatus.getLastReconciledSpec() == null
                 || (deploymentStatus.getJobStatus().getState().equals(JobStatus.FAILING)
-                        && BlueGreenSpecUtils.hasSpecChanged(context))) {
+                        && hasSpecChanged(context))) {
 
-            BlueGreenSpecUtils.setLastReconciledSpec(context);
+            setLastReconciledSpec(context);
             return deploymentService.initiateDeployment(
                     context,
                     DeploymentType.BLUE,