diff --git a/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/status/FlinkBlueGreenDeploymentStatus.java b/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/status/FlinkBlueGreenDeploymentStatus.java index 104f195efa..5939494cfd 100644 --- a/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/status/FlinkBlueGreenDeploymentStatus.java +++ b/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; diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/bluegreen/BlueGreenDeploymentService.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/bluegreen/BlueGreenDeploymentService.java index ad6f0d9bfa..f742648f0e 100644 --- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/bluegreen/BlueGreenDeploymentService.java +++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/bluegreen/BlueGreenDeploymentService.java @@ -879,6 +879,8 @@ public static UpdateControl patchStatusUpdateControl( } deploymentStatus.setLastReconciledTimestamp(java.time.Instant.now().toString()); + deploymentStatus.setObservedGeneration( + flinkBlueGreenDeployment.getMetadata().getGeneration()); flinkBlueGreenDeployment.setStatus(deploymentStatus); return UpdateControl.patchStatus(flinkBlueGreenDeployment); } diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/bluegreen/handlers/InitializingBlueStateHandler.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/bluegreen/handlers/InitializingBlueStateHandler.java index 8319314fb6..fbb17af188 100644 --- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/bluegreen/handlers/InitializingBlueStateHandler.java +++ b/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 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); diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractFlinkResourceReconciler.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractFlinkResourceReconciler.java index 7f8af3ca88..229f2846e7 100644 --- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractFlinkResourceReconciler.java +++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractFlinkResourceReconciler.java @@ -109,8 +109,14 @@ public void reconcile(FlinkResourceContext 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; } diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconcilerTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconcilerTest.java index a5d87bc946..bd1ec417d9 100644 --- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconcilerTest.java +++ b/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 {