[FLINK-39165] Prevent concurrent FlinkSessionJob / FlinkDeployment update

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

@@ -27,6 +27,7 @@
 import org.apache.flink.kubernetes.operator.api.spec.FlinkDeploymentSpec;
 import org.apache.flink.kubernetes.operator.api.status.FlinkDeploymentStatus;
 import org.apache.flink.kubernetes.operator.api.status.JobManagerDeploymentStatus;
+import org.apache.flink.kubernetes.operator.api.status.ReconciliationState;
 import org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions;
 import org.apache.flink.kubernetes.operator.controller.FlinkResourceContext;
 import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
@@ -42,6 +43,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
 import java.util.stream.Collectors;
@@ -64,9 +66,38 @@ public SessionReconciler(
 
     @Override
     protected boolean readyToReconcile(FlinkResourceContext<FlinkDeployment> ctx) {
+        var deployment = ctx.getResource();
+        // Only check when there is a pending spec change on the deployment itself
+        if (!Objects.equals(
+                deployment.getMetadata().getGeneration(),
+                deployment.getStatus().getObservedGeneration())) {
+            var deploymentName = deployment.getMetadata().getName();
+            boolean hasTransitioningSessionJob =
+                    ctx.getJosdkContext().getSecondaryResources(FlinkSessionJob.class).stream()
+                            .filter(job -> deploymentName.equals(job.getSpec().getDeploymentName()))
+                            .anyMatch(SessionReconciler::isSessionJobTransitioning);
+            if (hasTransitioningSessionJob) {
+                LOG.info(
+                        "Deferring session cluster upgrade: associated session jobs are being upgraded");
+                return false;
+            }
+        }
         return true;
     }
 
+    private static boolean isSessionJobTransitioning(FlinkSessionJob job) {
+        var reconStatus = job.getStatus().getReconciliationStatus();
+        // New session jobs state is UPGRADING by default before first deployment. Treating them as
+        // transitioning would deadlock: the deployment can't
+        // start because it waits for the job, and the job can't start because it waits for the
+        // cluster.
+        if (reconStatus.isBeforeFirstDeployment()) {
+            return false;
+        }
+        var state = reconStatus.getState();
+        return state == ReconciliationState.UPGRADING || state == ReconciliationState.ROLLING_BACK;
+    }
+
     @Override
     protected boolean reconcileSpecChange(
             DiffType diffType,

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/SessionJobReconciler.java

@@ -27,6 +27,7 @@
 import org.apache.flink.kubernetes.operator.api.spec.JobState;
 import org.apache.flink.kubernetes.operator.api.status.FlinkSessionJobStatus;
 import org.apache.flink.kubernetes.operator.api.status.JobManagerDeploymentStatus;
+import org.apache.flink.kubernetes.operator.api.status.ReconciliationState;
 import org.apache.flink.kubernetes.operator.autoscaler.KubernetesJobAutoScalerContext;
 import org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions;
 import org.apache.flink.kubernetes.operator.controller.FlinkResourceContext;
@@ -43,6 +44,7 @@
 import org.slf4j.LoggerFactory;
 
 import java.time.Instant;
+import java.util.Objects;
 import java.util.Optional;
 
 /** The reconciler for the {@link FlinkSessionJob}. */
@@ -60,8 +62,9 @@ public SessionJobReconciler(
 
     @Override
     public boolean readyToReconcile(FlinkResourceContext<FlinkSessionJob> ctx) {
-        return sessionClusterReady(
-                        ctx.getJosdkContext().getSecondaryResource(FlinkDeployment.class))
+        var flinkDeploymentOpt = ctx.getJosdkContext().getSecondaryResource(FlinkDeployment.class);
+        return sessionClusterReady(flinkDeploymentOpt)
+                && noDeploymentChangesPending(flinkDeploymentOpt)
                 && super.readyToReconcile(ctx);
     }
 
@@ -178,20 +181,59 @@ public DeleteControl cleanupInternal(FlinkResourceContext<FlinkSessionJob> ctx)
     }
 
     public static boolean sessionClusterReady(Optional<FlinkDeployment> flinkDeploymentOpt) {
-        if (flinkDeploymentOpt.isPresent()) {
-            var flinkdep = flinkDeploymentOpt.get();
-            var jobmanagerDeploymentStatus = flinkdep.getStatus().getJobManagerDeploymentStatus();
-            if (jobmanagerDeploymentStatus != JobManagerDeploymentStatus.READY) {
-                LOG.info(
-                        "Session cluster deployment is in {} status, not ready for serve",
-                        jobmanagerDeploymentStatus);
-                return false;
-            } else {
-                return true;
-            }
-        } else {
+        if (flinkDeploymentOpt.isEmpty()) {
             LOG.warn("Session cluster deployment is not found");
             return false;
         }
+        var flinkdep = flinkDeploymentOpt.get();
+        var jobmanagerDeploymentStatus = flinkdep.getStatus().getJobManagerDeploymentStatus();
+        if (jobmanagerDeploymentStatus != JobManagerDeploymentStatus.READY) {
+            LOG.info(
+                    "Session cluster deployment is in {} status, not ready for serve",
+                    jobmanagerDeploymentStatus);
+            return false;
+        }
+
+        // Block while FlinkDeployment is in a transitional reconciliation state
+        var reconciliationState = flinkdep.getStatus().getReconciliationStatus().getState();
+        if (reconciliationState != ReconciliationState.DEPLOYED
+                && reconciliationState != ReconciliationState.ROLLED_BACK) {
+            LOG.info(
+                    "Session cluster deployment reconciliation state is {}, not ready for serve",
+                    reconciliationState);
+            return false;
+        }
+
+        return true;
+    }
+
+    /**
+     * Checks that the FlinkDeployment has no pending spec changes
+     *
+     * <p>When they differ, the deployment spec was changed but the operator hasn't acted yet. The
+     * cluster may still appear healthy (JM READY, state DEPLOYED), but it is about to be updated
+     * (e.g. deleted and recreated). Allowing a session job to start upgrading in this window would
+     * risk the savepoint being destroyed during the cluster rebuild.
+     *
+     * <p>This check is only used in {@link #readyToReconcile}, not in {@link #sessionClusterReady},
+     * so it does not block cleanup or FlinkService creation.
+     */
+    private static boolean noDeploymentChangesPending(
+            Optional<FlinkDeployment> flinkDeploymentOpt) {
+        if (flinkDeploymentOpt.isEmpty()) {
+            return false;
+        }
+        var flinkdep = flinkDeploymentOpt.get();
+        if (!Objects.equals(
+                flinkdep.getMetadata().getGeneration(),
+                flinkdep.getStatus().getObservedGeneration())) {
+            LOG.info(
+                    "Session cluster deployment has pending spec changes "
+                            + "(generation={}, observedGeneration={}), not ready for sync",
+                    flinkdep.getMetadata().getGeneration(),
+                    flinkdep.getStatus().getObservedGeneration());
+            return false;
+        }
+        return true;
     }
 }

flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestUtils.java

@@ -26,6 +26,7 @@
 import org.apache.flink.kubernetes.operator.api.spec.JobReference;
 import org.apache.flink.kubernetes.operator.api.spec.UpgradeMode;
 import org.apache.flink.kubernetes.operator.api.status.JobManagerDeploymentStatus;
+import org.apache.flink.kubernetes.operator.api.status.ReconciliationState;
 import org.apache.flink.kubernetes.operator.api.utils.BaseTestUtils;
 import org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions;
 import org.apache.flink.kubernetes.operator.health.CanaryResourceManager;
@@ -226,6 +227,12 @@ public Optional<T> getSecondaryResource(Class expectedType, String eventSourceNa
                 session.getStatus()
                         .getReconciliationStatus()
                         .serializeAndSetLastReconciledSpec(session.getSpec(), session);
+                // ReconciliationStatus defaults to UPGRADING. Ready session
+                // cluster must be in DEPLOYED state, otherwise sessionClusterReady() will
+                // reject it due to the reconciliation state check.
+                session.getStatus()
+                        .getReconciliationStatus()
+                        .setState(ReconciliationState.DEPLOYED);
                 return (Optional<T>) Optional.of(session);
             }
 
@@ -498,17 +505,28 @@ public Optional<RetryInfo> getRetryInfo() {
         @Override
         public <T1> Set<T1> getSecondaryResources(Class<T1> aClass) {
             // TODO: improve this, even if we only support FlinkDeployment as a secondary resource
+            KubernetesClient client = getClient();
+            if (client == null) {
+                return Set.of();
+            }
             if (aClass.getSimpleName().equals(FlinkDeployment.class.getSimpleName())) {
-                KubernetesClient client = getClient();
                 var hasMetadata =
                         new HashSet<>(
                                 client.resources(FlinkDeployment.class)
                                         .inAnyNamespace()
                                         .list()
                                         .getItems());
                 return (Set<T1>) hasMetadata;
+            } else if (aClass.getSimpleName().equals(FlinkSessionJob.class.getSimpleName())) {
+                var hasMetadata =
+                        new HashSet<>(
+                                client.resources(FlinkSessionJob.class)
+                                        .inAnyNamespace()
+                                        .list()
+                                        .getItems());
+                return (Set<T1>) hasMetadata;
             } else {
-                return null;
+                return Set.of();
             }
         }
 

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

@@ -305,4 +305,90 @@ public void testGetNonTerminalJobs() throws Exception {
                 nonTerminalJobsAfterRemoval.size(),
                 "Should have no non-terminal jobs when only terminated jobs exist");
     }
+
+    @Test
+    public void testUpgradeDeferredWhenSessionJobUpgrading() throws Exception {
+        FlinkDeployment deployment = TestUtils.buildSessionCluster();
+        reconciler.reconcile(deployment, flinkService.getContext());
+        assertEquals(
+                ReconciliationState.DEPLOYED,
+                deployment.getStatus().getReconciliationStatus().getState());
+
+        FlinkSessionJob upgradingJob = TestUtils.buildSessionJob();
+        upgradingJob
+                .getStatus()
+                .getReconciliationStatus()
+                .serializeAndSetLastReconciledSpec(upgradingJob.getSpec(), upgradingJob);
+        upgradingJob.getStatus().getReconciliationStatus().setState(ReconciliationState.UPGRADING);
+        kubernetesClient.resource(upgradingJob).createOrReplace();
+
+        deployment.getMetadata().setGeneration(2L);
+        deployment.getSpec().setRestartNonce(2L);
+
+        // Reconcile should be deferred because session job is upgrading
+        reconciler.reconcile(deployment, flinkService.getContext());
+
+        // Should still be DEPLOYED (not UPGRADING) — upgrade was deferred
+        assertEquals(
+                ReconciliationState.DEPLOYED,
+                deployment.getStatus().getReconciliationStatus().getState());
+    }
+
+    @Test
+    public void testUpgradeProceedsWhenSessionJobDeployed() throws Exception {
+        FlinkDeployment deployment = TestUtils.buildSessionCluster();
+
+        reconciler.reconcile(deployment, flinkService.getContext());
+        assertEquals(
+                ReconciliationState.DEPLOYED,
+                deployment.getStatus().getReconciliationStatus().getState());
+
+        FlinkSessionJob deployedJob = TestUtils.buildSessionJob();
+        deployedJob
+                .getStatus()
+                .getReconciliationStatus()
+                .serializeAndSetLastReconciledSpec(deployedJob.getSpec(), deployedJob);
+        deployedJob.getStatus().getReconciliationStatus().setState(ReconciliationState.DEPLOYED);
+        kubernetesClient.resource(deployedJob).createOrReplace();
+
+        // Simulate a pending spec change on the deployment
+        deployment.getMetadata().setGeneration(2L);
+        deployment.getSpec().setRestartNonce(456L);
+
+        // Reconcile should proceed because session job is not upgrading
+        reconciler.reconcile(deployment, flinkService.getContext());
+
+        // Should be DEPLOYED with the new spec (upgrade went through)
+        assertEquals(
+                ReconciliationState.DEPLOYED,
+                deployment.getStatus().getReconciliationStatus().getState());
+        assertEquals(
+                456L,
+                deployment
+                        .getStatus()
+                        .getReconciliationStatus()
+                        .deserializeLastReconciledSpec()
+                        .getRestartNonce());
+    }
+
+    @Test
+    public void testDeploymentCleanupWaitsForSessionJobs() throws Exception {
+        FlinkDeployment deployment = TestUtils.buildSessionCluster();
+        reconciler.reconcile(deployment, flinkService.getContext());
+
+        FlinkSessionJob job = TestUtils.buildSessionJob();
+        job.getStatus()
+                .getReconciliationStatus()
+                .serializeAndSetLastReconciledSpec(job.getSpec(), job);
+        job.getStatus().getReconciliationStatus().setState(ReconciliationState.DEPLOYED);
+        kubernetesClient.resource(job).createOrReplace();
+
+        var deleteControl = reconciler.cleanup(deployment, flinkService.getContext());
+        assertFalse(deleteControl.isRemoveFinalizer());
+
+        kubernetesClient.resource(job).delete();
+
+        deleteControl = reconciler.cleanup(deployment, flinkService.getContext());
+        assertTrue(deleteControl.isRemoveFinalizer());
+    }
 }