[FLINK-41115][runtime] Isolate adaptive graph recovery from IO

docs/layouts/shortcodes/generated/all_jobmanager_section.html

@@ -126,7 +126,7 @@
             <td><h5>jobmanager.io-pool.size</h5></td>
             <td style="word-wrap: break-word;">(none)</td>
             <td>Integer</td>
-            <td>The size of the IO thread pool to run blocking operations for all spawned JobMasters. This includes recovery and completion of checkpoints. Increase this value if you experience slow checkpoint operations when running many jobs. If no value is specified, then Flink defaults to the number of available CPU cores.</td>
+            <td>The size of the IO thread pool to run blocking operations for all spawned JobMasters. This includes recovery and completion of checkpoints. The same size is used for a separate adaptive graph-recovery pool, shared across jobs, so the two pools together can use twice this number of threads. Asynchronous checkpoint-trigger initialization and cleanup continue to share the original IO pool. Synchronous storage operations during graph construction still use recovery workers. Increase this value if you experience slow checkpoint operations when running many jobs. If no value is specified, then Flink defaults to the number of available CPU cores.</td>
         </tr>
         <tr>
             <td><h5>jobmanager.partition.hybrid.partition-data-consume-constraint</h5></td>

docs/layouts/shortcodes/generated/expert_jobmanager_section.html

@@ -18,7 +18,7 @@
             <td><h5>jobmanager.io-pool.size</h5></td>
             <td style="word-wrap: break-word;">(none)</td>
             <td>Integer</td>
-            <td>The size of the IO thread pool to run blocking operations for all spawned JobMasters. This includes recovery and completion of checkpoints. Increase this value if you experience slow checkpoint operations when running many jobs. If no value is specified, then Flink defaults to the number of available CPU cores.</td>
+            <td>The size of the IO thread pool to run blocking operations for all spawned JobMasters. This includes recovery and completion of checkpoints. The same size is used for a separate adaptive graph-recovery pool, shared across jobs, so the two pools together can use twice this number of threads. Asynchronous checkpoint-trigger initialization and cleanup continue to share the original IO pool. Synchronous storage operations during graph construction still use recovery workers. Increase this value if you experience slow checkpoint operations when running many jobs. If no value is specified, then Flink defaults to the number of available CPU cores.</td>
         </tr>
     </tbody>
 </table>

docs/layouts/shortcodes/generated/job_manager_configuration.html

@@ -126,7 +126,7 @@
             <td><h5>jobmanager.io-pool.size</h5></td>
             <td style="word-wrap: break-word;">(none)</td>
             <td>Integer</td>
-            <td>The size of the IO thread pool to run blocking operations for all spawned JobMasters. This includes recovery and completion of checkpoints. Increase this value if you experience slow checkpoint operations when running many jobs. If no value is specified, then Flink defaults to the number of available CPU cores.</td>
+            <td>The size of the IO thread pool to run blocking operations for all spawned JobMasters. This includes recovery and completion of checkpoints. The same size is used for a separate adaptive graph-recovery pool, shared across jobs, so the two pools together can use twice this number of threads. Asynchronous checkpoint-trigger initialization and cleanup continue to share the original IO pool. Synchronous storage operations during graph construction still use recovery workers. Increase this value if you experience slow checkpoint operations when running many jobs. If no value is specified, then Flink defaults to the number of available CPU cores.</td>
         </tr>
         <tr>
             <td><h5>jobmanager.memory.enable-jvm-direct-memory-limit</h5></td>

flink-core/src/main/java/org/apache/flink/configuration/JobManagerOptions.java

