This is an automated email from the ASF dual-hosted git repository.
Myasuka pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new e19fd025372 [FLINK-39984][runtime] Dispatch thread dump to ioExecutor
e19fd025372 is described below
commit e19fd025372b0437fe9414004a528420043ba0b0
Author: xingsuo-zbz <[email protected]>
AuthorDate: Fri Jul 24 12:52:35 2026 +0800
[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 +-
.../flink/runtime/taskexecutor/TaskExecutor.java | 3 +-
.../DispatcherThreadDumpOffloadTest.java | 72 ++++++++++++++++
.../runtime/taskexecutor/TaskExecutorBuilder.java | 5 ++
.../TaskExecutorThreadDumpOffloadTest.java | 98 ++++++++++++++++++++++
.../taskexecutor/TaskManagerServicesBuilder.java | 9 +-
6 files changed, 187 insertions(+), 3 deletions(-)
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 90f1f732b2a..50bb43285e0 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
@@ -1866,7 +1866,8 @@ public abstract class Dispatcher extends
FencedRpcEndpoint<DispatcherId>
@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
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 166935cbd8e..a7558bf6d3f 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
@@ -1471,7 +1471,8 @@ public class TaskExecutor extends RpcEndpoint implements
TaskExecutorGateway {
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 00000000000..111e52a1414
--- /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 73f0b7d6e9a..6c213f6cc71 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 class TaskExecutorBuilder {
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 00000000000..5257890563c
--- /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<TestingRpcServiceExtension> 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 68d9d5a275d..62b8ed470ae 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.taskmanager.UnresolvedTaskManagerLocation;
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 class TaskManagerServicesBuilder {
return this;
}
+ public TaskManagerServicesBuilder setIoExecutor(ExecutorService
ioExecutor) {
+ this.ioExecutor = ioExecutor;
+ return this;
+ }
+
public TaskManagerServices build() {
return new TaskManagerServices(
unresolvedTaskManagerLocation,
@@ -197,7 +204,7 @@ public class TaskManagerServicesBuilder {
taskChangelogStoragesManager,
taskChannelStateExecutorFactoryManager,
taskEventDispatcher,
- Executors.newSingleThreadScheduledExecutor(),
+ ioExecutor != null ? ioExecutor :
Executors.newSingleThreadScheduledExecutor(),
libraryCacheManager,
slotAllocationSnapshotPersistenceService,
sharedResources,
