Fix observed generation

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

@@ -38,6 +38,9 @@ public class FlinkBlueGreenDeploymentStatus {
 
     private JobStatus jobStatus = new JobStatus();
 
+    /** Last observed generation of the FlinkBlueGreenDeployment. */
+    private Long observedGeneration;
+
     /** The state of the blue/green transition. */
     private FlinkBlueGreenDeploymentState blueGreenState;
 

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

@@ -879,6 +879,8 @@ public static UpdateControl<FlinkBlueGreenDeployment> patchStatusUpdateControl(
         }
 
         deploymentStatus.setLastReconciledTimestamp(java.time.Instant.now().toString());
+        deploymentStatus.setObservedGeneration(
+                flinkBlueGreenDeployment.getMetadata().getGeneration());
         flinkBlueGreenDeployment.setStatus(deploymentStatus);
         return UpdateControl.patchStatus(flinkBlueGreenDeployment);
     }

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

@@ -43,19 +43,24 @@ public InitializingBlueStateHandler(BlueGreenDeploymentService deploymentService
     public UpdateControl<FlinkBlueGreenDeployment> handle(BlueGreenContext context) {
         FlinkBlueGreenDeploymentStatus deploymentStatus = context.getDeploymentStatus();
 
-        // Block initial deployment if job.state is SUSPENDED - user must start with RUNNING
+        // Block initial deployment if job.state is SUSPENDED - user must start with RUNNING.
+        // We still record the spec and update observedGeneration so external tools know the
+        // operator has processed this generation.
         var jobSpec = context.getBgDeployment().getSpec().getTemplate().getSpec().getJob();
         if (jobSpec != null && jobSpec.getState() == JobState.SUSPENDED) {
             LOG.info(
                     "Blocking initial deployment '{}' - job.state is SUSPENDED, waiting for RUNNING",
                     context.getBgDeployment().getMetadata().getName());
+            setLastReconciledSpec(context);
             return patchStatusUpdateControl(context, null, JobStatus.SUSPENDED, null);
         }
 
-        // 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
+        // Deploy only if this is the initial deployment (no previous spec exists),
+        // recovering from a failure, or resuming from a suspended initial state
         if (deploymentStatus.getLastReconciledSpec() == null
                 || (deploymentStatus.getJobStatus().getState().equals(JobStatus.FAILING)
+                        && hasSpecChanged(context))
+                || (deploymentStatus.getJobStatus().getState().equals(JobStatus.SUSPENDED)
                         && hasSpecChanged(context))) {
 
             setLastReconciledSpec(context);

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractFlinkResourceReconciler.java

@@ -109,8 +109,14 @@ public void reconcile(FlinkResourceContext<CR> ctx) throws Exception {
         if (reconciliationStatus.isBeforeFirstDeployment()) {
             var spec = cr.getSpec();
 
-            // If the job is submitted in suspend state, no need to reconcile
+            // If the job is created in suspended state, acknowledge the spec without deploying.
+            // We must still update status (observedGeneration, lastReconciledSpec) so that
+            // external tools know the operator has processed this generation.
             if (spec.getJob() != null && spec.getJob().getState().equals(JobState.SUSPENDED)) {
+                LOG.info("Resource created in suspended state, acknowledging without deployment");
+                var deployConfig = ctx.getDeployConfig(spec);
+                ReconciliationUtils.updateStatusForDeployedSpec(cr, deployConfig, clock);
+                statusRecorder.patchAndCacheStatus(cr, ctx.getKubernetesClient());
                 return;
             }
 

flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconcilerTest.java

@@ -1343,6 +1343,46 @@ public void testUpgradeReconciledGeneration() throws Exception {
         assertEquals(2L, deployment.getStatus().getObservedGeneration());
     }
 
+    @Test
+    public void testSuspendedFirstDeploymentSetsObservedGeneration() throws Exception {
+        FlinkDeployment deployment = TestUtils.buildApplicationCluster();
+        deployment.getMetadata().setGeneration(1L);
+        deployment.getSpec().getJob().setState(JobState.SUSPENDED);
+
+        // Verify initial state: nothing has been reconciled yet
+        assertNull(deployment.getStatus().getObservedGeneration());
+        assertTrue(deployment.getStatus().getReconciliationStatus().isBeforeFirstDeployment());
+
+        // Reconcile the suspended deployment
+        reconciler.reconcile(deployment, context);
+
+        // observedGeneration must be set so external tools know we processed the spec
+        assertEquals(1L, deployment.getStatus().getObservedGeneration());
+
+        // lastReconciledSpec must be recorded so isBeforeFirstDeployment() returns false
+        assertFalse(deployment.getStatus().getReconciliationStatus().isBeforeFirstDeployment());
+        var lastReconciledSpec =
+                deployment.getStatus().getReconciliationStatus().deserializeLastReconciledSpec();
+        assertEquals(JobState.SUSPENDED, lastReconciledSpec.getJob().getState());
+
+        // State should be DEPLOYED and marked stable (suspended = stable by convention)
+        assertEquals(
+                ReconciliationState.DEPLOYED,
+                deployment.getStatus().getReconciliationStatus().getState());
+        assertTrue(deployment.getStatus().getReconciliationStatus().isLastReconciledSpecStable());
+
+        // No Flink cluster should have been created
+        assertEquals(0, flinkService.listJobs().size());
+
+        // Later: resuming the job should trigger a real deployment
+        deployment.getSpec().getJob().setState(JobState.RUNNING);
+        deployment.getMetadata().setGeneration(2L);
+        reconciler.reconcile(deployment, context);
+
+        assertEquals(1, flinkService.listJobs().size());
+        assertEquals(2L, deployment.getStatus().getObservedGeneration());
+    }
+
     @ParameterizedTest
     @ValueSource(booleans = {true, false})
     public void testRollbackUpgradeModeHandling(boolean jmStarted) throws Exception {