diff --git a/docs/layouts/shortcodes/generated/cluster_configuration.html b/docs/layouts/shortcodes/generated/cluster_configuration.html index 04e605d53e351..06d8aa367ce70 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 879d46790035a..093fab8306ca4 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 0bf11a5d8aa59..85aab72fbc53b 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 6d03db19229fb..d6b21e05ad7e5 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 0c133ac7714e4..15e9915c91558 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 777c66e0f5aab..39d5e0f810506 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 99fb4667eec3d..239c49b7c879a 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 0000000000000..94d066bf08230 --- /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 ae58c2722a802..2b156474037a2 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 99cf2b97205d1..c0b0ef0f286db 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 5454e5e59db4c..9507104fb3b01 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<>();