diff --git a/docs/layouts/shortcodes/generated/dynamic_section.html b/docs/layouts/shortcodes/generated/dynamic_section.html index 11eafe81cb..e1a94943ef 100644 --- a/docs/layouts/shortcodes/generated/dynamic_section.html +++ b/docs/layouts/shortcodes/generated/dynamic_section.html @@ -194,6 +194,12 @@ Duration The interval before a savepoint trigger attempt is marked as unsuccessful. + +
kubernetes.operator.session-job.cancel-existing-on-submit
+ false + Boolean + 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. +
kubernetes.operator.session.deletion.block-on-session-jobs
true diff --git a/docs/layouts/shortcodes/generated/kubernetes_operator_config_configuration.html b/docs/layouts/shortcodes/generated/kubernetes_operator_config_configuration.html index 22a8da165e..a79cc787a9 100644 --- a/docs/layouts/shortcodes/generated/kubernetes_operator_config_configuration.html +++ b/docs/layouts/shortcodes/generated/kubernetes_operator_config_configuration.html @@ -428,6 +428,12 @@ Duration The interval before a savepoint trigger attempt is marked as unsuccessful. + +
kubernetes.operator.session-job.cancel-existing-on-submit
+ false + Boolean + 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. +
kubernetes.operator.session.deletion.block-on-session-jobs
true diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/KubernetesOperatorConfigOptions.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/KubernetesOperatorConfigOptions.java index 878f6ed416..a8acbe539c 100644 --- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/KubernetesOperatorConfigOptions.java +++ b/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 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 OPERATOR_DYNAMIC_CONFIG_ENABLED = operatorConfig("dynamic.config.enabled") diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/SessionJobReconciler.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/SessionJobReconciler.java index 4491ab9712..f91a99b2d3 100644 --- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/SessionJobReconciler.java +++ b/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 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 ctx, SuspendMode suspendMode) throws Exception { diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java index 543b937f2d..c256fa34c1 100644 --- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java +++ b/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 clusterClient public Optional 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.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 cancelRunningJobs(String jobName, Configuration conf) throws Exception { + var cancelledJobIds = new HashSet(); + 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 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 listJobs(RestClusterClient 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 diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/FlinkService.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/FlinkService.java index b1a078fbb2..2563703ccf 100644 --- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/FlinkService.java +++ b/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 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 cancelRunningJobs(String jobName, Configuration conf) throws Exception; + JobResult requestJobResult(Configuration conf, JobID jobID) throws Exception; CancelResult cancelJob(FlinkDeployment deployment, SuspendMode suspendMode, Configuration conf) diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkService.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkService.java index b17fb624a1..11741c80aa 100644 --- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkService.java +++ b/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 cancelRunningJobs(String jobName, Configuration conf) { + cancelRunningJobsCallCount++; + var cancelledJobIds = new HashSet(); + 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, diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/SessionJobReconcilerTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/SessionJobReconcilerTest.java index 3416d47479..00581bd21b 100644 --- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/SessionJobReconcilerTest.java +++ b/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();