[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

docs/layouts/shortcodes/generated/cluster_configuration.html

@@ -26,6 +26,12 @@
             <td>Integer</td>
             <td>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.</td>
         </tr>
+        <tr>
+            <td><h5>cluster.job-error-isolation.enabled</h5></td>
+            <td style="word-wrap: break-word;">false</td>
+            <td>Boolean</td>
+            <td>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.</td>
+        </tr>
         <tr>
             <td><h5>cluster.processes.halt-on-fatal-error</h5></td>
             <td style="word-wrap: break-word;">false</td>

docs/layouts/shortcodes/generated/expert_cluster_section.html

@@ -20,6 +20,12 @@
             <td><p>Enum</p></td>
             <td>Flag to check user code exiting system by terminating JVM (e.g., System.exit()). Note that this configuration option can interfere with <code class="highlighter-rouge">cluster.processes.halt-on-fatal-error</code>: In intercepted user-code, a call to System.exit() will not cause the JVM to halt, when <code class="highlighter-rouge">THROW</code> is configured.<br /><br />Possible values:<ul><li>"DISABLED": Flink is not monitoring or intercepting calls to System.exit()</li><li>"LOG": Log exit attempt with stack trace but still allowing exit to be performed</li><li>"THROW": Throw exception when exit is attempted disallowing JVM termination</li></ul></td>
         </tr>
+        <tr>
+            <td><h5>cluster.job-error-isolation.enabled</h5></td>
+            <td style="word-wrap: break-word;">false</td>
+            <td>Boolean</td>
+            <td>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.</td>
+        </tr>
         <tr>
             <td><h5>cluster.processes.halt-on-fatal-error</h5></td>
             <td style="word-wrap: break-word;">false</td>

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);
     }
 

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<Boolean> 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<UserSystemExitMode> INTERCEPT_USER_SYSTEM_EXIT =
             key("cluster.intercept-user-system-exit")

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<Void> 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<JobID> getJobIds() {
 
     private Optional<ExecutionPlan> 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<ExecutionPlan> 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);
     }
 }

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);
     }
 }

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);
     }
 

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);
+    }
+}

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 "

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<Collection<ExecutionPlan>> recoveredExecutionPlansFuture =
+                new CompletableFuture<>();
+        final CompletableFuture<Collection<JobResult>> 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);
     }
 }