@@ -504,6 +504,10 @@ public enum ArchivedApplicationStoreType {
                     .withDescription(
                             "The size of the IO thread pool to run blocking operations for all spawned JobMasters. "
                                     + "This includes recovery and completion of checkpoints. "
+                                    + "The same size is used for a separate adaptive graph-recovery pool, "
+                                    + "shared across jobs, so the two pools together can use twice this number of threads. "
+                                    + "Asynchronous checkpoint-trigger initialization and cleanup continue to share the original IO pool. "
+                                    + "Synchronous storage operations during graph construction still use recovery workers. "
                                     + "Increase this value if you experience slow checkpoint operations when running many jobs. "
                                     + "If no value is specified, then Flink defaults to the number of available CPU cores.");
 

flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/DefaultSlotPoolServiceSchedulerFactory.java

@@ -111,6 +111,7 @@ public SchedulerNG createScheduler(
             Logger log,
             ExecutionPlan executionPlan,
             Executor ioExecutor,
+            Executor recoveryExecutor,
             Configuration configuration,
             SlotPoolService slotPoolService,
             ScheduledExecutorService futureExecutor,
@@ -134,6 +135,7 @@ public SchedulerNG createScheduler(
                 log,
                 executionPlan,
                 ioExecutor,
+                recoveryExecutor,
                 configuration,
                 slotPoolService,
                 futureExecutor,

flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobManagerSharedServices.java

@@ -59,6 +59,8 @@ public class JobManagerSharedServices {
 
     private final ExecutorService ioExecutor;
 
+    private final ExecutorService recoveryExecutor;
+
     private final LibraryCacheManager libraryCacheManager;
 
     private final ShuffleMaster<?> shuffleMaster;
@@ -71,9 +73,25 @@ public JobManagerSharedServices(
             LibraryCacheManager libraryCacheManager,
             ShuffleMaster<?> shuffleMaster,
             @Nonnull BlobWriter blobWriter) {
+        this(
+                futureExecutor,
+                ioExecutor,
+                ioExecutor,
+                libraryCacheManager,
+                shuffleMaster,
+                blobWriter);
+    }
 
+    public JobManagerSharedServices(
+            ScheduledExecutorService futureExecutor,
+            ExecutorService ioExecutor,
+            ExecutorService recoveryExecutor,
+            LibraryCacheManager libraryCacheManager,
+            ShuffleMaster<?> shuffleMaster,
+            @Nonnull BlobWriter blobWriter) {
         this.futureExecutor = checkNotNull(futureExecutor);
         this.ioExecutor = checkNotNull(ioExecutor);
+        this.recoveryExecutor = checkNotNull(recoveryExecutor);
         this.libraryCacheManager = checkNotNull(libraryCacheManager);
         this.shuffleMaster = checkNotNull(shuffleMaster);
         this.blobWriter = blobWriter;
@@ -87,6 +105,15 @@ public Executor getIoExecutor() {
         return ioExecutor;
     }
 
+    /**
+     * Shared capacity for adaptive graph creation and state assignment, including synchronous
+     * storage calls within that work. Asynchronous checkpoint-trigger initialization and cleanup
+     * must continue to use the general IO executor to preserve their existing feedback.
+     */
+    public Executor getRecoveryExecutor() {
+        return recoveryExecutor;
+    }
+
     public LibraryCacheManager getLibraryCacheManager() {
         return libraryCacheManager;
     }
@@ -113,8 +140,12 @@ public void shutdown() throws Exception {
         Throwable exception = null;
 
         try {
+            final ExecutorService[] executors =
+                    recoveryExecutor == ioExecutor
+                            ? new ExecutorService[] {futureExecutor, ioExecutor}
+                            : new ExecutorService[] {futureExecutor, ioExecutor, recoveryExecutor};
             ExecutorUtils.gracefulShutdown(
-                    SHUTDOWN_TIMEOUT.toMillis(), TimeUnit.MILLISECONDS, futureExecutor, ioExecutor);
+                    SHUTDOWN_TIMEOUT.toMillis(), TimeUnit.MILLISECONDS, executors);
         } catch (Throwable t) {
             exception = t;
         }
@@ -183,7 +214,19 @@ public static JobManagerSharedServices fromConfiguration(
                         .createShuffleMaster(shuffleMasterContext);
         shuffleMaster.start();
 
+        // One fixed pool per JobManager, not per job or graph. The queue remains unbounded;
+        // BackgroundTask serializes suppliers within each adaptive scheduler. This separates
+        // worker availability, not storage clients or synchronous work inside a supplier.
+        final ExecutorService recoveryExecutor =
+                Executors.newFixedThreadPool(
+                        jobManagerIoPoolSize, new ExecutorThreadFactory("jobmanager-recovery"));
+
         return new JobManagerSharedServices(
-                futureExecutor, ioExecutor, libraryCacheManager, shuffleMaster, blobServer);
+                futureExecutor,
+                ioExecutor,
+                recoveryExecutor,
+                libraryCacheManager,
+                shuffleMaster,
+                blobServer);
     }
 }

flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java

@@ -178,6 +178,8 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId>
 
     private final Executor ioExecutor;
 
+    private final Executor recoveryExecutor;
+
     private final OnCompletionActions jobCompletionActions;
 
     private final FatalErrorHandler fatalErrorHandler;
@@ -346,6 +348,8 @@ public void onUnknownDeploymentsOf(
         this.futureExecutor =
                 MdcUtils.scopeToJob(jid, jobManagerSharedServices.getFutureExecutor());
         this.ioExecutor = MdcUtils.scopeToJob(jid, jobManagerSharedServices.getIoExecutor());
+        this.recoveryExecutor =
+                MdcUtils.scopeToJob(jid, jobManagerSharedServices.getRecoveryExecutor());
         this.jobCompletionActions = checkNotNull(jobCompletionActions);
         this.fatalErrorHandler = checkNotNull(fatalErrorHandler);
         this.userCodeLoader = checkNotNull(userCodeLoader);
@@ -431,6 +435,7 @@ private SchedulerNG createScheduler(
                         log,
                         executionPlan,
                         ioExecutor,
+                        recoveryExecutor,
                         jobMasterConfiguration.getConfiguration(),
                         slotPoolService,
                         futureExecutor,

flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/SlotPoolServiceSchedulerFactory.java

@@ -71,7 +71,9 @@ SlotPoolService createSlotPoolService(
     JobManagerOptions.SchedulerType getSchedulerType();
 
     /**
-     * Creates a {@link SchedulerNG}.
+     * Creates a {@link SchedulerNG}. All executors are borrowed. The recovery executor is only for
+     * adaptive graph creation and state assignment, not asynchronous checkpoint-trigger
+     * initialization or cleanup.
      *
      * @return created SchedulerNG
      * @throws Exception if the scheduler creation fails
@@ -80,6 +82,7 @@ SchedulerNG createScheduler(
             Logger log,
             ExecutionPlan executionPlan,
             Executor ioExecutor,
+            Executor recoveryExecutor,
             Configuration configuration,
             SlotPoolService slotPoolService,
             ScheduledExecutorService futureExecutor,

flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/DefaultSchedulerFactory.java

@@ -65,6 +65,7 @@ public SchedulerNG createInstance(
             final Logger log,
             final ExecutionPlan executionPlan,
             final Executor ioExecutor,
+            final Executor recoveryExecutor,
             final Configuration jobMasterConfiguration,
             final SlotPoolService slotPoolService,
             final ScheduledExecutorService futureExecutor,

flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/SchedulerNGFactory.java

@@ -45,10 +45,16 @@
 /** Factory for {@link SchedulerNG}. */
 public interface SchedulerNGFactory {
 
+    /**
+     * Creates a scheduler borrowing the supplied executors. The recovery executor is reserved for
+     * adaptive graph creation and state assignment; it must not replace the graph/coordinator IO
+     * executor or be shut down by the scheduler. Other scheduler types do not use it.
+     */
     SchedulerNG createInstance(
             Logger log,
             ExecutionPlan executionPlan,
             Executor ioExecutor,
+            Executor recoveryExecutor,
             Configuration jobMasterConfiguration,
             SlotPoolService slotPoolService,
             ScheduledExecutorService futureExecutor,

flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java

@@ -421,6 +421,7 @@ public JobRescaleConfigInfo toJobRescaleConfigInfo() {
     private final long initializationTimestamp;
 
     private final Executor ioExecutor;
+    private final Executor recoveryExecutor;
     private final ClassLoader userCodeClassLoader;
 
     private final CheckpointsCleaner checkpointsCleaner;
@@ -479,6 +480,7 @@ public AdaptiveScheduler(
             DeclarativeSlotPool declarativeSlotPool,
             SlotAllocator slotAllocator,
             Executor ioExecutor,
+            Executor recoveryExecutor,
             ClassLoader userCodeClassLoader,
             CheckpointsCleaner checkpointsCleaner,
             CheckpointRecoveryFactory checkpointRecoveryFactory,
@@ -506,6 +508,7 @@ public AdaptiveScheduler(
                 declarativeSlotPool,
                 slotAllocator,
                 ioExecutor,
+                recoveryExecutor,
                 userCodeClassLoader,
                 checkpointsCleaner,
                 checkpointRecoveryFactory,
@@ -531,6 +534,7 @@ public AdaptiveScheduler(
             DeclarativeSlotPool declarativeSlotPool,
             SlotAllocator slotAllocator,
             Executor ioExecutor,
+            Executor recoveryExecutor,
             ClassLoader userCodeClassLoader,
             CheckpointsCleaner checkpointsCleaner,
             CheckpointRecoveryFactory checkpointRecoveryFactory,
@@ -573,6 +577,7 @@ public AdaptiveScheduler(
         this.declarativeSlotPool = declarativeSlotPool;
         this.initializationTimestamp = initializationTimestamp;
         this.ioExecutor = ioExecutor;
+        this.recoveryExecutor = Preconditions.checkNotNull(recoveryExecutor);
         this.userCodeClassLoader = userCodeClassLoader;
         this.restartBackoffTimeStrategy = restartBackoffTimeStrategy;
         this.fatalErrorHandler = fatalErrorHandler;
@@ -1595,7 +1600,7 @@ private CompletableFuture<ExecutionGraph> createExecutionGraphAndRestoreStateAsy
         backgroundTask =
                 backgroundTask.runAfter(
                         () -> createExecutionGraphAndRestoreState(adjustedParallelismStore),
-                        ioExecutor);
+                        recoveryExecutor);
 
         return FutureUtils.switchExecutor(
                 backgroundTask.getResultFuture(), getMainThreadExecutor());

flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveSchedulerFactory.java

@@ -69,6 +69,7 @@ public SchedulerNG createInstance(
             Logger log,
             ExecutionPlan executionPlan,
             Executor ioExecutor,
+            Executor recoveryExecutor,
             Configuration jobMasterConfiguration,
             SlotPoolService slotPoolService,
             ScheduledExecutorService futureExecutor,
@@ -150,6 +151,7 @@ public SchedulerNG createInstance(
                 declarativeSlotPool,
                 slotAllocator,
                 ioExecutor,
+                recoveryExecutor,
                 userCodeLoader,
                 new CheckpointsCleaner(),
                 checkpointRecoveryFactory,

flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptivebatch/AdaptiveBatchSchedulerFactory.java

@@ -103,6 +103,7 @@ public SchedulerNG createInstance(
             Logger log,
             ExecutionPlan executionPlan,
             Executor ioExecutor,
+            Executor recoveryExecutor,
             Configuration jobMasterConfiguration,
             SlotPoolService slotPoolService,
             ScheduledExecutorService futureExecutor,