ObserveGeneration and lastStableGeneration for FlinkBlueGreenDeployment

docs/content/docs/custom-resource/reference.md

@@ -361,6 +361,8 @@ This serves as a full reference for FlinkDeployment and FlinkSessionJob custom r
 | ----------| ---- | ---- |
 | jobStatus | org.apache.flink.kubernetes.operator.api.status.JobStatus |  |
 | blueGreenState | org.apache.flink.kubernetes.operator.api.status.FlinkBlueGreenDeploymentState | The state of the blue/green transition. |
+| observedGeneration | java.lang.Long | Last observed generation of the FlinkBlueGreenDeployment. |
+| lastStableGeneration | java.lang.Long | Last generation that reached a stable terminal state. |
 | lastReconciledSpec | java.lang.String | Last reconciled (serialized) deployment spec. |
 | lastReconciledTimestamp | java.lang.String | Timestamp of last reconciliation. |
 | abortTimestamp | java.lang.String | Computed from abortGracePeriodMs, timestamp after which the deployment should be aborted. |

flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/status/FlinkBlueGreenDeploymentStatus.java

@@ -41,6 +41,12 @@ public class FlinkBlueGreenDeploymentStatus {
     /** The state of the blue/green transition. */
     private FlinkBlueGreenDeploymentState blueGreenState;
 
+    /** Last observed generation of the FlinkBlueGreenDeployment. */
+    private Long observedGeneration;
+
+    /** Last generation that reached a stable terminal state. */
+    private Long lastStableGeneration;
+
     /** Last reconciled (serialized) deployment spec. */
     private String lastReconciledSpec;
 

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

@@ -17,6 +17,7 @@
 
 package org.apache.flink.kubernetes.operator.controller;
 
+import org.apache.flink.api.common.JobStatus;
 import org.apache.flink.kubernetes.operator.api.FlinkBlueGreenDeployment;
 import org.apache.flink.kubernetes.operator.api.FlinkDeployment;
 import org.apache.flink.kubernetes.operator.api.status.FlinkBlueGreenDeploymentState;
@@ -45,6 +46,8 @@
 import java.util.ArrayList;
 import java.util.List;
 
+import static org.apache.flink.kubernetes.operator.api.status.FlinkBlueGreenDeploymentState.ACTIVE_BLUE;
+import static org.apache.flink.kubernetes.operator.api.status.FlinkBlueGreenDeploymentState.ACTIVE_GREEN;
 import static org.apache.flink.kubernetes.operator.api.status.FlinkBlueGreenDeploymentState.INITIALIZING_BLUE;
 
 /**
@@ -151,11 +154,61 @@ public UpdateControl<FlinkBlueGreenDeployment> reconcile(
 
             BlueGreenStateHandler handler = handlerRegistry.getHandler(currentState);
             UpdateControl<FlinkBlueGreenDeployment> updateControl = handler.handle(context);
+            updateControl =
+                    syncGenerationStatusIfNeeded(
+                            bgDeployment, context, currentState, deploymentStatus, updateControl);
+
             statusRecorder.patchAndCacheStatus(bgDeployment, josdkContext.getClient());
             return updateControl;
         }
     }
 
+    private static UpdateControl<FlinkBlueGreenDeployment> syncGenerationStatusIfNeeded(
+            FlinkBlueGreenDeployment bgDeployment,
+            BlueGreenContext context,
+            FlinkBlueGreenDeploymentState currentState,
+            FlinkBlueGreenDeploymentStatus deploymentStatus,
+            UpdateControl<FlinkBlueGreenDeployment> updateControl) {
+
+        // Migration/no-op path: kapp may reconcile an existing stable CR without rollout work when
+        // the rendered BlueGreen spec is unchanged, or when only non-spec resources/metadata
+        // changed. Handlers return noUpdate() in those cases, so sync generation status here.
+        if (updateControl.isNoUpdate()) {
+            var needsSync = !BlueGreenDeploymentService.isGenerationObserved(context);
+
+            if (!BlueGreenDeploymentService.isGenerationStable(context)
+                    && isStableBlueGreenJobStatus(currentState, deploymentStatus)) {
+                deploymentStatus.setLastStableGeneration(
+                        bgDeployment.getMetadata().getGeneration());
+                needsSync = true;
+            }
+
+            if (needsSync) {
+                var statusUpdateControl =
+                        BlueGreenDeploymentService.patchStatusUpdateControl(
+                                context, null, null, null);
+                updateControl.getScheduleDelay().ifPresent(statusUpdateControl::rescheduleAfter);
+                return statusUpdateControl;
+            }
+        }
+
+        return updateControl;
+    }
+
+    private static boolean isStableBlueGreenJobStatus(
+            FlinkBlueGreenDeploymentState currentState,
+            FlinkBlueGreenDeploymentStatus deploymentStatus) {
+        // Suspension is represented as ACTIVE_* with parent jobStatus SUSPENDED, not as a separate
+        // BlueGreen state. Transitional or failing states must continue waiting for finalization.
+        if (currentState != ACTIVE_BLUE && currentState != ACTIVE_GREEN) {
+            return false;
+        }
+
+        var jobStatus = deploymentStatus.getJobStatus();
+        var jobState = jobStatus == null ? null : jobStatus.getState();
+        return jobState == JobStatus.RUNNING || jobState == JobStatus.SUSPENDED;
+    }
+
     public static void logAndThrow(String message) {
         throw new RuntimeException(message);
     }

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

@@ -859,6 +859,18 @@ private static void resetTransitionMarkers(FlinkBlueGreenDeploymentStatus status
         status.setSavepointTriggerId(null);
     }
 
+    public static boolean isGenerationObserved(BlueGreenContext context) {
+        return Objects.equals(
+                context.getDeploymentStatus().getObservedGeneration(),
+                context.getBgDeployment().getMetadata().getGeneration());
+    }
+
+    public static boolean isGenerationStable(BlueGreenContext context) {
+        return Objects.equals(
+                context.getDeploymentStatus().getLastStableGeneration(),
+                context.getBgDeployment().getMetadata().getGeneration());
+    }
+
     public static UpdateControl<FlinkBlueGreenDeployment> patchStatusUpdateControl(
             BlueGreenContext context,
             FlinkBlueGreenDeploymentState deploymentState,
@@ -884,6 +896,13 @@ public static UpdateControl<FlinkBlueGreenDeployment> patchStatusUpdateControl(
             deploymentStatus.setError(null);
         }
 
+        if (jobState == JobStatus.RUNNING || jobState == JobStatus.SUSPENDED) {
+            deploymentStatus.setLastStableGeneration(
+                    flinkBlueGreenDeployment.getMetadata().getGeneration());
+        }
+
+        deploymentStatus.setObservedGeneration(
+                flinkBlueGreenDeployment.getMetadata().getGeneration());
         deploymentStatus.setLastReconciledTimestamp(java.time.Instant.now().toString());
         flinkBlueGreenDeployment.setStatus(deploymentStatus);
         return UpdateControl.patchStatus(flinkBlueGreenDeployment);

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

@@ -52,6 +52,7 @@
 import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
 import org.jetbrains.annotations.NotNull;
 import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.Arguments;
 import org.junit.jupiter.params.provider.MethodSource;
@@ -128,6 +129,118 @@ public void verifyBasicDeployment(FlinkVersion flinkVersion, String initialSavep
         executeBasicDeployment(flinkVersion, blueGreenDeployment, true, initialSavepointPath);
     }
 
+    @Test
+    public void verifyObservedGenerationUpdatedWhenNewSpecObserved() throws Exception {
+        var blueGreenDeployment =
+                buildSessionCluster(
+                        TEST_DEPLOYMENT_NAME,
+                        TEST_NAMESPACE,
+                        FlinkVersion.v1_20,
+                        null,
+                        UpgradeMode.STATELESS);
+        var rs = executeBasicDeployment(FlinkVersion.v1_20, blueGreenDeployment, false, null);
+
+        Long previousGeneration = rs.deployment.getMetadata().getGeneration();
+        assertNotNull(previousGeneration);
+        assertEquals(previousGeneration, rs.reconciledStatus.getObservedGeneration());
+        assertEquals(previousGeneration, rs.reconciledStatus.getLastStableGeneration());
+
+        simulateSpecChange(rs.deployment, UUID.randomUUID().toString());
+        Long newGeneration = previousGeneration + 1;
+        rs.deployment.getMetadata().setGeneration(newGeneration);
+        kubernetesClient.resource(rs.deployment).createOrReplace();
+
+        rs = reconcile(rs.deployment);
+
+        assertEquals(newGeneration, rs.reconciledStatus.getObservedGeneration());
+        assertEquals(previousGeneration, rs.reconciledStatus.getLastStableGeneration());
+        assertEquals(
+                FlinkBlueGreenDeploymentState.TRANSITIONING_TO_GREEN,
+                rs.reconciledStatus.getBlueGreenState());
+        assertEquals(JobStatus.RECONCILING, rs.reconciledStatus.getJobStatus().getState());
+    }
+
+    @Test
+    public void verifyGenerationStatusSyncedForExistingStableStatus() throws Exception {
+        var blueGreenDeployment =
+                buildSessionCluster(
+                        TEST_DEPLOYMENT_NAME,
+                        TEST_NAMESPACE,
+                        FlinkVersion.v1_20,
+                        null,
+                        UpgradeMode.STATELESS);
+        var rs = executeBasicDeployment(FlinkVersion.v1_20, blueGreenDeployment, false, null);
+
+        rs.deployment.getStatus().setObservedGeneration(null);
+        rs.deployment.getStatus().setLastStableGeneration(null);
+
+        rs = reconcile(rs.deployment);
+
+        assertTrue(rs.updateControl.isPatchStatus());
+        assertEquals(
+                rs.deployment.getMetadata().getGeneration(),
+                rs.reconciledStatus.getObservedGeneration());
+        assertEquals(
+                rs.deployment.getMetadata().getGeneration(),
+                rs.reconciledStatus.getLastStableGeneration());
+        assertEquals(
+                FlinkBlueGreenDeploymentState.ACTIVE_BLUE, rs.reconciledStatus.getBlueGreenState());
+        assertEquals(JobStatus.RUNNING, rs.reconciledStatus.getJobStatus().getState());
+    }
+
+    @Test
+    public void verifyStableGenerationNotUpdatedWhenTransitionFailsBeforeStart() throws Exception {
+        var blueGreenDeployment =
+                buildSessionCluster(
+                        TEST_DEPLOYMENT_NAME,
+                        TEST_NAMESPACE,
+                        FlinkVersion.v1_20,
+                        null,
+                        UpgradeMode.STATELESS);
+        var rs = executeBasicDeployment(FlinkVersion.v1_20, blueGreenDeployment, false, null);
+
+        Long previousGeneration = rs.deployment.getMetadata().getGeneration();
+        simulateSpecChange(rs.deployment, UUID.randomUUID().toString());
+        Long newGeneration = previousGeneration + 1;
+        rs.deployment.getMetadata().setGeneration(newGeneration);
+        kubernetesClient.resource(rs.deployment).createOrReplace();
+        simulateJobFailure(getFlinkDeployments().get(0));
+
+        rs = reconcile(rs.deployment);
+
+        assertEquals(newGeneration, rs.reconciledStatus.getObservedGeneration());
+        assertEquals(previousGeneration, rs.reconciledStatus.getLastStableGeneration());
+        assertEquals(JobStatus.FAILING, rs.reconciledStatus.getJobStatus().getState());
+        assertEquals(
+                FlinkBlueGreenDeploymentState.ACTIVE_BLUE, rs.reconciledStatus.getBlueGreenState());
+    }
+
+    @Test
+    public void verifyStableGenerationNotUpdatedDuringSavepointHandoff() throws Exception {
+        var blueGreenDeployment =
+                buildSessionCluster(
+                        TEST_DEPLOYMENT_NAME,
+                        TEST_NAMESPACE,
+                        FlinkVersion.v1_20,
+                        null,
+                        UpgradeMode.SAVEPOINT);
+        var rs = executeBasicDeployment(FlinkVersion.v1_20, blueGreenDeployment, false, null);
+
+        Long previousGeneration = rs.deployment.getMetadata().getGeneration();
+        simulateSpecChange(rs.deployment, UUID.randomUUID().toString());
+        Long newGeneration = previousGeneration + 1;
+        rs.deployment.getMetadata().setGeneration(newGeneration);
+        kubernetesClient.resource(rs.deployment).createOrReplace();
+
+        rs = handleSavepoint(rs);
+
+        assertEquals(newGeneration, rs.reconciledStatus.getObservedGeneration());
+        assertEquals(previousGeneration, rs.reconciledStatus.getLastStableGeneration());
+        assertEquals(
+                FlinkBlueGreenDeploymentState.ACTIVE_BLUE, rs.reconciledStatus.getBlueGreenState());
+        assertEquals(JobStatus.RUNNING, rs.reconciledStatus.getJobStatus().getState());
+    }
+
     @ParameterizedTest
     @MethodSource("flinkVersionsAndSavepointPathsAndUpgradeModes")
     public void verifyBasicTransition(
@@ -1181,6 +1294,12 @@ public void verifyInitialSuspendedIsBlockedThenDeploysOnRunning(FlinkVersion fli
                 JobStatus.SUSPENDED,
                 rs.reconciledStatus.getJobStatus().getState(),
                 "Job status should be SUSPENDED when initial deployment is blocked");
+        assertEquals(
+                rs.deployment.getMetadata().getGeneration(),
+                rs.reconciledStatus.getObservedGeneration());
+        assertEquals(
+                rs.deployment.getMetadata().getGeneration(),
+                rs.reconciledStatus.getLastStableGeneration());
 
         // Flip to RUNNING and reconcile again
         bg = rs.deployment;
@@ -1507,6 +1626,12 @@ private void assertFinalized(
                 rs.reconciledStatus.getLastReconciledSpec());
         assertEquals(expectedBGDeploymentState, rs.reconciledStatus.getBlueGreenState());
         assertEquals(JobStatus.RUNNING, rs.reconciledStatus.getJobStatus().getState());
+        assertEquals(
+                rs.deployment.getMetadata().getGeneration(),
+                rs.reconciledStatus.getObservedGeneration());
+        assertEquals(
+                rs.deployment.getMetadata().getGeneration(),
+                rs.reconciledStatus.getLastStableGeneration());
         assertEquals(0, instantStrToMillis(rs.reconciledStatus.getDeploymentReadyTimestamp()));
         assertEquals(0, instantStrToMillis(rs.reconciledStatus.getAbortTimestamp()));
 
@@ -1796,6 +1921,7 @@ private static FlinkBlueGreenDeployment buildSessionCluster(
                         .withCreationTimestamp(Instant.now().toString())
                         .withUid(UUID.randomUUID().toString())
                         .withResourceVersion("1")
+                        .withGeneration(1L)
                         .build());
         var bgDeploymentSpec = getTestFlinkDeploymentSpec(version);
 

helm/flink-kubernetes-operator/crds/flinkbluegreendeployments.flink.apache.org-v1.yml

@@ -10938,6 +10938,10 @@ spec:
                 type: "string"
               lastReconciledTimestamp:
                 type: "string"
+              lastStableGeneration:
+                type: "integer"
+              observedGeneration:
+                type: "integer"
               savepointTriggerId:
                 type: "string"
             type: "object"