[FLINK-39891] Add option to cancel orphaned session jobs on submit

docs/layouts/shortcodes/generated/dynamic_section.html

@@ -194,6 +194,12 @@
             <td>Duration</td>
             <td>The interval before a savepoint trigger attempt is marked as unsuccessful.</td>
         </tr>
+        <tr>
+            <td><h5>kubernetes.operator.session-job.cancel-existing-on-submit</h5></td>
+            <td style="word-wrap: break-word;">false</td>
+            <td>Boolean</td>
+            <td>When enabled, before submitting a FlinkSessionJob the operator cancels any non-terminal job already running on the session cluster that has the same Flink job name (pipeline.name, which defaults to the resource name). This prevents duplicate jobs when a previous job was orphaned on the session cluster (for example its FlinkSessionJob was deleted while the cluster was unreachable) and the resource is later recreated. Applies only to new submissions, not to deployment retries.</td>
+        </tr>
         <tr>
             <td><h5>kubernetes.operator.session.deletion.block-on-session-jobs</h5></td>
             <td style="word-wrap: break-word;">true</td>

docs/layouts/shortcodes/generated/kubernetes_operator_config_configuration.html

@@ -428,6 +428,12 @@
             <td>Duration</td>
             <td>The interval before a savepoint trigger attempt is marked as unsuccessful.</td>
         </tr>
+        <tr>
+            <td><h5>kubernetes.operator.session-job.cancel-existing-on-submit</h5></td>
+            <td style="word-wrap: break-word;">false</td>
+            <td>Boolean</td>
+            <td>When enabled, before submitting a FlinkSessionJob the operator cancels any non-terminal job already running on the session cluster that has the same Flink job name (pipeline.name, which defaults to the resource name). This prevents duplicate jobs when a previous job was orphaned on the session cluster (for example its FlinkSessionJob was deleted while the cluster was unreachable) and the resource is later recreated. Applies only to new submissions, not to deployment retries.</td>
+        </tr>
         <tr>
             <td><h5>kubernetes.operator.session.deletion.block-on-session-jobs</h5></td>
             <td style="word-wrap: break-word;">true</td>

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

