From 3e045845e603625df854215fdc14dea71ac77c46 Mon Sep 17 00:00:00 2001 From: xingsuo-zbz Date: Fri, 24 Jul 2026 12:52:35 +0800 Subject: [PATCH] [FLINK-39984][runtime] Dispatch thread dump to ioExecutor Offload the dump to ioExecutor in both TaskExecutor#requestThreadDump and Dispatcher#requestThreadDump via CompletableFuture.supplyAsync, matching the pattern already used by other heavy TE handlers. --- .../flink/runtime/dispatcher/Dispatcher.java | 3 +- .../runtime/taskexecutor/TaskExecutor.java | 3 +- .../DispatcherThreadDumpOffloadTest.java | 72 ++++++++++++++ .../taskexecutor/TaskExecutorBuilder.java | 5 + .../TaskExecutorThreadDumpOffloadTest.java | 98 +++++++++++++++++++ .../TaskManagerServicesBuilder.java | 9 +- 6 files changed, 187 insertions(+), 3 deletions(-) create mode 100644 flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherThreadDumpOffloadTest.java create mode 100644 flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorThreadDumpOffloadTest.java diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java index 6d53fa0e2216be..1a67c1322bb639 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java @@ -1857,7 +1857,8 @@ public CompletableFuture> requestMetricQueryServiceAddresses( @Override public CompletableFuture 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 diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java index e5b9fc9b79d4ca..0e370412dff385 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java @@ -1465,7 +1465,8 @@ public CompletableFuture 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 diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherThreadDumpOffloadTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherThreadDumpOffloadTest.java new file mode 100644 index 00000000000000..111e52a1414e25 --- /dev/null +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherThreadDumpOffloadTest.java @@ -0,0 +1,72 @@ +/* + * 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.dispatcher; + +import org.apache.flink.runtime.rest.messages.ThreadDumpInfo; +import org.apache.flink.runtime.rpc.RpcUtils; + +import org.junit.After; +import org.junit.Test; + +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests that {@link Dispatcher#requestThreadDump} is offloaded to the ioExecutor. */ +public class DispatcherThreadDumpOffloadTest extends AbstractDispatcherTest { + + private static final String IO_THREAD_NAME = "test-jm-io-thread"; + + private TestingDispatcher dispatcher; + private ExecutorService ioExecutor; + + @After + @Override + public void tearDown() throws Exception { + if (dispatcher != null) { + RpcUtils.terminateRpcEndpoint(dispatcher); + } + if (ioExecutor != null) { + ioExecutor.shutdownNow(); + } + super.tearDown(); + } + + @Test + public void requestThreadDumpRunsOnIoExecutor() throws Exception { + // Named single-thread executor: if the dump runs on it, dumpAllThreads() captures + // the thread and it appears in the returned ThreadDumpInfo. + ioExecutor = Executors.newSingleThreadExecutor(r -> new Thread(r, IO_THREAD_NAME)); + + dispatcher = createTestingDispatcherBuilder().setIoExecutor(ioExecutor).build(rpcService); + dispatcher.start(); + + final ThreadDumpInfo dump = + dispatcher + .getSelfGateway(DispatcherGateway.class) + .requestThreadDump(TIMEOUT) + .get(20, TimeUnit.SECONDS); + + assertThat(dump.getThreadInfos()) + .as("dump must include ioExecutor thread '%s' (proves offload)", IO_THREAD_NAME) + .anyMatch(t -> IO_THREAD_NAME.equals(t.getThreadName())); + } +} diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorBuilder.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorBuilder.java index 73f0b7d6e9a145..6c213f6cc71446 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorBuilder.java +++ b/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; diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorThreadDumpOffloadTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorThreadDumpOffloadTest.java new file mode 100644 index 00000000000000..5257890563c053 --- /dev/null +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorThreadDumpOffloadTest.java @@ -0,0 +1,98 @@ +/* + * 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; + +/** Tests that {@link TaskExecutor#requestThreadDump(Duration)} is offloaded to the ioExecutor. */ +@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 eachWrapper = + new EachCallbackWrapper<>(rpcServiceExtension); + + @Test + void requestThreadDumpRunsOnIoExecutor(@TempDir File tempDir) throws Exception { + // Named single-thread executor: if the dump runs on it, dumpAllThreads() captures + // the thread and it appears in the returned ThreadDumpInfo. + 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( + "dump must include ioExecutor thread '%s' (proves offload)", + IO_THREAD_NAME) + .anyMatch(t -> IO_THREAD_NAME.equals(t.getThreadName())); + } finally { + RpcUtils.terminateRpcEndpoint(taskExecutor); + } + } finally { + ioExecutor.shutdownNow(); + } + } +} diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskManagerServicesBuilder.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskManagerServicesBuilder.java index 68d9d5a275de73..62b8ed470aefbd 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskManagerServicesBuilder.java +++ b/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(); @@ -181,6 +183,11 @@ public TaskManagerServicesBuilder setSlotAllocationSnapshotPersistenceService( return this; } + public TaskManagerServicesBuilder setIoExecutor(ExecutorService ioExecutor) { + this.ioExecutor = ioExecutor; + return this; + } + public TaskManagerServices build() { return new TaskManagerServices( unresolvedTaskManagerLocation, @@ -197,7 +204,7 @@ public TaskManagerServices build() { taskChangelogStoragesManager, taskChannelStateExecutorFactoryManager, taskEventDispatcher, - Executors.newSingleThreadScheduledExecutor(), + ioExecutor != null ? ioExecutor : Executors.newSingleThreadScheduledExecutor(), libraryCacheManager, slotAllocationSnapshotPersistenceService, sharedResources,