[FLINK-39432] Add option to delete with session running session jobs

Author: gyfora

docs/layouts/shortcodes/generated/dynamic_section.html

@@ -195,7 +195,13 @@
             <td>The interval before a savepoint trigger attempt is marked as unsuccessful.</td>
         </tr>
         <tr>
-            <td><h5>kubernetes.operator.session.block-on-unmanaged-jobs</h5></td>
+            <td><h5>kubernetes.operator.session.deletion.block-on-session-jobs</h5></td>
+            <td style="word-wrap: break-word;">true</td>
+            <td>Boolean</td>
+            <td>Block FlinkDeployment deletion if managed jobs are running in the session cluster.</td>
+        </tr>
+        <tr>
+            <td><h5>kubernetes.operator.session.deletion.block-on-unmanaged-jobs</h5></td>
             <td style="word-wrap: break-word;">true</td>
             <td>Boolean</td>
             <td>Block FlinkDeployment deletion if unmanaged jobs (jobs not managed by FlinkSessionJob resources) are running in the session cluster. Example: Jobs submitted via CLI.</td>

docs/layouts/shortcodes/generated/kubernetes_operator_config_configuration.html

@@ -429,7 +429,13 @@
             <td>The interval before a savepoint trigger attempt is marked as unsuccessful.</td>
         </tr>
         <tr>
-            <td><h5>kubernetes.operator.session.block-on-unmanaged-jobs</h5></td>
+            <td><h5>kubernetes.operator.session.deletion.block-on-session-jobs</h5></td>
+            <td style="word-wrap: break-word;">true</td>
+            <td>Boolean</td>
+            <td>Block FlinkDeployment deletion if managed jobs are running in the session cluster.</td>
+        </tr>
+        <tr>
+            <td><h5>kubernetes.operator.session.deletion.block-on-unmanaged-jobs</h5></td>
             <td style="word-wrap: break-word;">true</td>
             <td>Boolean</td>
             <td>Block FlinkDeployment deletion if unmanaged jobs (jobs not managed by FlinkSessionJob resources) are running in the session cluster. Example: Jobs submitted via CLI.</td>

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/KubernetesOperatorConfigOptions.java