@@ -180,6 +180,21 @@ public static String operatorConfigKey(String key) {
                     .withDescription(
                             "Whether to enable inplace scaling for Flink 1.18+ using the resource requirements API. On failure or earlier Flink versions it falls back to regular full redeployment.");
 
+    @Documentation.Section(SECTION_DYNAMIC)
+    public static final ConfigOption<Boolean> OPERATOR_SESSION_JOB_CANCEL_EXISTING_ON_SUBMIT =
+            operatorConfig("session-job.cancel-existing-on-submit")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription(
+                            "When enabled, before submitting a FlinkSessionJob the operator cancels"
+                                    + " any non-terminal job already running on the session cluster"
+                                    + " that has the same Flink job name (pipeline.name, which"
+                                    + " defaults to the resource name). This prevents duplicate jobs"
+                                    + " when a previous job was orphaned on the session cluster (for"
+                                    + " example its FlinkSessionJob was deleted while the cluster was"
+                                    + " unreachable) and the resource is later recreated. Applies"
+                                    + " only to new submissions, not to deployment retries.");
+
     @Documentation.Section(SECTION_ADVANCED)
     public static final ConfigOption<Boolean> OPERATOR_DYNAMIC_CONFIG_ENABLED =
             operatorConfig("dynamic.config.enabled")

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

@@ -21,6 +21,7 @@
 import org.apache.flink.api.common.JobStatus;
 import org.apache.flink.autoscaler.JobAutoScaler;
 import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.PipelineOptions;
 import org.apache.flink.kubernetes.operator.api.FlinkDeployment;
 import org.apache.flink.kubernetes.operator.api.FlinkSessionJob;
 import org.apache.flink.kubernetes.operator.api.lifecycle.ResourceLifecycleState;
@@ -101,6 +102,13 @@ public void deploy(
             statusRecorder.patchAndCacheStatus(ctx.getResource(), ctx.getKubernetesClient());
         }
 
+        if (!reuseJobId
+                && deployConfig.get(
+                        KubernetesOperatorConfigOptions
+                                .OPERATOR_SESSION_JOB_CANCEL_EXISTING_ON_SUBMIT)) {
+            cancelExistingJobsWithSameName(ctx, deployConfig);
+        }
+
         ctx.getFlinkService()
                 .submitJobToSessionCluster(
                         ctx.getResource().getMetadata(),
@@ -110,6 +118,23 @@ public void deploy(
                         savepoint.orElse(null));
     }
 
+    private void cancelExistingJobsWithSameName(
+            FlinkResourceContext<FlinkSessionJob> ctx, Configuration deployConfig)
+            throws Exception {
+        var jobName =
+                deployConfig
+                        .getOptional(PipelineOptions.NAME)
+                        .orElseGet(() -> ctx.getResource().getMetadata().getName());
+        var cancelledJobIds = ctx.getFlinkService().cancelRunningJobs(jobName, deployConfig);
+        if (!cancelledJobIds.isEmpty()) {
+            LOG.warn(
+                    "Cancelled {} orphaned job(s) named '{}' on the session cluster before submission: {}",
+                    cancelledJobIds.size(),
+                    jobName,
+                    cancelledJobIds);
+        }
+    }
+
     @Override
     protected boolean cancelJob(FlinkResourceContext<FlinkSessionJob> ctx, SuspendMode suspendMode)
             throws Exception {

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java

@@ -139,12 +139,15 @@
 import java.nio.file.Paths;
 import java.time.Duration;
 import java.util.Arrays;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
+import java.util.Set;
 import java.util.concurrent.Callable;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
@@ -294,23 +297,61 @@ protected SocketAddress getSocketAddress(RestClusterClient<String> clusterClient
     public Optional<JobStatusMessage> getJobStatus(Configuration conf, JobID jobId)
             throws Exception {
         try (var clusterClient = getClusterClient(conf)) {
-            return clusterClient
-                    .sendRequest(
-                            JobsOverviewHeaders.getInstance(),
-                            EmptyMessageParameters.getInstance(),
-                            EmptyRequestBody.getInstance())
-                    .thenApply(
-                            mjd -> {
-                                if (mjd.getJobs() == null) {
-                                    return Optional.<JobStatusMessage>empty();
-                                }
-                                return mjd.getJobs().stream()
-                                        .filter(jd -> jd.getJobId().equals(jobId))
-                                        .findAny()
-                                        .map(JobStatusUtils::toJobStatusMessage);
-                            })
-                    .get(operatorConfig.getFlinkClientTimeout().toSeconds(), TimeUnit.SECONDS);
+            return listJobs(clusterClient).stream()
+                    .filter(job -> job.getJobId().equals(jobId))
+                    .findAny();
+        }
+    }
+
+    @Override
+    public Set<JobID> cancelRunningJobs(String jobName, Configuration conf) throws Exception {
+        var cancelledJobIds = new HashSet<JobID>();
+        try (var clusterClient = getClusterClient(conf)) {
+            for (var job : listJobs(clusterClient)) {
+                if (!jobName.equals(job.getJobName())
+                        || job.getJobState().isGloballyTerminalState()) {
+                    continue;
+                }
+                LOG.info(
+                        "Cancelling existing job {} named '{}' on the session cluster before submission",
+                        job.getJobId(),
+                        jobName);
+                cancelAndAwaitTerminal(clusterClient, job.getJobId());
+                cancelledJobIds.add(job.getJobId());
+            }
+        }
+        return cancelledJobIds;
+    }
+
+    private void cancelAndAwaitTerminal(RestClusterClient<String> clusterClient, JobID jobId)
+            throws Exception {
+        var timeoutSeconds = operatorConfig.getFlinkCancelJobTimeout().toSeconds();
+        try {
+            clusterClient.cancel(jobId).get(timeoutSeconds, TimeUnit.SECONDS);
+            clusterClient.requestJobResult(jobId).get(timeoutSeconds, TimeUnit.SECONDS);
+        } catch (Exception e) {
+            if (isJobMissing(e) || isJobTerminated(e)) {
+                return;
+            }
+            throw e;
+        }
+    }
+
+    private Collection<JobStatusMessage> listJobs(RestClusterClient<String> clusterClient)
+            throws Exception {
+        var multipleJobsDetails =
+                clusterClient
+                        .sendRequest(
+                                JobsOverviewHeaders.getInstance(),
+                                EmptyMessageParameters.getInstance(),
+                                EmptyRequestBody.getInstance())
+                        .get(operatorConfig.getFlinkClientTimeout().toSeconds(), TimeUnit.SECONDS);
+        if (multipleJobsDetails.getJobs() == null) {
+            return List.of();
         }
+        return multipleJobsDetails.getJobs().stream()
+                .map(JobStatusUtils::toJobStatusMessage)
+                .collect(Collectors.toList());
     }
 
     @Override

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/FlinkService.java

@@ -47,6 +47,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.Set;
 
 /** Service for submitting and interacting with Flink clusters and jobs. */
 public interface FlinkService {
@@ -74,6 +75,18 @@ JobID submitJobToSessionCluster(
 
     Optional<JobStatusMessage> getJobStatus(Configuration conf, JobID jobId) throws Exception;
 
+    /**
+     * Cancels every non-terminal job currently running on the (session) cluster whose Flink job
+     * name equals {@code jobName} and blocks until each reaches a terminal state. Intended to
+     * remove a job that was orphaned on a session cluster (one no longer backed by a {@link
+     * FlinkSessionJob}) before an identically named job is resubmitted, so the two never run
+     * concurrently. Matching is by Flink job name ({@code pipeline.name}), which the operator
+     * defaults to the resource name.
+     *
+     * @return the ids of the jobs that were cancelled
+     */
+    Set<JobID> cancelRunningJobs(String jobName, Configuration conf) throws Exception;
+
     JobResult requestJobResult(Configuration conf, JobID jobID) throws Exception;
 
     CancelResult cancelJob(FlinkDeployment deployment, SuspendMode suspendMode, Configuration conf)

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

@@ -155,6 +155,7 @@ public class TestingFlinkService extends AbstractFlinkService {
 
     @Getter private int desiredReplicas = 0;
     @Getter private int cancelJobCallCount = 0;
+    @Getter private int cancelRunningJobsCallCount = 0;
 
     @Getter private Configuration submittedConf;
 
@@ -354,6 +355,37 @@ public long getRunningCount() {
         return jobs.stream().filter(t -> !t.f1.getJobState().isTerminalState()).count();
     }
 
+    @Override
+    public Set<JobID> cancelRunningJobs(String jobName, Configuration conf) {
+        cancelRunningJobsCallCount++;
+        var cancelledJobIds = new HashSet<JobID>();
+        for (var job : jobs) {
+            var status = job.f1;
+            if (!jobName.equals(status.getJobName())
+                    || status.getJobState().isGloballyTerminalState()) {
+                continue;
+            }
+            job.f1 =
+                    new JobStatusMessage(
+                            status.getJobId(),
+                            status.getJobName(),
+                            JobStatus.CANCELED,
+                            status.getStartTime());
+            cancelledJobIds.add(status.getJobId());
+        }
+        return cancelledJobIds;
+    }
+
+    @VisibleForTesting
+    public void addRunningJob(JobID jobId, String jobName) {
+        jobs.add(
+                Tuple3.of(
+                        null,
+                        new JobStatusMessage(
+                                jobId, jobName, JobStatus.RUNNING, System.currentTimeMillis()),
+                        new Configuration()));
+    }
+
     public void triggerSavepointLegacy(
             String jobId,
             SnapshotTriggerType triggerType,

flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/SessionJobReconcilerTest.java

@@ -210,6 +210,47 @@ public void testSubmitAndCleanUp() throws Exception {
         assertEquals(CANCELED, flinkService.listJobs().get(0).f1.getJobState());
     }
 
+    @Test
+    public void testCancelsOrphanedJobWithSameNameOnSubmit() throws Exception {
+        FlinkSessionJob sessionJob = TestUtils.buildSessionJob();
+        sessionJob
+                .getSpec()
+                .getFlinkConfiguration()
+                .put(
+                        KubernetesOperatorConfigOptions
+                                .OPERATOR_SESSION_JOB_CANCEL_EXISTING_ON_SUBMIT
+                                .key(),
+                        "true");
+        var jobName = sessionJob.getMetadata().getName();
+        var orphanedJobId = JobID.generate();
+        flinkService.addRunningJob(orphanedJobId, jobName);
+
+        reconciler.reconcile(
+                sessionJob, TestUtils.createContextWithReadyFlinkDeployment(kubernetesClient));
+
+        assertEquals(1, flinkService.getCancelRunningJobsCallCount());
+        assertEquals(1, flinkService.getRunningCount());
+        var orphaned =
+                flinkService.listJobs().stream()
+                        .filter(job -> job.f1.getJobId().equals(orphanedJobId))
+                        .findAny()
+                        .orElseThrow();
+        assertEquals(CANCELED, orphaned.f1.getJobState());
+    }
+
+    @Test
+    public void testKeepsExistingJobWithSameNameWhenCancelOnSubmitDisabled() throws Exception {
+        FlinkSessionJob sessionJob = TestUtils.buildSessionJob();
+        var jobName = sessionJob.getMetadata().getName();
+        flinkService.addRunningJob(JobID.generate(), jobName);
+
+        reconciler.reconcile(
+                sessionJob, TestUtils.createContextWithReadyFlinkDeployment(kubernetesClient));
+
+        assertEquals(0, flinkService.getCancelRunningJobsCallCount());
+        assertEquals(2, flinkService.getRunningCount());
+    }
+
     @Test
     public void testCancelJobRescheduled() throws Exception {
         FlinkSessionJob sessionJob = TestUtils.buildSessionJob();