[FLINK-39984][runtime] Dispatch thread dump to ioExecutor

flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java

@@ -1857,7 +1857,8 @@ public CompletableFuture<Collection<String>> requestMetricQueryServiceAddresses(
     @Override
     public CompletableFuture<ThreadDumpInfo> requestThreadDump(Duration timeout) {
         int stackTraceMaxDepth = configuration.get(ClusterOptions.THREAD_DUMP_STACKTRACE_MAX_DEPTH);
-        return CompletableFuture.completedFuture(ThreadDumpInfo.dumpAndCreate(stackTraceMaxDepth));
+        return CompletableFuture.supplyAsync(
+                () -> ThreadDumpInfo.dumpAndCreate(stackTraceMaxDepth), ioExecutor);
     }
 
     @Override

flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java

@@ -1465,7 +1465,8 @@ public CompletableFuture<ThreadDumpInfo> requestThreadDump(Duration timeout) {
                 taskManagerConfiguration
                         .getConfiguration()
                         .get(ClusterOptions.THREAD_DUMP_STACKTRACE_MAX_DEPTH);
-        return CompletableFuture.completedFuture(ThreadDumpInfo.dumpAndCreate(stacktraceMaxDepth));
+        return CompletableFuture.supplyAsync(
+                () -> ThreadDumpInfo.dumpAndCreate(stacktraceMaxDepth), ioExecutor);
     }
 
     @Override

flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorBuilder.java

@@ -94,6 +94,11 @@ public TaskExecutorBuilder setResourceId(ResourceID resourceId) {
         return this;
     }
 
+    public TaskExecutorBuilder setTaskManagerServices(TaskManagerServices taskManagerServices) {
+        this.taskManagerServices = taskManagerServices;
+        return this;
+    }
+
     public TaskExecutor build() throws Exception {
 
         final TaskExecutorBlobService resolvedTaskExecutorBlobService;

flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorThreadDumpOffloadTest.java

@@ -0,0 +1,107 @@
+/*
+ * 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.taskexecutor;
+
+import org.apache.flink.core.testutils.EachCallbackWrapper;
+import org.apache.flink.runtime.entrypoint.WorkingDirectory;
+import org.apache.flink.runtime.highavailability.TestingHighAvailabilityServicesBuilder;
+import org.apache.flink.runtime.rest.messages.ThreadDumpInfo;
+import org.apache.flink.runtime.rpc.RpcUtils;
+import org.apache.flink.runtime.rpc.TestingRpcServiceExtension;
+import org.apache.flink.runtime.taskmanager.LocalUnresolvedTaskManagerLocation;
+import org.apache.flink.util.TestLoggerExtension;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.File;
+import java.time.Duration;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Verifies that {@link TaskExecutor#requestThreadDump(Duration)} constructs the dump on the
+ * TaskManagerServices IO executor rather than on the RPC main thread. Constructing it on the main
+ * thread would block heartbeat replies and other RPCs behind a potentially multi-second JVM
+ * safepoint pause.
+ */
+@ExtendWith(TestLoggerExtension.class)
+class TaskExecutorThreadDumpOffloadTest {
+
+    private static final Duration RPC_TIMEOUT = Duration.ofSeconds(10);
+    private static final String IO_THREAD_NAME = "test-tm-io-thread";
+
+    private final TestingRpcServiceExtension rpcServiceExtension = new TestingRpcServiceExtension();
+
+    @RegisterExtension
+    private final EachCallbackWrapper<TestingRpcServiceExtension> eachWrapper =
+            new EachCallbackWrapper<>(rpcServiceExtension);
+
+    @Test
+    void requestThreadDumpRunsOnIoExecutor(@TempDir File tempDir) throws Exception {
+        // A single-thread ioExecutor with a distinctive name: ThreadMXBean.dumpAllThreads()
+        // captures every live thread in the JVM, so if the dump runs on this executor the
+        // named thread will appear in the returned ThreadDumpInfo -- direct proof that the
+        // dump was offloaded to it instead of running on the RPC main thread.
+        final ExecutorService ioExecutor =
+                Executors.newSingleThreadExecutor(r -> new Thread(r, IO_THREAD_NAME));
+        try {
+            final TaskManagerServices services =
+                    new TaskManagerServicesBuilder()
+                            .setUnresolvedTaskManagerLocation(
+                                    new LocalUnresolvedTaskManagerLocation())
+                            .setIoExecutor(ioExecutor)
+                            .build();
+
+            final TaskExecutor taskExecutor =
+                    TaskExecutorBuilder.newBuilder(
+                                    rpcServiceExtension.getTestingRpcService(),
+                                    new TestingHighAvailabilityServicesBuilder().build(),
+                                    WorkingDirectory.create(tempDir))
+                            .setTaskManagerServices(services)
+                            .build();
+            try {
+                taskExecutor.start();
+
+                final ThreadDumpInfo dump =
+                        taskExecutor
+                                .getSelfGateway(TaskExecutorGateway.class)
+                                .requestThreadDump(RPC_TIMEOUT)
+                                .get(20, TimeUnit.SECONDS);
+
+                assertThat(dump.getThreadInfos())
+                        .as(
+                                "Thread dump must include the ioExecutor thread '%s' -- its "
+                                        + "presence proves the dump was offloaded to it instead "
+                                        + "of running on the RPC main thread.",
+                                IO_THREAD_NAME)
+                        .anyMatch(t -> IO_THREAD_NAME.equals(t.getThreadName()));
+            } finally {
+                RpcUtils.terminateRpcEndpoint(taskExecutor);
+            }
+        } finally {
+            ioExecutor.shutdownNow();
+        }
+    }
+}

flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskManagerServicesBuilder.java

@@ -42,6 +42,7 @@
 import org.apache.flink.runtime.util.NoOpGroupCache;
 
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 
 import static org.mockito.Mockito.mock;
@@ -68,6 +69,7 @@ public class TaskManagerServicesBuilder {
     private SharedResources sharedResources;
     private long managedMemorySize;
     private SlotAllocationSnapshotPersistenceService slotAllocationSnapshotPersistenceService;
+    private ExecutorService ioExecutor;
 
     public TaskManagerServicesBuilder() {
         unresolvedTaskManagerLocation = new LocalUnresolvedTaskManagerLocation();
@@ -197,12 +199,17 @@ public TaskManagerServices build() {
                 taskChangelogStoragesManager,
                 taskChannelStateExecutorFactoryManager,
                 taskEventDispatcher,
-                Executors.newSingleThreadScheduledExecutor(),
+                ioExecutor != null ? ioExecutor : Executors.newSingleThreadScheduledExecutor(),
                 libraryCacheManager,
                 slotAllocationSnapshotPersistenceService,
                 sharedResources,
                 new NoOpGroupCache<>(),
                 new NoOpGroupCache<>(),
                 new NoOpGroupCache<>());
     }
+
+    public TaskManagerServicesBuilder setIoExecutor(ExecutorService ioExecutor) {
+        this.ioExecutor = ioExecutor;
+        return this;
+    }
 }