@@ -657,12 +657,21 @@ public static String operatorConfigKey(String key) {
 
     @Documentation.Section(SECTION_DYNAMIC)
     public static final ConfigOption<Boolean> BLOCK_ON_UNMANAGED_JOBS =
-            operatorConfig("session.block-on-unmanaged-jobs")
+            operatorConfig("session.deletion.block-on-unmanaged-jobs")
                     .booleanType()
                     .defaultValue(true)
+                    .withDeprecatedKeys(operatorConfigKey("session.block-on-unmanaged-jobs"))
                     .withDescription(
                             "Block FlinkDeployment deletion if unmanaged jobs (jobs not managed by FlinkSessionJob resources) are running in the session cluster. Example: Jobs submitted via CLI.");
 
+    @Documentation.Section(SECTION_DYNAMIC)
+    public static final ConfigOption<Boolean> BLOCK_ON_SESSION_JOBS =
+            operatorConfig("session.deletion.block-on-session-jobs")
+                    .booleanType()
+                    .defaultValue(true)
+                    .withDescription(
+                            "Block FlinkDeployment deletion if managed jobs are running in the session cluster.");
+
     @Documentation.Section(SECTION_ADVANCED)
     public static final ConfigOption<Duration> REFRESH_CLUSTER_RESOURCE_VIEW =
             operatorConfig("cluster.resource-view.refresh-interval")

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

@@ -160,10 +160,14 @@ Set<JobID> getNonTerminalJobs(FlinkResourceContext<FlinkDeployment> ctx) {
 
     @Override
     public DeleteControl cleanupInternal(FlinkResourceContext<FlinkDeployment> ctx) {
-        Set<FlinkSessionJob> sessionJobs =
-                ctx.getJosdkContext().getSecondaryResources(FlinkSessionJob.class);
+        var sessionJobs = ctx.getJosdkContext().getSecondaryResources(FlinkSessionJob.class);
         var deployment = ctx.getResource();
-        if (!sessionJobs.isEmpty()) {
+
+        boolean blockOnSessionJobs =
+                ctx.getObserveConfig()
+                        .getBoolean(KubernetesOperatorConfigOptions.BLOCK_ON_SESSION_JOBS);
+
+        if (blockOnSessionJobs && !sessionJobs.isEmpty()) {
             var error =
                     String.format(
                             "The session jobs %s should be deleted first",
@@ -188,7 +192,7 @@ public DeleteControl cleanupInternal(FlinkResourceContext<FlinkDeployment> ctx)
         boolean blockOnUnmanagedJobs =
                 ctx.getObserveConfig()
                         .getBoolean(KubernetesOperatorConfigOptions.BLOCK_ON_UNMANAGED_JOBS);
-        if (blockOnUnmanagedJobs) {
+        if (blockOnSessionJobs && blockOnUnmanagedJobs) {
             Set<JobID> nonTerminalJobs = getNonTerminalJobs(ctx);
             if (!nonTerminalJobs.isEmpty()) {
                 var error =

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

@@ -38,6 +38,7 @@
 import io.fabric8.kubernetes.api.model.ObjectMeta;
 import io.fabric8.kubernetes.client.KubernetesClient;
 import io.fabric8.kubernetes.client.server.mock.EnableKubernetesMockClient;
+import io.javaoperatorsdk.operator.api.reconciler.DeleteControl;
 import lombok.Getter;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
@@ -305,4 +306,81 @@ public void testGetNonTerminalJobs() throws Exception {
                 nonTerminalJobsAfterRemoval.size(),
                 "Should have no non-terminal jobs when only terminated jobs exist");
     }
+
+    @Test
+    public void testDeleteSessionWithBlockOnSessionJobsFalse() throws Exception {
+        FlinkDeployment deployment = TestUtils.buildSessionCluster();
+        deployment
+                .getSpec()
+                .getFlinkConfiguration()
+                .put(KubernetesOperatorConfigOptions.BLOCK_ON_SESSION_JOBS.key(), "false");
+
+        reconciler.reconcile(deployment, flinkService.getContext());
+
+        assertEquals(
+                ReconciliationState.DEPLOYED,
+                deployment.getStatus().getReconciliationStatus().getState());
+
+        // Create some running jobs
+        JobID managedJobId1 = new JobID();
+        JobID managedJobId2 = new JobID();
+        JobID unmanagedRunningJobId = new JobID();
+
+        flinkService
+                .listJobs()
+                .add(
+                        Tuple3.of(
+                                null,
+                                new JobStatusMessage(
+                                        managedJobId1,
+                                        "managed-job-1",
+                                        JobStatus.RUNNING,
+                                        System.currentTimeMillis()),
+                                new Configuration()));
+        flinkService
+                .listJobs()
+                .add(
+                        Tuple3.of(
+                                null,
+                                new JobStatusMessage(
+                                        managedJobId2,
+                                        "managed-job-2",
+                                        JobStatus.RUNNING,
+                                        System.currentTimeMillis()),
+                                new Configuration()));
+        flinkService
+                .listJobs()
+                .add(
+                        Tuple3.of(
+                                null,
+                                new JobStatusMessage(
+                                        unmanagedRunningJobId,
+                                        "unmanaged-running-job",
+                                        JobStatus.RUNNING,
+                                        System.currentTimeMillis()),
+                                new Configuration()));
+
+        // Create FlinkSessionJob resources for the managed jobs
+        FlinkSessionJob managedSessionJob1 = TestUtils.buildSessionJob();
+        managedSessionJob1.getMetadata().setName("managed-session-job-1");
+        managedSessionJob1.getStatus().getJobStatus().setJobId(managedJobId1.toHexString());
+        kubernetesClient.resource(managedSessionJob1).createOrReplace();
+
+        FlinkSessionJob managedSessionJob2 = TestUtils.buildSessionJob();
+        managedSessionJob2.getMetadata().setName("managed-session-job-2");
+        managedSessionJob2.getStatus().getJobStatus().setJobId(managedJobId2.toHexString());
+        kubernetesClient.resource(managedSessionJob2).createOrReplace();
+
+        // Test cleanup with BLOCK_ON_SESSION_JOBS=false
+        var context = TestUtils.createContextWithReadyFlinkDeployment(kubernetesClient);
+        var resourceContext = getResourceContext(deployment, context);
+
+        var sessionReconciler = (SessionReconciler) reconciler.getReconciler();
+        DeleteControl deleteControl = sessionReconciler.cleanupInternal(resourceContext);
+
+        // Verify that deletion proceeds immediately despite running jobs
+        assertTrue(
+                deleteControl.isRemoveFinalizer(),
+                "Session should be deleted immediately when BLOCK_ON_SESSION_JOBS is false");
+    }
 }