From 2371f5fd8a1f59fcf4f3b30da34f35fad5ad5d16 Mon Sep 17 00:00:00 2001 From: Matthias Pohl Date: Fri, 4 Sep 2026 14:10:57 +0200 Subject: [PATCH] [FLINK-40554][runtime] Skipping job recovery for jobs with broken execution plan instead of failing cluster recovery hidden behind newly introduced config parameter cluster.job-error-isolation.enabled The old implementation treated ExecutionPlan recovery errors as cluster-wide fatal causing the JobManager to fail over. This would cause other jobs on the same JobManager to failover as well. Enabling the new feature through the newly added configuration parameter makes the job recovery being skipped rather than failing the JobManager since the job wouldn't recover anyway. Cleanup needs to be handled manually by the operator. The error is revealed via error logs. --- .../generated/cluster_configuration.html | 6 ++ .../generated/expert_cluster_section.html | 6 ++ ...DispatcherLeaderProcessFactoryFactory.java | 2 + .../flink/configuration/ClusterOptions.java | 15 ++++ .../SessionDispatcherLeaderProcess.java | 24 +++++- ...SessionDispatcherLeaderProcessFactory.java | 4 + ...DispatcherLeaderProcessFactoryFactory.java | 4 + ...okenExecutionPlanStateHandleException.java | 39 +++++++++ .../jobmanager/DefaultExecutionPlanStore.java | 4 +- .../SessionDispatcherLeaderProcessTest.java | 84 +++++++++++++++++++ .../DefaultExecutionPlanStoreTest.java | 34 ++++++++ 11 files changed, 218 insertions(+), 4 deletions(-) create mode 100644 flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/BrokenExecutionPlanStateHandleException.java diff --git a/docs/layouts/shortcodes/generated/cluster_configuration.html b/docs/layouts/shortcodes/generated/cluster_configuration.html index 04e605d53e3510..06d8aa367ce702 100644 --- a/docs/layouts/shortcodes/generated/cluster_configuration.html +++ b/docs/layouts/shortcodes/generated/cluster_configuration.html @@ -26,6 +26,12 @@ Integer The size of the IO executor pool used by the cluster to execute blocking IO operations (Master as well as TaskManager processes). By default it will use 4 * the number of CPU cores (hardware contexts) that the cluster process has access to. Increasing the pool size allows to run more IO operations concurrently. + +
cluster.job-error-isolation.enabled
+ false + Boolean + Whether errors that can be scoped to a single job are handled by failing only the affected job instead of the whole cluster. For example, if a job's persisted execution plan cannot be recovered because its state handle is broken, enabling this option marks only that job as failed instead of aborting recovery for every job in the cluster. +
cluster.processes.halt-on-fatal-error
false diff --git a/docs/layouts/shortcodes/generated/expert_cluster_section.html b/docs/layouts/shortcodes/generated/expert_cluster_section.html index 879d46790035a0..093fab8306ca45 100644 --- a/docs/layouts/shortcodes/generated/expert_cluster_section.html +++ b/docs/layouts/shortcodes/generated/expert_cluster_section.html @@ -20,6 +20,12 @@

Enum

Flag to check user code exiting system by terminating JVM (e.g., System.exit()). Note that this configuration option can interfere with cluster.processes.halt-on-fatal-error: In intercepted user-code, a call to System.exit() will not cause the JVM to halt, when THROW is configured.

