Clearer comments

Author: schongloo

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

@@ -58,10 +58,10 @@
  *
  * <p>Orchestration Process
  *
- * <p>FlinkBlueGreenDeploymentController.reconcile() ↓ 1. Create BlueGreenContext with current
- * deployment state ↓ 2. Query StateHandlerRegistry for appropriate handler ↓ 3. Delegate to
- * specific StateHandler.handle(context) ↓ 4. StateHandler invokes BlueGreenDeploymentService
- * operations ↓ 5. Return UpdateControl with next reconciliation schedule
+ * <p>FlinkBlueGreenDeploymentController.reconcile() 1. Create BlueGreenContext with current
+ * deployment state 2. Query StateHandlerRegistry for appropriate handler 3. Delegate to specific
+ * StateHandler.handle(context) 4. StateHandler invokes BlueGreenDeploymentService operations 5.
+ * Return UpdateControl with next reconciliation schedule
  */
 @ControllerConfiguration
 public class FlinkBlueGreenDeploymentController implements Reconciler<FlinkBlueGreenDeployment> {

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

@@ -114,8 +114,8 @@ public UpdateControl<FlinkBlueGreenDeployment> checkAndInitiateDeployment(
             if (isFlinkDeploymentReady(currentFlinkDeployment)) {
                 if (specDiff == BlueGreenDiffType.TRANSITION) {
                     if (handleSavepoint(context, currentFlinkDeployment)) {
-                        // This is the only portion where the last reconciled spec is not set,
-                        // so we can reprocess TRANSITION after the savepoint is done
+                        // Spec is intentionally not marked as reconciled here to allow
+                        // reprocessing the TRANSITION once savepoint creation completes
                         var savepointingState = calculateSavepointingState(currentDeploymentType);
                         return patchStatusUpdateControl(context, savepointingState, null)
                                 .rescheduleAfter(getReconciliationReschedInterval(context));
@@ -218,7 +218,8 @@ private Savepoint configureInitialSavepoint(
                 context.getCtxFactory()
                         .getResourceContext(currentFlinkDeployment, context.getJosdkContext());
 
-        // For now UpgradeMode != STATELESS will use a savepoint, originally only SAVEPOINT
+        // Create savepoint for all upgrade modes except STATELESS
+        // (originally only SAVEPOINT mode required savepoints)
         if (isSavepointRequired(context)) {
             String savepointTriggerId = context.getDeploymentStatus().getSavepointTriggerId();
             var savepointFetchResult = fetchSavepointInfo(ctx, savepointTriggerId);
@@ -236,8 +237,8 @@ private Savepoint configureInitialSavepoint(
                     savepointFormatType);
         }
 
-        // The logic below looked for the last checkpoint in case upgradeMode = LAST_STATE
-        // We don't want to rely on last checkpoint for now.
+        // Currently not using last checkpoint recovery for LAST_STATE upgrade mode
+        // This could be re-enabled in the future by uncommenting the logic below
         return null;
 
         //        if (!lookForCheckpoint(context)) {
@@ -315,12 +316,12 @@ private void handleSpecChangesDuringTransition(
             FlinkDeployment incomingFlinkDeployment = transitionState.nextDeployment;
 
             if (diffType != BlueGreenDiffType.IGNORE) {
-                // Apply the new spec changes to the currently active deployment
+                // Apply new spec changes to the deployment currently being transitioned to
                 incomingFlinkDeployment.setSpec(
                         context.getBgDeployment().getSpec().getTemplate().getSpec());
-                // Resetting the abort grace period
+                // Reset abort grace period to allow more time for the updated deployment
                 setAbortTimestamp(context);
-                // Ack'ing the spec change
+                // Mark spec change as reconciled
                 setLastReconciledSpec(context);
                 updateFlinkDeployment(incomingFlinkDeployment, context);
 

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

@@ -41,8 +41,8 @@ public InitializingBlueStateHandler(BlueGreenDeploymentService deploymentService
     public UpdateControl<FlinkBlueGreenDeployment> handle(BlueGreenContext context) {
         FlinkBlueGreenDeploymentStatus deploymentStatus = context.getDeploymentStatus();
 
-        // We only allow a deployment if it's indeed the first (null last spec)
-        // or if we're recovering (failing status) and the spec has changed
+        // Deploy only if this is the initial deployment (no previous spec exists)
+        // or if we're recovering from a failure and the spec has changed since the last attempt
         if (deploymentStatus.getLastReconciledSpec() == null
                 || (deploymentStatus.getJobStatus().getState().equals(JobStatus.FAILING)
                         && hasSpecChanged(context))) {

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

@@ -42,7 +42,7 @@ public UpdateControl<FlinkBlueGreenDeployment> handle(BlueGreenContext context)
         DeploymentType currentType = getCurrentDeploymentType();
         var isSavepointReady = deploymentService.monitorSavepoint(context, currentType);
 
-        // Savepoint complete, continue with the transition
+        // Savepoint creation completed, transition back to active state to continue deployment
         if (isSavepointReady) {
             var nextState =
                     getSupportedState() == FlinkBlueGreenDeploymentState.SAVEPOINTING_BLUE

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

@@ -219,6 +219,13 @@ public static void setAbortTimestamp(BlueGreenContext context) {
 
     // ==================== Savepoint/Checkpoint Operations ====================
 
+    /**
+     * Determines if a savepoint is required based on the deployment's upgrade mode. Currently all
+     * upgrade modes except STATELESS require savepoints.
+     *
+     * @param context the Blue/Green transition context
+     * @return true if savepoint is required, false otherwise
+     */
     public static boolean isSavepointRequired(BlueGreenContext context) {
         UpgradeMode upgradeMode =
                 context.getBgDeployment()
@@ -228,7 +235,8 @@ public static boolean isSavepointRequired(BlueGreenContext context) {
                         .getJob()
                         .getUpgradeMode();
         //        return UpgradeMode.SAVEPOINT == upgradeMode;
-        // For now we're taking savepoints in STATELESS or LAST-STATE
+        // Currently taking savepoints for all modes except STATELESS
+        // (previously only SAVEPOINT mode required savepoints)
         return UpgradeMode.STATELESS != upgradeMode;
     }
 
@@ -289,7 +297,7 @@ public static Savepoint getLastCheckpoint(
                                                 .getJobId()),
                                 resourceContext.getObserveConfig());
 
-        // Alternative action if no checkpoint is available?
+        // Alternative fallback if no checkpoint is available could be implemented here
         if (lastCheckpoint.isEmpty()) {
             throw new IllegalStateException(
                     "Last Checkpoint for Job "
@@ -302,6 +310,17 @@ public static Savepoint getLastCheckpoint(
 
     // ==================== Deployment Preparation Utilities ====================
 
+    /**
+     * Creates a new FlinkDeployment resource for a Blue/Green deployment transition. This method
+     * prepares the deployment with proper metadata, specs, and savepoint configuration.
+     *
+     * @param context the Blue/Green transition context
+     * @param deploymentType the type of deployment (BLUE or GREEN)
+     * @param lastCheckpoint the savepoint/checkpoint to restore from (can be null)
+     * @param isFirstDeployment whether this is the initial deployment
+     * @param bgMeta the metadata of the parent Blue/Green deployment
+     * @return configured FlinkDeployment ready for deployment
+     */
     public static FlinkDeployment prepareFlinkDeployment(
             BlueGreenContext context,
             DeploymentType deploymentType,
@@ -323,7 +342,7 @@ public static FlinkDeployment prepareFlinkDeployment(
                         "spec",
                         FlinkBlueGreenDeploymentSpec.class);
 
-        // The B/G initialSavepointPath is only used in first time deployments
+        // The Blue/Green initialSavepointPath is only used for first-time deployments
         if (isFirstDeployment) {
             String initialSavepointPath =
                     adjustedSpec.getTemplate().getSpec().getJob().getInitialSavepointPath();
@@ -335,11 +354,11 @@ public static FlinkDeployment prepareFlinkDeployment(
                         .getJob()
                         .setInitialSavepointPath(initialSavepointPath);
             } else {
-                LOG.info("Clean start up, no checkpoint/savepoint");
+                LOG.info("Clean startup with no checkpoint/savepoint restoration");
             }
         } else if (lastCheckpoint != null) {
             String location = lastCheckpoint.getLocation().replace("file:", "");
-            LOG.info("Using B/G checkpoint: " + location);
+            LOG.info("Using Blue/Green savepoint/checkpoint: " + location);
             adjustedSpec.getTemplate().getSpec().getJob().setInitialSavepointPath(location);
         }