Possible values: + +
cluster.job-error-isolation.enabled
+ false + Boolean + Whether errors that can be scoped to a single job are handled by failing only the affected job instead of the whole cluster. For example, if a job's persisted execution plan cannot be recovered because its state handle is broken, enabling this option marks only that job as failed instead of aborting recovery for every job in the cluster. +
cluster.processes.halt-on-fatal-error
false diff --git a/flink-clients/src/main/java/org/apache/flink/client/deployment/application/ApplicationDispatcherLeaderProcessFactoryFactory.java b/flink-clients/src/main/java/org/apache/flink/client/deployment/application/ApplicationDispatcherLeaderProcessFactoryFactory.java index 0bf11a5d8aa596..85aab72fbc53b4 100644 --- a/flink-clients/src/main/java/org/apache/flink/client/deployment/application/ApplicationDispatcherLeaderProcessFactoryFactory.java +++ b/flink-clients/src/main/java/org/apache/flink/client/deployment/application/ApplicationDispatcherLeaderProcessFactoryFactory.java @@ -20,6 +20,7 @@ import org.apache.flink.annotation.Internal; import org.apache.flink.client.program.PackagedProgram; +import org.apache.flink.configuration.ClusterOptions; import org.apache.flink.configuration.Configuration; import org.apache.flink.runtime.dispatcher.DispatcherFactory; import org.apache.flink.runtime.dispatcher.PartialDispatcherServices; @@ -78,6 +79,7 @@ public DispatcherLeaderProcessFactory createFactory( persistenceComponentFactory, partialDispatcherServices.getBlobServer(), ioExecutor, + configuration.get(ClusterOptions.JOB_ERROR_ISOLATION_ENABLED), fatalErrorHandler); } diff --git a/flink-core/src/main/java/org/apache/flink/configuration/ClusterOptions.java b/flink-core/src/main/java/org/apache/flink/configuration/ClusterOptions.java index 6d03db19229fb0..d6b21e05ad7e5a 100644 --- a/flink-core/src/main/java/org/apache/flink/configuration/ClusterOptions.java +++ b/flink-core/src/main/java/org/apache/flink/configuration/ClusterOptions.java @@ -117,6 +117,21 @@ public class ClusterOptions { "FLINK-16510")) .build()); + @Documentation.Section(Documentation.Sections.EXPERT_CLUSTER) + public static final ConfigOption JOB_ERROR_ISOLATION_ENABLED = + key("cluster.job-error-isolation.enabled") + .booleanType() + .defaultValue(false) + .withDescription( + Description.builder() + .text( + "Whether errors that can be scoped to a single job are handled by failing " + + "only the affected job instead of the whole cluster. For example, if a " + + "job's persisted execution plan cannot be recovered because its state " + + "handle is broken, enabling this option marks only that job as failed " + + "instead of aborting recovery for every job in the cluster.") + .build()); + @Documentation.Section(Documentation.Sections.EXPERT_CLUSTER) public static final ConfigOption INTERCEPT_USER_SYSTEM_EXIT = key("cluster.intercept-user-system-exit") diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/runner/SessionDispatcherLeaderProcess.java b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/runner/SessionDispatcherLeaderProcess.java index 0c133ac7714e42..15e9915c91558d 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/runner/SessionDispatcherLeaderProcess.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/runner/SessionDispatcherLeaderProcess.java @@ -22,6 +22,7 @@ import org.apache.flink.api.common.JobID; import org.apache.flink.api.common.JobInfo; import org.apache.flink.api.common.JobInfoImpl; +import org.apache.flink.configuration.ClusterOptions; import org.apache.flink.configuration.RpcOptions; import org.apache.flink.runtime.application.AbstractApplication; import org.apache.flink.runtime.application.SingleJobApplication; @@ -34,6 +35,7 @@ import org.apache.flink.runtime.highavailability.JobResultStore; import org.apache.flink.runtime.jobmanager.ApplicationStore; import org.apache.flink.runtime.jobmanager.ApplicationStoreEntry; +import org.apache.flink.runtime.jobmanager.BrokenExecutionPlanStateHandleException; import org.apache.flink.runtime.jobmanager.ExecutionPlanStore; import org.apache.flink.runtime.jobmaster.JobResult; import org.apache.flink.runtime.messages.FlinkApplicationNotFoundException; @@ -82,6 +84,8 @@ public class SessionDispatcherLeaderProcess extends AbstractDispatcherLeaderProc private final Executor ioExecutor; + private final boolean jobErrorIsolationEnabled; + private CompletableFuture onGoingRecoveryOperation = FutureUtils.completedVoidFuture(); private SessionDispatcherLeaderProcess( @@ -93,6 +97,7 @@ private SessionDispatcherLeaderProcess( ApplicationResultStore applicationResultStore, BlobServer blobServer, Executor ioExecutor, + boolean jobErrorIsolationEnabled, FatalErrorHandler fatalErrorHandler) { super(leaderSessionId, fatalErrorHandler); @@ -103,6 +108,7 @@ private SessionDispatcherLeaderProcess( this.applicationResultStore = applicationResultStore; this.blobServer = blobServer; this.ioExecutor = ioExecutor; + this.jobErrorIsolationEnabled = jobErrorIsolationEnabled; } @Override @@ -262,6 +268,7 @@ private Collection getJobIds() { private Optional tryRecoverJob(JobID jobId) { log.info("Trying to recover job with job id {}.", jobId); + final String errorMessage = String.format("Could not recover job with job id %s.", jobId); try { final ExecutionPlan executionPlan = executionPlanStore.recoverExecutionPlan(jobId); if (executionPlan == null) { @@ -270,9 +277,20 @@ private Optional tryRecoverJob(JobID jobId) { jobId); } return Optional.ofNullable(executionPlan); + } catch (BrokenExecutionPlanStateHandleException e) { + if (!jobErrorIsolationEnabled) { + throw new FlinkRuntimeException(errorMessage, e); + } + log.error( + "The persisted ExecutionPlan of job {} is broken beyond repair and cannot be recovered. Skipping " + + "recovery for this job. This job will not be resubmitted and no automatic cleanup will " + + "be performed for it; manual cleanup of its dangling HA state is required. See cause " + + "for details.", + jobId, + e); + return Optional.empty(); } catch (Exception e) { - throw new FlinkRuntimeException( - String.format("Could not recover job with job id %s.", jobId), e); + throw new FlinkRuntimeException(errorMessage, e); } } @@ -553,6 +571,7 @@ public static SessionDispatcherLeaderProcess create( ApplicationResultStore applicationResultStore, BlobServer blobServer, Executor ioExecutor, + boolean jobErrorIsolationEnabled, FatalErrorHandler fatalErrorHandler) { return new SessionDispatcherLeaderProcess( leaderSessionId, @@ -563,6 +582,7 @@ public static SessionDispatcherLeaderProcess create( applicationResultStore, blobServer, ioExecutor, + jobErrorIsolationEnabled, fatalErrorHandler); } } diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/runner/SessionDispatcherLeaderProcessFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/runner/SessionDispatcherLeaderProcessFactory.java index 777c66e0f5aab8..39d5e0f810506b 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/runner/SessionDispatcherLeaderProcessFactory.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/runner/SessionDispatcherLeaderProcessFactory.java @@ -35,6 +35,7 @@ public class SessionDispatcherLeaderProcessFactory implements DispatcherLeaderPr private final PersistenceComponentFactory persistenceComponentFactory; private final BlobServer blobServer; private final Executor ioExecutor; + private final boolean jobErrorIsolationEnabled; private final FatalErrorHandler fatalErrorHandler; public SessionDispatcherLeaderProcessFactory( @@ -43,11 +44,13 @@ public SessionDispatcherLeaderProcessFactory( PersistenceComponentFactory persistenceComponentFactory, BlobServer blobServer, Executor ioExecutor, + boolean jobErrorIsolationEnabled, FatalErrorHandler fatalErrorHandler) { this.dispatcherGatewayServiceFactory = dispatcherGatewayServiceFactory; this.persistenceComponentFactory = persistenceComponentFactory; this.blobServer = blobServer; this.ioExecutor = ioExecutor; + this.jobErrorIsolationEnabled = jobErrorIsolationEnabled; this.fatalErrorHandler = fatalErrorHandler; } @@ -62,6 +65,7 @@ public DispatcherLeaderProcess create(UUID leaderSessionID) { persistenceComponentFactory.createApplicationResultStore(), blobServer, ioExecutor, + jobErrorIsolationEnabled, fatalErrorHandler); } } diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/runner/SessionDispatcherLeaderProcessFactoryFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/runner/SessionDispatcherLeaderProcessFactoryFactory.java index 99fb4667eec3d4..239c49b7c879a2 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/runner/SessionDispatcherLeaderProcessFactoryFactory.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/runner/SessionDispatcherLeaderProcessFactoryFactory.java @@ -18,6 +18,7 @@ package org.apache.flink.runtime.dispatcher.runner; +import org.apache.flink.configuration.ClusterOptions; import org.apache.flink.runtime.dispatcher.DispatcherFactory; import org.apache.flink.runtime.dispatcher.PartialDispatcherServices; import org.apache.flink.runtime.jobmanager.PersistenceComponentFactory; @@ -53,6 +54,9 @@ public DispatcherLeaderProcessFactory createFactory( persistenceComponentFactory, partialDispatcherServices.getBlobServer(), ioExecutor, + partialDispatcherServices + .getConfiguration() + .get(ClusterOptions.JOB_ERROR_ISOLATION_ENABLED), fatalErrorHandler); } diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/BrokenExecutionPlanStateHandleException.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/BrokenExecutionPlanStateHandleException.java new file mode 100644 index 00000000000000..94d066bf08230b --- /dev/null +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/BrokenExecutionPlanStateHandleException.java @@ -0,0 +1,39 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.runtime.jobmanager; + +import org.apache.flink.api.common.JobID; +import org.apache.flink.util.FlinkException; + +/** + * Thrown by {@link ExecutionPlanStore#recoverExecutionPlan(JobID)} when a job's persisted {@link + * org.apache.flink.streaming.api.graph.ExecutionPlan} cannot be deserialized because its state + * handle is broken (e.g. the backing file is missing/corrupted, or written by an incompatible Flink + * version). Unlike other recovery failures (e.g. a temporarily unreachable backend), this is not + * transient, so callers can react by skipping just the affected job instead of failing the whole + * recovery. + */ +public class BrokenExecutionPlanStateHandleException extends FlinkException { + + private static final long serialVersionUID = 1L; + + public BrokenExecutionPlanStateHandleException(String message, Throwable cause) { + super(message, cause); + } +} diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/DefaultExecutionPlanStore.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/DefaultExecutionPlanStore.java index ae58c2722a802a..2b156474037a25 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/DefaultExecutionPlanStore.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/DefaultExecutionPlanStore.java @@ -168,14 +168,14 @@ public ExecutionPlan recoverExecutionPlan(JobID jobId) throws Exception { try { executionPlan = executionPlanRetrievableStateHandle.retrieveState(); } catch (ClassNotFoundException cnfe) { - throw new FlinkException( + throw new BrokenExecutionPlanStateHandleException( "Could not retrieve submitted ExecutionPlan from state handle under " + name + ". This indicates that you are trying to recover from state written by an " + "older Flink version which is not compatible. Try cleaning the state handle store.", cnfe); } catch (IOException ioe) { - throw new FlinkException( + throw new BrokenExecutionPlanStateHandleException( "Could not retrieve submitted ExecutionPlan from state handle under " + name + ". This indicates that the retrieved state handle is broken. Try cleaning the state handle " diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/runner/SessionDispatcherLeaderProcessTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/runner/SessionDispatcherLeaderProcessTest.java index 99cf2b97205d1e..c0b0ef0f286dbf 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/runner/SessionDispatcherLeaderProcessTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/runner/SessionDispatcherLeaderProcessTest.java @@ -34,6 +34,7 @@ import org.apache.flink.runtime.jobgraph.JobGraphTestUtils; import org.apache.flink.runtime.jobmanager.ApplicationStore; import org.apache.flink.runtime.jobmanager.ApplicationStoreEntry; +import org.apache.flink.runtime.jobmanager.BrokenExecutionPlanStateHandleException; import org.apache.flink.runtime.jobmanager.ExecutionPlanStore; import org.apache.flink.runtime.jobmanager.TestingApplicationStoreEntry; import org.apache.flink.runtime.jobmaster.JobResult; @@ -59,6 +60,7 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; +import java.io.IOException; import java.util.Arrays; import java.util.Collection; import java.util.Collections; @@ -103,6 +105,8 @@ class SessionDispatcherLeaderProcessTest { private BlobServer blobServer; + private boolean jobErrorIsolationEnabled; + private AbstractDispatcherLeaderProcess.DispatcherGatewayServiceFactory dispatcherServiceFactory; @@ -125,6 +129,7 @@ void setup() { jobResultStore = TestingJobResultStore.builder().build(); applicationStore = TestingApplicationStore.newBuilder().build(); applicationResultStore = TestingApplicationResultStore.builder().build(); + jobErrorIsolationEnabled = true; dispatcherServiceFactory = createFactoryBasedOnGenericSupplier( () -> TestingDispatcherGatewayService.newBuilder().build()); @@ -896,6 +901,84 @@ void recoverJobs_withJobIdRecoveryFailure_failsFatally() throws Exception { runRecoveryFailureTest(testException); } + @Test + void recoverJobs_withBrokenExecutionPlan_skipsJobAndRecoversRemainingJobs() throws Exception { + final ExecutionPlan healthyJobGraph = JobGraphTestUtils.emptyJobGraph(); + healthyJobGraph.setApplicationId(JOB_GRAPH.getApplicationId().get()); + final BrokenExecutionPlanStateHandleException brokenStateException = + new BrokenExecutionPlanStateHandleException( + "Broken state handle.", new IOException("Test IO exception.")); + + executionPlanStore = + TestingExecutionPlanStore.newBuilder() + .setJobIdsFunction( + ignored -> + Arrays.asList( + JOB_GRAPH.getJobID(), healthyJobGraph.getJobID())) + .setRecoverExecutionPlanFunction( + (jobId, jobs) -> { + if (jobId.equals(JOB_GRAPH.getJobID())) { + throw brokenStateException; + } + return healthyJobGraph; + }) + .build(); + + final CompletableFuture> recoveredExecutionPlansFuture = + new CompletableFuture<>(); + final CompletableFuture> recoveredDirtyJobResultsFuture = + new CompletableFuture<>(); + dispatcherServiceFactory = + (ignoredDispatcherId, + recoveredJobs, + recoveredDirtyJobResults, + ignoredRecoveredApplications, + ignoredRecoveredDirtyApplicationResults, + ignoredExecutionPlanWriter, + ignoredJobResultStore, + ignoredApplicationStore, + ignoredApplicationResultStore) -> { + recoveredExecutionPlansFuture.complete(recoveredJobs); + recoveredDirtyJobResultsFuture.complete(recoveredDirtyJobResults); + return TestingDispatcherGatewayService.newBuilder().build(); + }; + + try (final SessionDispatcherLeaderProcess dispatcherLeaderProcess = + createDispatcherLeaderProcess()) { + dispatcherLeaderProcess.start(); + + assertThat(recoveredExecutionPlansFuture.get()) + .singleElement() + .isEqualTo(healthyJobGraph); + + // the broken job is silently skipped: no JobResult is recorded for it, since this + // codebase requires every dirty JobResult's application to be independently + // recoverable (either a real, persisted multi-job application, or a still-recoverable + // ExecutionPlan re-wrapped into a fresh SingleJobApplication), neither of which holds + // for a job whose plan is permanently unreadable + assertThat(recoveredDirtyJobResultsFuture.get()).isEmpty(); + } + } + + @Test + void recoverJobs_withBrokenExecutionPlanAndIsolationDisabled_failsFatally() throws Exception { + final BrokenExecutionPlanStateHandleException brokenStateException = + new BrokenExecutionPlanStateHandleException( + "Broken state handle.", new IOException("Test IO exception.")); + + jobErrorIsolationEnabled = false; + executionPlanStore = + TestingExecutionPlanStore.newBuilder() + .setRecoverExecutionPlanFunction( + (jobId, jobs) -> { + throw brokenStateException; + }) + .setInitialExecutionPlans(Collections.singleton(JOB_GRAPH)) + .build(); + + runRecoveryFailureTest(brokenStateException); + } + @Test void recoverApplications_withRecoveryFailure_failsFatally() throws Exception { final FlinkException testException = new FlinkException("Test exception"); @@ -1072,6 +1155,7 @@ private SessionDispatcherLeaderProcess createDispatcherLeaderProcess() { applicationResultStore, blobServer, ioExecutor, + jobErrorIsolationEnabled, fatalErrorHandler); } } diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/DefaultExecutionPlanStoreTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/DefaultExecutionPlanStoreTest.java index 5454e5e59db4cd..9507104fb3b019 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/DefaultExecutionPlanStoreTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/DefaultExecutionPlanStoreTest.java @@ -39,6 +39,7 @@ import org.junit.Before; import org.junit.Test; +import java.io.IOException; import java.util.Arrays; import java.util.Collection; import java.util.HashMap; @@ -143,6 +144,39 @@ public void testRecoverExecutionPlanFailedShouldReleaseHandle() throws Exception assertThat(testingExecutionPlan.getJobID()).hasToString(actual); } + @Test + public void testRecoverExecutionPlanWithBrokenStateHandleThrowsDedicatedException() + throws Exception { + final RetrievableStateHandle brokenStateHandle = + new RetrievableStateHandle() { + private static final long serialVersionUID = 1L; + + @Override + public ExecutionPlan retrieveState() throws IOException { + throw new IOException("Test IO exception simulating a broken handle."); + } + + @Override + public void discardState() {} + + @Override + public long getStateSize() { + return 0; + } + }; + final TestingStateHandleStore stateHandleStore = + builder.setGetFunction(ignore -> brokenStateHandle).build(); + + final ExecutionPlanStore executionPlanStore = + createAndStartExecutionPlanStore(stateHandleStore); + + assertThatThrownBy( + () -> + executionPlanStore.recoverExecutionPlan( + testingExecutionPlan.getJobID())) + .isInstanceOf(BrokenExecutionPlanStateHandleException.class); + } + @Test public void testPutExecutionPlanWhenNotExist() throws Exception { final CompletableFuture addFuture = new CompletableFuture<>();