diff --git a/docs/static/generated/rest_v1_dispatcher.yml b/docs/static/generated/rest_v1_dispatcher.yml
index 24ecb76ed2e961..6fa37a389fe293 100644
--- a/docs/static/generated/rest_v1_dispatcher.yml
+++ b/docs/static/generated/rest_v1_dispatcher.yml
@@ -457,6 +457,15 @@ paths:
get:
description: Returns the thread dump of the JobManager.
operationId: getJobManagerThreadDump
+ parameters:
+ - name: mode
+ in: query
+ description: "Controls how much lock information is collected. Supported values:\
+ \ [SAFE, FULL]. When omitted, cluster.thread-dump.default-mode is used."
+ required: false
+ style: form
+ schema:
+ $ref: "#/components/schemas/ThreadDumpMode"
responses:
"200":
description: The request was successful.
@@ -1863,6 +1872,14 @@ paths:
required: true
schema:
$ref: "#/components/schemas/ResourceID"
+ - name: mode
+ in: query
+ description: "Controls how much lock information is collected. Supported values:\
+ \ [SAFE, FULL]. When omitted, cluster.thread-dump.default-mode is used."
+ required: false
+ style: form
+ schema:
+ $ref: "#/components/schemas/ThreadDumpMode"
responses:
"200":
description: The request was successful.
@@ -3911,6 +3928,11 @@ components:
type: array
items:
$ref: "#/components/schemas/ThreadInfo"
+ ThreadDumpMode:
+ type: string
+ enum:
+ - SAFE
+ - FULL
ThreadInfo:
type: object
properties:
diff --git a/flink-core/src/main/java/org/apache/flink/configuration/ClusterOptions.java b/flink-core/src/main/java/org/apache/flink/configuration/ClusterOptions.java
index 4957f486a08e58..8b111b1939577f 100644
--- a/flink-core/src/main/java/org/apache/flink/configuration/ClusterOptions.java
+++ b/flink-core/src/main/java/org/apache/flink/configuration/ClusterOptions.java
@@ -140,6 +140,31 @@ public class ClusterOptions {
.withDescription(
"The maximum stacktrace depth of TaskManager and JobManager's thread dump web-frontend displayed.");
+ @Documentation.Section(Documentation.Sections.EXPERT_CLUSTER)
+ public static final ConfigOption THREAD_DUMP_DEFAULT_MODE =
+ key("cluster.thread-dump.default-mode")
+ .stringType()
+ .defaultValue("FULL")
+ .withDescription(
+ Description.builder()
+ .text(
+ "Default granularity of the JobManager/TaskManager thread-dump REST endpoint "
+ + "when no explicit %s query parameter is supplied. ",
+ code("mode"))
+ .text(
+ "Allowed values (case-insensitive): %s (stack traces only, negligible JVM pause) "
+ + "and %s (adds locked monitors and j.u.c. synchronizers, equivalent "
+ + "to jstack -l; pauses the JVM in a safepoint for a duration that "
+ + "scales with heap size and thread count -- seconds on large "
+ + "TaskManagers). ",
+ code("SAFE"), code("FULL"))
+ .text(
+ "The default is %s to preserve historical behavior; operators of large "
+ + "clusters are strongly encouraged to switch to %s to avoid "
+ + "heartbeat timeouts caused by long safepoint pauses.",
+ code("FULL"), code("SAFE"))
+ .build());
+
@Documentation.Section(Documentation.Sections.EXPERT_CLUSTER)
public static final ConfigOption UNCAUGHT_EXCEPTION_HANDLING =
ConfigOptions.key("cluster.uncaught-exception-handling")
diff --git a/flink-runtime-web/src/test/resources/rest_api_v1.snapshot b/flink-runtime-web/src/test/resources/rest_api_v1.snapshot
index 890ac91ef3ce0b..a1dd691a7d29d1 100644
--- a/flink-runtime-web/src/test/resources/rest_api_v1.snapshot
+++ b/flink-runtime-web/src/test/resources/rest_api_v1.snapshot
@@ -1085,7 +1085,10 @@
"pathParameters" : [ ]
},
"query-parameters" : {
- "queryParameters" : [ ]
+ "queryParameters" : [ {
+ "key" : "mode",
+ "mandatory" : false
+ } ]
},
"request" : {
"type" : "object",
@@ -5429,7 +5432,10 @@
} ]
},
"query-parameters" : {
- "queryParameters" : [ ]
+ "queryParameters" : [ {
+ "key" : "mode",
+ "mandatory" : false
+ } ]
},
"request" : {
"type" : "object",
diff --git a/flink-runtime-web/web-dashboard/src/app/pages/job-manager/thread-dump/job-manager-thread-dump.component.html b/flink-runtime-web/web-dashboard/src/app/pages/job-manager/thread-dump/job-manager-thread-dump.component.html
index f7572891985a9e..10aec866a96960 100644
--- a/flink-runtime-web/web-dashboard/src/app/pages/job-manager/thread-dump/job-manager-thread-dump.component.html
+++ b/flink-runtime-web/web-dashboard/src/app/pages/job-manager/thread-dump/job-manager-thread-dump.component.html
@@ -22,8 +22,34 @@
[ngModel]="dump"
[nzEditorOption]="editorOptions"
>
-();
@@ -66,11 +84,24 @@ export class JobManagerThreadDumpComponent implements OnInit, OnDestroy {
this.destroy$.complete();
}
+ /**
+ * Switch dump mode. Does NOT auto-reload; user must press the refresh button.
+ * Download link is updated immediately so a download matches the user's selection.
+ */
+ public selectMode(mode: 'safe' | 'full'): void {
+ if (this.mode === mode) {
+ return;
+ }
+ this.mode = mode;
+ this.updateDownloadUrl();
+ this.cdr.markForCheck();
+ }
+
public reload(): void {
this.loading = true;
this.cdr.markForCheck();
this.jobManagerService
- .loadThreadDump()
+ .loadThreadDump(this.mode)
.pipe(
catchError(() => of('')),
takeUntil(this.destroy$)
@@ -81,4 +112,9 @@ export class JobManagerThreadDumpComponent implements OnInit, OnDestroy {
this.cdr.markForCheck();
});
}
+
+ private updateDownloadUrl(): void {
+ const base = `${this.configService.BASE_URL}/jobmanager/thread-dump`;
+ this.downloadUrl = this.mode ? `${base}?mode=${this.mode}` : base;
+ }
}
diff --git a/flink-runtime-web/web-dashboard/src/app/pages/task-manager/thread-dump/task-manager-thread-dump.component.html b/flink-runtime-web/web-dashboard/src/app/pages/task-manager/thread-dump/task-manager-thread-dump.component.html
index 30941b1370960b..42537ae8bf2a14 100644
--- a/flink-runtime-web/web-dashboard/src/app/pages/task-manager/thread-dump/task-manager-thread-dump.component.html
+++ b/flink-runtime-web/web-dashboard/src/app/pages/task-manager/thread-dump/task-manager-thread-dump.component.html
@@ -23,8 +23,34 @@
[nzEditorOption]="editorOptions"
(nzEditorInitialized)="nzEditorInitialized($event)"
>
- is also position:absolute by default; neutralize
+ // it so it stacks naturally to the right of the mode selector inside this flex
+ // container.
+ flink-addon-compact {
+ position: static;
+ top: auto;
+ right: auto;
+ }
+}
+
+.thread-dump-mode-group {
+ .thread-dump-mode-warn {
+ margin-left: 4px;
+ }
+}
diff --git a/flink-runtime-web/web-dashboard/src/app/pages/task-manager/thread-dump/task-manager-thread-dump.component.ts b/flink-runtime-web/web-dashboard/src/app/pages/task-manager/thread-dump/task-manager-thread-dump.component.ts
index 961a84493daea2..f45db99a2ab595 100644
--- a/flink-runtime-web/web-dashboard/src/app/pages/task-manager/thread-dump/task-manager-thread-dump.component.ts
+++ b/flink-runtime-web/web-dashboard/src/app/pages/task-manager/thread-dump/task-manager-thread-dump.component.ts
@@ -31,16 +31,36 @@ import {
} from '@flink-runtime-web/pages/task-manager/task-manager.config';
import { ConfigService, TaskManagerService } from '@flink-runtime-web/services';
import { editor } from 'monaco-editor';
+import { NzButtonModule } from 'ng-zorro-antd/button';
import { NzCodeEditorModule, EditorOptions } from 'ng-zorro-antd/code-editor';
+import { NzIconModule } from 'ng-zorro-antd/icon';
+import { NzSpaceModule } from 'ng-zorro-antd/space';
+import { NzTooltipModule } from 'ng-zorro-antd/tooltip';
import IStandaloneCodeEditor = editor.IStandaloneCodeEditor;
+/**
+ * Thread-dump mode controlling how much lock information is collected. Mirrors the server-side
+ * {@code ThreadDumpMode} enum. {@code undefined} defers to the cluster default
+ * (cluster.thread-dump.default-mode).
+ */
+type ThreadDumpMode = 'safe' | 'full' | undefined;
+
@Component({
selector: 'flink-task-manager-thread-dump',
templateUrl: './task-manager-thread-dump.component.html',
styleUrls: ['./task-manager-thread-dump.component.less'],
changeDetection: ChangeDetectionStrategy.OnPush,
- imports: [NzCodeEditorModule, AutoResizeDirective, FormsModule, AddonCompactComponent]
+ imports: [
+ NzCodeEditorModule,
+ AutoResizeDirective,
+ FormsModule,
+ AddonCompactComponent,
+ NzButtonModule,
+ NzIconModule,
+ NzSpaceModule,
+ NzTooltipModule
+ ]
})
export class TaskManagerThreadDumpComponent implements OnInit, OnDestroy {
public editorOptions: EditorOptions;
@@ -51,6 +71,12 @@ export class TaskManagerThreadDumpComponent implements OnInit, OnDestroy {
public taskManagerId: string;
public downloadUrl = '';
public downloadName = '';
+ /**
+ * Currently selected mode. Starts as undefined so the FIRST request honors the
+ * cluster default (cluster.thread-dump.default-mode); subsequent user interactions
+ * always set it to 'safe' or 'full'.
+ */
+ public mode: ThreadDumpMode = undefined;
private readonly destroy$ = new Subject();
@@ -66,8 +92,8 @@ export class TaskManagerThreadDumpComponent implements OnInit, OnDestroy {
public ngOnInit(): void {
this.taskManagerId = this.activatedRoute.parent!.snapshot.params.taskManagerId;
- this.downloadUrl = `${this.configService.BASE_URL}/taskmanagers/${this.taskManagerId}/thread-dump`;
this.downloadName = `taskmanager_${this.taskManagerId}_thread_dump`;
+ this.updateDownloadUrl();
this.activatedRoute.queryParams.subscribe(params => {
this.vertexName = decodeURIComponent(params.vertexName);
});
@@ -98,11 +124,26 @@ export class TaskManagerThreadDumpComponent implements OnInit, OnDestroy {
this.reload();
}
+ /**
+ * Switch dump mode. Intentionally does NOT auto-reload: the user must press the
+ * refresh button to actually request a new dump (especially important for 'full',
+ * which can be expensive). The download link is updated immediately so a download
+ * always matches the user's current selection.
+ */
+ public selectMode(mode: 'safe' | 'full'): void {
+ if (this.mode === mode) {
+ return;
+ }
+ this.mode = mode;
+ this.updateDownloadUrl();
+ this.cdr.markForCheck();
+ }
+
public reload(): void {
this.loading = true;
this.cdr.markForCheck();
this.taskManagerService
- .loadThreadDump(this.taskManagerId)
+ .loadThreadDump(this.taskManagerId, this.mode)
.pipe(
catchError(() => of('')),
takeUntil(this.destroy$)
@@ -113,4 +154,9 @@ export class TaskManagerThreadDumpComponent implements OnInit, OnDestroy {
this.cdr.markForCheck();
});
}
+
+ private updateDownloadUrl(): void {
+ const base = `${this.configService.BASE_URL}/taskmanagers/${this.taskManagerId}/thread-dump`;
+ this.downloadUrl = this.mode ? `${base}?mode=${this.mode}` : base;
+ }
}
diff --git a/flink-runtime-web/web-dashboard/src/app/services/job-manager.service.ts b/flink-runtime-web/web-dashboard/src/app/services/job-manager.service.ts
index 7a3525b6eb0872..6c33e829aca76d 100644
--- a/flink-runtime-web/web-dashboard/src/app/services/job-manager.service.ts
+++ b/flink-runtime-web/web-dashboard/src/app/services/job-manager.service.ts
@@ -82,8 +82,12 @@ export class JobManagerService {
);
}
- loadThreadDump(): Observable {
- return this.httpClient.get(`${this.configService.BASE_URL}/jobmanager/thread-dump`).pipe(
+ loadThreadDump(mode?: 'safe' | 'full'): Observable {
+ let url = `${this.configService.BASE_URL}/jobmanager/thread-dump`;
+ if (mode) {
+ url += `?mode=${mode}`;
+ }
+ return this.httpClient.get(url).pipe(
map(JobManagerThreadDump => {
return JobManagerThreadDump.threadInfos.map(threadInfo => threadInfo.stringifiedThreadInfo).join('');
})
diff --git a/flink-runtime-web/web-dashboard/src/app/services/task-manager.service.ts b/flink-runtime-web/web-dashboard/src/app/services/task-manager.service.ts
index 3a8c3227ddcf23..445353a8d6fd32 100644
--- a/flink-runtime-web/web-dashboard/src/app/services/task-manager.service.ts
+++ b/flink-runtime-web/web-dashboard/src/app/services/task-manager.service.ts
@@ -74,14 +74,16 @@ export class TaskManagerService {
);
}
- loadThreadDump(taskManagerId: string): Observable {
- return this.httpClient
- .get(`${this.configService.BASE_URL}/taskmanagers/${taskManagerId}/thread-dump`)
- .pipe(
- map(taskManagerThreadDump => {
- return taskManagerThreadDump.threadInfos.map(threadInfo => threadInfo.stringifiedThreadInfo).join('');
- })
- );
+ loadThreadDump(taskManagerId: string, mode?: 'safe' | 'full'): Observable {
+ let url = `${this.configService.BASE_URL}/taskmanagers/${taskManagerId}/thread-dump`;
+ if (mode) {
+ url += `?mode=${mode}`;
+ }
+ return this.httpClient.get(url).pipe(
+ map(taskManagerThreadDump => {
+ return taskManagerThreadDump.threadInfos.map(threadInfo => threadInfo.stringifiedThreadInfo).join('');
+ })
+ );
}
loadLogs(taskManagerId: string): Observable {
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..62334ab1e9e8a5 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
@@ -107,6 +107,7 @@
import org.apache.flink.runtime.rest.handler.async.OperationResult;
import org.apache.flink.runtime.rest.handler.job.AsynchronousJobOperationKey;
import org.apache.flink.runtime.rest.messages.ThreadDumpInfo;
+import org.apache.flink.runtime.rest.messages.ThreadDumpMode;
import org.apache.flink.runtime.rpc.FatalErrorHandler;
import org.apache.flink.runtime.rpc.FencedRpcEndpoint;
import org.apache.flink.runtime.rpc.RpcService;
@@ -1855,9 +1856,28 @@ 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));
+ public CompletableFuture requestThreadDump(
+ ThreadDumpMode mode, Duration timeout) {
+ final int stackTraceMaxDepth =
+ configuration.get(ClusterOptions.THREAD_DUMP_STACKTRACE_MAX_DEPTH);
+ final ThreadDumpMode resolvedMode =
+ ThreadDumpMode.resolve(
+ mode, configuration.get(ClusterOptions.THREAD_DUMP_DEFAULT_MODE));
+ return CompletableFuture.supplyAsync(
+ () -> {
+ final long startNanos = System.nanoTime();
+ final ThreadDumpInfo info =
+ ThreadDumpInfo.dumpAndCreate(stackTraceMaxDepth, resolvedMode);
+ if (log.isDebugEnabled()) {
+ log.debug(
+ "JobManager thread dump completed: mode={}, threads={}, elapsedMs={}",
+ resolvedMode,
+ info.getThreadInfos().size(),
+ (System.nanoTime() - startNanos) / 1_000_000L);
+ }
+ return info;
+ },
+ ioExecutor);
}
@Override
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java
index 13af32c8884422..d1f2c2a0a03d8f 100755
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java
@@ -62,6 +62,7 @@
import org.apache.flink.runtime.rest.messages.LogInfo;
import org.apache.flink.runtime.rest.messages.ProfilingInfo;
import org.apache.flink.runtime.rest.messages.ThreadDumpInfo;
+import org.apache.flink.runtime.rest.messages.ThreadDumpMode;
import org.apache.flink.runtime.rest.messages.taskmanager.TaskManagerInfo;
import org.apache.flink.runtime.rpc.FatalErrorHandler;
import org.apache.flink.runtime.rpc.FencedRpcEndpoint;
@@ -896,7 +897,7 @@ public CompletableFuture> listDataSe
@Override
public CompletableFuture requestThreadDump(
- ResourceID taskManagerId, Duration timeout) {
+ ResourceID taskManagerId, ThreadDumpMode mode, Duration timeout) {
final WorkerRegistration taskExecutor = taskExecutors.get(taskManagerId);
if (taskExecutor == null) {
@@ -906,7 +907,7 @@ public CompletableFuture requestThreadDump(
return FutureUtils.completedExceptionally(
new UnknownTaskExecutorException(taskManagerId));
} else {
- return taskExecutor.getTaskExecutorGateway().requestThreadDump(timeout);
+ return taskExecutor.getTaskExecutorGateway().requestThreadDump(mode, timeout);
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerGateway.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerGateway.java
index 938f3c58621bfd..ae147c9393c394 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerGateway.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerGateway.java
@@ -39,6 +39,7 @@
import org.apache.flink.runtime.rest.messages.ProfilingInfo;
import org.apache.flink.runtime.rest.messages.ProfilingInfo.ProfilingMode;
import org.apache.flink.runtime.rest.messages.ThreadDumpInfo;
+import org.apache.flink.runtime.rest.messages.ThreadDumpMode;
import org.apache.flink.runtime.rest.messages.taskmanager.TaskManagerInfo;
import org.apache.flink.runtime.rpc.FencedRpcGateway;
import org.apache.flink.runtime.rpc.RpcTimeout;
@@ -257,13 +258,13 @@ CompletableFuture> requestTaskManagerLogList(
/**
* Requests the thread dump from the given {@link TaskExecutor}.
*
- * @param taskManagerId taskManagerId identifying the {@link TaskExecutor} to get the thread
- * dump from
+ * @param taskManagerId identifies the {@link TaskExecutor} to dump
+ * @param mode dump granularity; when {@code null} the cluster default is used.
* @param timeout timeout of the asynchronous operation
* @return Future containing the thread dump information
*/
CompletableFuture requestThreadDump(
- ResourceID taskManagerId, @RpcTimeout Duration timeout);
+ ResourceID taskManagerId, ThreadDumpMode mode, @RpcTimeout Duration timeout);
/**
* Requests the {@link TaskExecutorGateway}.
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/cluster/JobManagerThreadDumpHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/cluster/JobManagerThreadDumpHandler.java
index 15dcbf322d8adc..d4e043a9dfef94 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/cluster/JobManagerThreadDumpHandler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/cluster/JobManagerThreadDumpHandler.java
@@ -21,10 +21,13 @@
import org.apache.flink.runtime.rest.handler.AbstractRestHandler;
import org.apache.flink.runtime.rest.handler.HandlerRequest;
import org.apache.flink.runtime.rest.handler.RestHandlerException;
-import org.apache.flink.runtime.rest.messages.EmptyMessageParameters;
+import org.apache.flink.runtime.rest.handler.util.HandlerRequestUtils;
import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
import org.apache.flink.runtime.rest.messages.MessageHeaders;
import org.apache.flink.runtime.rest.messages.ThreadDumpInfo;
+import org.apache.flink.runtime.rest.messages.ThreadDumpMode;
+import org.apache.flink.runtime.rest.messages.ThreadDumpModeQueryParameter;
+import org.apache.flink.runtime.rest.messages.cluster.JobManagerThreadDumpMessageParameters;
import org.apache.flink.runtime.webmonitor.RestfulGateway;
import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever;
@@ -37,13 +40,16 @@
/** Rest handler which serves the thread dump info from the JobManager. */
public class JobManagerThreadDumpHandler
extends AbstractRestHandler<
- RestfulGateway, EmptyRequestBody, ThreadDumpInfo, EmptyMessageParameters> {
+ RestfulGateway,
+ EmptyRequestBody,
+ ThreadDumpInfo,
+ JobManagerThreadDumpMessageParameters> {
public JobManagerThreadDumpHandler(
GatewayRetriever leaderRetriever,
Duration timeout,
Map responseHeaders,
- MessageHeaders
+ MessageHeaders
messageHeaders) {
super(leaderRetriever, timeout, responseHeaders, messageHeaders);
}
@@ -52,6 +58,9 @@ public JobManagerThreadDumpHandler(
protected CompletableFuture handleRequest(
@Nonnull HandlerRequest request, @Nonnull RestfulGateway gateway)
throws RestHandlerException {
- return gateway.requestThreadDump(timeout);
+ final ThreadDumpMode mode =
+ HandlerRequestUtils.getQueryParameter(
+ request, ThreadDumpModeQueryParameter.class, null);
+ return gateway.requestThreadDump(mode, timeout);
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/taskmanager/TaskManagerThreadDumpHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/taskmanager/TaskManagerThreadDumpHandler.java
index d0dc803262e84d..109c9511ec5fc9 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/taskmanager/TaskManagerThreadDumpHandler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/taskmanager/TaskManagerThreadDumpHandler.java
@@ -23,11 +23,14 @@
import org.apache.flink.runtime.rest.handler.HandlerRequest;
import org.apache.flink.runtime.rest.handler.RestHandlerException;
import org.apache.flink.runtime.rest.handler.resourcemanager.AbstractResourceManagerHandler;
+import org.apache.flink.runtime.rest.handler.util.HandlerRequestUtils;
import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
import org.apache.flink.runtime.rest.messages.MessageHeaders;
import org.apache.flink.runtime.rest.messages.ThreadDumpInfo;
+import org.apache.flink.runtime.rest.messages.ThreadDumpMode;
+import org.apache.flink.runtime.rest.messages.ThreadDumpModeQueryParameter;
import org.apache.flink.runtime.rest.messages.taskmanager.TaskManagerIdPathParameter;
-import org.apache.flink.runtime.rest.messages.taskmanager.TaskManagerMessageParameters;
+import org.apache.flink.runtime.rest.messages.taskmanager.TaskManagerThreadDumpMessageParameters;
import org.apache.flink.runtime.taskexecutor.TaskExecutor;
import org.apache.flink.runtime.webmonitor.RestfulGateway;
import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever;
@@ -41,13 +44,16 @@
/** Rest handler which serves the thread dump info from a {@link TaskExecutor}. */
public class TaskManagerThreadDumpHandler
extends AbstractResourceManagerHandler<
- RestfulGateway, EmptyRequestBody, ThreadDumpInfo, TaskManagerMessageParameters> {
+ RestfulGateway,
+ EmptyRequestBody,
+ ThreadDumpInfo,
+ TaskManagerThreadDumpMessageParameters> {
public TaskManagerThreadDumpHandler(
GatewayRetriever leaderRetriever,
Duration timeout,
Map responseHeaders,
- MessageHeaders
+ MessageHeaders
messageHeaders,
GatewayRetriever resourceManagerGatewayRetriever) {
super(
@@ -64,6 +70,9 @@ protected CompletableFuture handleRequest(
@Nonnull ResourceManagerGateway gateway)
throws RestHandlerException {
final ResourceID taskManagerId = request.getPathParameter(TaskManagerIdPathParameter.class);
- return gateway.requestThreadDump(taskManagerId, timeout);
+ final ThreadDumpMode mode =
+ HandlerRequestUtils.getQueryParameter(
+ request, ThreadDumpModeQueryParameter.class, null);
+ return gateway.requestThreadDump(taskManagerId, mode, timeout);
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ThreadDumpInfo.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ThreadDumpInfo.java
index e752ac297078d1..3409deadc4edd2 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ThreadDumpInfo.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ThreadDumpInfo.java
@@ -54,9 +54,11 @@ public static ThreadDumpInfo create(
return new ThreadDumpInfo(threadInfos);
}
- public static ThreadDumpInfo dumpAndCreate(int stacktraceMaxDepth) {
+ /** Dumps all threads of the current JVM at the granularity indicated by {@code mode}. */
+ public static ThreadDumpInfo dumpAndCreate(int stacktraceMaxDepth, ThreadDumpMode mode) {
return create(
- JvmUtils.createThreadDump().stream()
+ JvmUtils.createThreadDump(mode.isLockedMonitors(), mode.isLockedSynchronizers())
+ .stream()
.map(
threadInfo ->
ThreadDumpInfo.ThreadInfo.create(
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ThreadDumpMode.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ThreadDumpMode.java
new file mode 100644
index 00000000000000..95d1979147ae51
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ThreadDumpMode.java
@@ -0,0 +1,83 @@
+/*
+ * 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.rest.messages;
+
+/**
+ * Granularity of the thread dump collected via {@link
+ * java.lang.management.ThreadMXBean#dumpAllThreads(boolean, boolean)}. Information about the lock
+ * each thread is currently waiting on ({@link java.lang.management.ThreadInfo#getLockInfo()}) is
+ * populated in both modes.
+ */
+public enum ThreadDumpMode {
+
+ /**
+ * {@code dumpAllThreads(false, false)}: stack traces only, no lock info (jstack without {@code
+ * -l}). Negligible JVM pause.
+ */
+ SAFE(false, false),
+
+ /**
+ * {@code dumpAllThreads(true, true)}: also collects locked monitors and j.u.c. synchronizers
+ * (equivalent to {@code jstack -l}). Pauses the JVM in a safepoint for a duration that scales
+ * with heap size and thread count -- seconds on large TaskManagers.
+ */
+ FULL(true, true);
+
+ private final boolean lockedMonitors;
+ private final boolean lockedSynchronizers;
+
+ ThreadDumpMode(boolean lockedMonitors, boolean lockedSynchronizers) {
+ this.lockedMonitors = lockedMonitors;
+ this.lockedSynchronizers = lockedSynchronizers;
+ }
+
+ public boolean isLockedMonitors() {
+ return lockedMonitors;
+ }
+
+ public boolean isLockedSynchronizers() {
+ return lockedSynchronizers;
+ }
+
+ /**
+ * Resolves a mode name (case-insensitive), returning {@code fallback} when the input is
+ * null/blank or does not match a known mode.
+ */
+ public static ThreadDumpMode fromStringOrDefault(String name, ThreadDumpMode fallback) {
+ if (name == null || name.trim().isEmpty()) {
+ return fallback;
+ }
+ for (ThreadDumpMode m : values()) {
+ if (m.name().equalsIgnoreCase(name.trim())) {
+ return m;
+ }
+ }
+ return fallback;
+ }
+
+ /**
+ * Returns the effective mode for a REST request: {@code explicit} takes precedence; otherwise
+ * {@code clusterDefaultName} (typically the value of {@code cluster.thread-dump.default-mode})
+ * is parsed via {@link #fromStringOrDefault(String, ThreadDumpMode)}, falling back to {@link
+ * #FULL} for null/blank/unknown values (matching the historical behavior).
+ */
+ public static ThreadDumpMode resolve(ThreadDumpMode explicit, String clusterDefaultName) {
+ return explicit != null ? explicit : fromStringOrDefault(clusterDefaultName, FULL);
+ }
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ThreadDumpModeQueryParameter.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ThreadDumpModeQueryParameter.java
new file mode 100644
index 00000000000000..926747f443bb52
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ThreadDumpModeQueryParameter.java
@@ -0,0 +1,53 @@
+/*
+ * 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.rest.messages;
+
+import java.util.Arrays;
+
+/**
+ * Optional {@code mode} query parameter for the thread-dump REST endpoints. When omitted, the
+ * cluster default ({@code cluster.thread-dump.default-mode}) is used.
+ */
+public class ThreadDumpModeQueryParameter extends MessageQueryParameter {
+
+ public static final String KEY = "mode";
+
+ public ThreadDumpModeQueryParameter() {
+ super(KEY, MessageParameterRequisiteness.OPTIONAL);
+ }
+
+ @Override
+ public ThreadDumpMode convertStringToValue(String value) {
+ // Case-insensitive; unknown values throw IllegalArgumentException, which the REST layer
+ // translates into a 400 Bad Request.
+ return ThreadDumpMode.valueOf(value.trim().toUpperCase());
+ }
+
+ @Override
+ public String convertValueToString(ThreadDumpMode value) {
+ return value.name().toLowerCase();
+ }
+
+ @Override
+ public String getDescription() {
+ return "Controls how much lock information is collected. Supported values: "
+ + Arrays.toString(ThreadDumpMode.values())
+ + ". When omitted, cluster.thread-dump.default-mode is used.";
+ }
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/cluster/JobManagerThreadDumpHeaders.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/cluster/JobManagerThreadDumpHeaders.java
index f0dfba9439a749..b021d8f5889acb 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/cluster/JobManagerThreadDumpHeaders.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/cluster/JobManagerThreadDumpHeaders.java
@@ -20,7 +20,6 @@
import org.apache.flink.runtime.rest.HttpMethodWrapper;
import org.apache.flink.runtime.rest.handler.cluster.JobManagerThreadDumpHandler;
-import org.apache.flink.runtime.rest.messages.EmptyMessageParameters;
import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
import org.apache.flink.runtime.rest.messages.RuntimeMessageHeaders;
import org.apache.flink.runtime.rest.messages.ThreadDumpInfo;
@@ -29,7 +28,8 @@
/** Headers for the {@link JobManagerThreadDumpHandler}. */
public class JobManagerThreadDumpHeaders
- implements RuntimeMessageHeaders {
+ implements RuntimeMessageHeaders<
+ EmptyRequestBody, ThreadDumpInfo, JobManagerThreadDumpMessageParameters> {
private static final JobManagerThreadDumpHeaders INSTANCE = new JobManagerThreadDumpHeaders();
@@ -43,8 +43,8 @@ public Class getRequestClass() {
}
@Override
- public EmptyMessageParameters getUnresolvedMessageParameters() {
- return EmptyMessageParameters.getInstance();
+ public JobManagerThreadDumpMessageParameters getUnresolvedMessageParameters() {
+ return new JobManagerThreadDumpMessageParameters();
}
@Override
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/cluster/JobManagerThreadDumpMessageParameters.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/cluster/JobManagerThreadDumpMessageParameters.java
new file mode 100644
index 00000000000000..84459686691d18
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/cluster/JobManagerThreadDumpMessageParameters.java
@@ -0,0 +1,44 @@
+/*
+ * 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.rest.messages.cluster;
+
+import org.apache.flink.runtime.rest.messages.MessageParameters;
+import org.apache.flink.runtime.rest.messages.MessagePathParameter;
+import org.apache.flink.runtime.rest.messages.MessageQueryParameter;
+import org.apache.flink.runtime.rest.messages.ThreadDumpModeQueryParameter;
+
+import java.util.Collection;
+import java.util.Collections;
+
+/** Message parameters for the JobManager thread-dump REST handler. */
+public class JobManagerThreadDumpMessageParameters extends MessageParameters {
+
+ public final ThreadDumpModeQueryParameter modeQueryParameter =
+ new ThreadDumpModeQueryParameter();
+
+ @Override
+ public Collection> getPathParameters() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public Collection> getQueryParameters() {
+ return Collections.singleton(modeQueryParameter);
+ }
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/taskmanager/TaskManagerThreadDumpHeaders.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/taskmanager/TaskManagerThreadDumpHeaders.java
index 8695abb56fe405..22318cde5afb09 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/taskmanager/TaskManagerThreadDumpHeaders.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/taskmanager/TaskManagerThreadDumpHeaders.java
@@ -29,7 +29,7 @@
/** Headers for the {@link TaskManagerThreadDumpHandler}. */
public class TaskManagerThreadDumpHeaders
implements RuntimeMessageHeaders<
- EmptyRequestBody, ThreadDumpInfo, TaskManagerMessageParameters> {
+ EmptyRequestBody, ThreadDumpInfo, TaskManagerThreadDumpMessageParameters> {
private static final TaskManagerThreadDumpHeaders INSTANCE = new TaskManagerThreadDumpHeaders();
@@ -44,8 +44,8 @@ public Class getRequestClass() {
}
@Override
- public TaskManagerMessageParameters getUnresolvedMessageParameters() {
- return new TaskManagerMessageParameters();
+ public TaskManagerThreadDumpMessageParameters getUnresolvedMessageParameters() {
+ return new TaskManagerThreadDumpMessageParameters();
}
@Override
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/taskmanager/TaskManagerThreadDumpMessageParameters.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/taskmanager/TaskManagerThreadDumpMessageParameters.java
new file mode 100644
index 00000000000000..1079212835a7f8
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/taskmanager/TaskManagerThreadDumpMessageParameters.java
@@ -0,0 +1,37 @@
+/*
+ * 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.rest.messages.taskmanager;
+
+import org.apache.flink.runtime.rest.messages.MessageQueryParameter;
+import org.apache.flink.runtime.rest.messages.ThreadDumpModeQueryParameter;
+
+import java.util.Collection;
+import java.util.Collections;
+
+/** Message parameters for the TaskManager thread-dump REST handler. */
+public class TaskManagerThreadDumpMessageParameters extends TaskManagerMessageParameters {
+
+ public final ThreadDumpModeQueryParameter modeQueryParameter =
+ new ThreadDumpModeQueryParameter();
+
+ @Override
+ public Collection> getQueryParameters() {
+ return Collections.singleton(modeQueryParameter);
+ }
+}
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..2e33ee1655a04c 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
@@ -23,6 +23,7 @@
import org.apache.flink.api.common.JobID;
import org.apache.flink.configuration.BatchExecutionOptions;
import org.apache.flink.configuration.ClusterOptions;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.management.jmx.JMXService;
import org.apache.flink.runtime.accumulators.AccumulatorSnapshot;
import org.apache.flink.runtime.blob.JobPermanentBlobService;
@@ -95,6 +96,7 @@
import org.apache.flink.runtime.rest.messages.LogInfo;
import org.apache.flink.runtime.rest.messages.ProfilingInfo;
import org.apache.flink.runtime.rest.messages.ThreadDumpInfo;
+import org.apache.flink.runtime.rest.messages.ThreadDumpMode;
import org.apache.flink.runtime.rpc.FatalErrorHandler;
import org.apache.flink.runtime.rpc.RpcEndpoint;
import org.apache.flink.runtime.rpc.RpcService;
@@ -1460,12 +1462,27 @@ public CompletableFuture sendOperatorEventToTask(
}
@Override
- public CompletableFuture requestThreadDump(Duration timeout) {
- int stacktraceMaxDepth =
- taskManagerConfiguration
- .getConfiguration()
- .get(ClusterOptions.THREAD_DUMP_STACKTRACE_MAX_DEPTH);
- return CompletableFuture.completedFuture(ThreadDumpInfo.dumpAndCreate(stacktraceMaxDepth));
+ public CompletableFuture requestThreadDump(
+ ThreadDumpMode mode, Duration timeout) {
+ final Configuration config = taskManagerConfiguration.getConfiguration();
+ final int stacktraceMaxDepth = config.get(ClusterOptions.THREAD_DUMP_STACKTRACE_MAX_DEPTH);
+ final ThreadDumpMode resolvedMode =
+ ThreadDumpMode.resolve(mode, config.get(ClusterOptions.THREAD_DUMP_DEFAULT_MODE));
+ return CompletableFuture.supplyAsync(
+ () -> {
+ final long startNanos = System.nanoTime();
+ final ThreadDumpInfo info =
+ ThreadDumpInfo.dumpAndCreate(stacktraceMaxDepth, resolvedMode);
+ if (log.isDebugEnabled()) {
+ log.debug(
+ "TaskExecutor thread dump completed: mode={}, threads={}, elapsedMs={}",
+ resolvedMode,
+ info.getThreadInfos().size(),
+ (System.nanoTime() - startNanos) / 1_000_000L);
+ }
+ return info;
+ },
+ ioExecutor);
}
@Override
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorGateway.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorGateway.java
index 29607c2c2f8130..a330682f6a2af0 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorGateway.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorGateway.java
@@ -41,6 +41,7 @@
import org.apache.flink.runtime.rest.messages.LogInfo;
import org.apache.flink.runtime.rest.messages.ProfilingInfo;
import org.apache.flink.runtime.rest.messages.ThreadDumpInfo;
+import org.apache.flink.runtime.rest.messages.ThreadDumpMode;
import org.apache.flink.runtime.rpc.RpcGateway;
import org.apache.flink.runtime.rpc.RpcTimeout;
import org.apache.flink.runtime.shuffle.PartitionWithMetrics;
@@ -306,10 +307,12 @@ CompletableFuture sendOperatorEventToTask(
/**
* Requests the thread dump from this TaskManager.
*
+ * @param mode dump granularity; when {@code null} the cluster default is used.
* @param timeout timeout for the asynchronous operation
* @return the {@link ThreadDumpInfo} for this TaskManager.
*/
- CompletableFuture requestThreadDump(@RpcTimeout Duration timeout);
+ CompletableFuture requestThreadDump(
+ ThreadDumpMode mode, @RpcTimeout Duration timeout);
/**
* Sends new delegation tokens to this TaskManager.
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorGatewayDecoratorBase.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorGatewayDecoratorBase.java
index 5d33ddd65a05e9..1f40bfaa61a6bf 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorGatewayDecoratorBase.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorGatewayDecoratorBase.java
@@ -41,6 +41,7 @@
import org.apache.flink.runtime.rest.messages.LogInfo;
import org.apache.flink.runtime.rest.messages.ProfilingInfo;
import org.apache.flink.runtime.rest.messages.ThreadDumpInfo;
+import org.apache.flink.runtime.rest.messages.ThreadDumpMode;
import org.apache.flink.runtime.webmonitor.threadinfo.ThreadInfoSamplesRequest;
import org.apache.flink.types.SerializableOptional;
import org.apache.flink.util.SerializedValue;
@@ -238,8 +239,9 @@ public CompletableFuture sendOperatorEventToTask(
}
@Override
- public CompletableFuture requestThreadDump(Duration timeout) {
- return originalGateway.requestThreadDump(timeout);
+ public CompletableFuture requestThreadDump(
+ ThreadDumpMode mode, Duration timeout) {
+ return originalGateway.requestThreadDump(mode, timeout);
}
@Override
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/util/JvmUtils.java b/flink-runtime/src/main/java/org/apache/flink/runtime/util/JvmUtils.java
index 89a8f8fe0cc3d5..6781b7f02a62a4 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/util/JvmUtils.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/util/JvmUtils.java
@@ -40,14 +40,14 @@ public final class JvmUtils {
private static final Logger LOG = LoggerFactory.getLogger(JvmUtils.class);
/**
- * Creates a thread dump of the current JVM.
- *
- * @return the thread dump of current JVM
+ * Creates a thread dump of the current JVM. See {@link
+ * java.lang.management.ThreadMXBean#dumpAllThreads(boolean, boolean)} for the semantics of the
+ * two flags.
*/
- public static Collection createThreadDump() {
+ public static Collection createThreadDump(
+ boolean lockedMonitors, boolean lockedSynchronizers) {
ThreadMXBean threadMxBean = ManagementFactory.getThreadMXBean();
-
- return Arrays.asList(threadMxBean.dumpAllThreads(true, true));
+ return Arrays.asList(threadMxBean.dumpAllThreads(lockedMonitors, lockedSynchronizers));
}
/**
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/NonLeaderRetrievalRestfulGateway.java b/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/NonLeaderRetrievalRestfulGateway.java
index e162a2823f93d9..d5f4114018a9cf 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/NonLeaderRetrievalRestfulGateway.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/NonLeaderRetrievalRestfulGateway.java
@@ -30,6 +30,7 @@
import org.apache.flink.runtime.messages.webmonitor.MultipleApplicationsDetails;
import org.apache.flink.runtime.messages.webmonitor.MultipleJobsDetails;
import org.apache.flink.runtime.rest.messages.ThreadDumpInfo;
+import org.apache.flink.runtime.rest.messages.ThreadDumpMode;
import org.apache.flink.runtime.scheduler.ExecutionGraphInfo;
import java.time.Duration;
@@ -125,7 +126,8 @@ public CompletableFuture> requestMetricQueryServiceAddresses(
}
@Override
- public CompletableFuture requestThreadDump(Duration timeout) {
+ public CompletableFuture requestThreadDump(
+ ThreadDumpMode mode, Duration timeout) {
throw new UnsupportedOperationException(MESSAGE);
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/RestfulGateway.java b/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/RestfulGateway.java
index 62cd830a6535cd..5191941a07ff3b 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/RestfulGateway.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/RestfulGateway.java
@@ -44,6 +44,7 @@
import org.apache.flink.runtime.rest.handler.async.OperationResult;
import org.apache.flink.runtime.rest.handler.job.AsynchronousJobOperationKey;
import org.apache.flink.runtime.rest.messages.ThreadDumpInfo;
+import org.apache.flink.runtime.rest.messages.ThreadDumpMode;
import org.apache.flink.runtime.rpc.RpcGateway;
import org.apache.flink.runtime.rpc.RpcTimeout;
import org.apache.flink.runtime.scheduler.ExecutionGraphInfo;
@@ -188,10 +189,12 @@ CompletableFuture> requestMetricQueryServiceAddresses(
/**
* Requests the thread dump from the JobManager.
*
+ * @param mode dump granularity; when {@code null} the cluster default is used.
* @param timeout timeout of the asynchronous operation
* @return Future containing the thread dump information
*/
- CompletableFuture requestThreadDump(@RpcTimeout Duration timeout);
+ CompletableFuture requestThreadDump(
+ ThreadDumpMode mode, @RpcTimeout Duration timeout);
/**
* Triggers a checkpoint with the given savepoint directory as a target.
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/utils/TestingResourceManagerGateway.java b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/utils/TestingResourceManagerGateway.java
index 603eee966255e2..f7c691ae8a7aea 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/utils/TestingResourceManagerGateway.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/utils/TestingResourceManagerGateway.java
@@ -46,6 +46,7 @@
import org.apache.flink.runtime.rest.messages.LogInfo;
import org.apache.flink.runtime.rest.messages.ProfilingInfo;
import org.apache.flink.runtime.rest.messages.ThreadDumpInfo;
+import org.apache.flink.runtime.rest.messages.ThreadDumpMode;
import org.apache.flink.runtime.rest.messages.taskmanager.TaskManagerInfo;
import org.apache.flink.runtime.shuffle.ShuffleDescriptor;
import org.apache.flink.runtime.slots.ResourceRequirements;
@@ -475,7 +476,7 @@ public CompletableFuture> requestTaskManagerLogList(
@Override
public CompletableFuture requestThreadDump(
- ResourceID taskManagerId, Duration timeout) {
+ ResourceID taskManagerId, ThreadDumpMode mode, Duration timeout) {
final Function> function =
this.requestThreadDumpFunction;
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/ThreadDumpModeQueryParameterTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/ThreadDumpModeQueryParameterTest.java
new file mode 100644
index 00000000000000..b326717936d48b
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/ThreadDumpModeQueryParameterTest.java
@@ -0,0 +1,66 @@
+/*
+ * 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.rest.messages;
+
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link ThreadDumpModeQueryParameter}. */
+class ThreadDumpModeQueryParameterTest {
+
+ private final ThreadDumpModeQueryParameter param = new ThreadDumpModeQueryParameter();
+
+ @Test
+ void keyIsMode() {
+ assertThat(param.getKey()).isEqualTo("mode");
+ }
+
+ @Test
+ void parameterIsOptional() {
+ assertThat(param.isMandatory()).isFalse();
+ }
+
+ @Test
+ void parsesSafeCaseInsensitively() {
+ assertThat(param.convertStringToValue("safe")).isEqualTo(ThreadDumpMode.SAFE);
+ assertThat(param.convertStringToValue("SAFE")).isEqualTo(ThreadDumpMode.SAFE);
+ assertThat(param.convertStringToValue(" Safe ")).isEqualTo(ThreadDumpMode.SAFE);
+ }
+
+ @Test
+ void parsesFullCaseInsensitively() {
+ assertThat(param.convertStringToValue("full")).isEqualTo(ThreadDumpMode.FULL);
+ assertThat(param.convertStringToValue("FULL")).isEqualTo(ThreadDumpMode.FULL);
+ }
+
+ @Test
+ void rejectsUnknownValueWithIllegalArgumentException() {
+ // The handler layer (RestHandlerUtils#convertQueryParameter) turns this into a 400.
+ assertThatThrownBy(() -> param.convertStringToValue("nonsense"))
+ .isInstanceOf(IllegalArgumentException.class);
+ }
+
+ @Test
+ void convertValueToStringIsLowerCase() {
+ assertThat(param.convertValueToString(ThreadDumpMode.SAFE)).isEqualTo("safe");
+ assertThat(param.convertValueToString(ThreadDumpMode.FULL)).isEqualTo("full");
+ }
+}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorThreadDumpTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorThreadDumpTest.java
new file mode 100644
index 00000000000000..c57c4f3f61deda
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorThreadDumpTest.java
@@ -0,0 +1,105 @@
+/*
+ * 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.configuration.ClusterOptions;
+import org.apache.flink.configuration.Configuration;
+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.rest.messages.ThreadDumpMode;
+import org.apache.flink.runtime.rpc.RpcUtils;
+import org.apache.flink.runtime.rpc.TestingRpcServiceExtension;
+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.TimeUnit;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests for {@link TaskExecutor#requestThreadDump(ThreadDumpMode, Duration)}: verifies that the RPC
+ * returns a non-empty dump for each supported mode and honors the {@code
+ * cluster.thread-dump.default-mode} fallback.
+ */
+@ExtendWith(TestLoggerExtension.class)
+class TaskExecutorThreadDumpTest {
+
+ private static final Duration RPC_TIMEOUT = Duration.ofSeconds(10);
+
+ private final TestingRpcServiceExtension rpcServiceExtension = new TestingRpcServiceExtension();
+
+ @RegisterExtension
+ private final EachCallbackWrapper eachWrapper =
+ new EachCallbackWrapper<>(rpcServiceExtension);
+
+ @Test
+ void requestThreadDumpReturnsNonEmptyDumpForEachMode(@TempDir File tempDir) throws Exception {
+ for (ThreadDumpMode mode :
+ new ThreadDumpMode[] {null, ThreadDumpMode.SAFE, ThreadDumpMode.FULL}) {
+ final ThreadDumpInfo dump = requestDump(tempDir, new Configuration(), mode);
+ assertThat(dump.getThreadInfos()).isNotEmpty();
+ }
+ }
+
+ @Test
+ void requestThreadDumpFallsBackToClusterDefaultWhenModeOmitted(@TempDir File tempDir)
+ throws Exception {
+ // Override the cluster default to SAFE so the assertion below distinguishes "config was
+ // honored" from "hardcoded FULL". SAFE omits the "Number of locked synchronizers" section
+ // that FULL always emits.
+ final Configuration configuration = new Configuration();
+ configuration.set(ClusterOptions.THREAD_DUMP_DEFAULT_MODE, "SAFE");
+
+ final ThreadDumpInfo dump = requestDump(tempDir, configuration, /* mode= */ null);
+
+ assertThat(dump.getThreadInfos())
+ .noneMatch(
+ t ->
+ t.getStringifiedThreadInfo()
+ .contains("Number of locked synchronizers"));
+ }
+
+ private ThreadDumpInfo requestDump(
+ File tempDir, Configuration configuration, ThreadDumpMode mode) throws Exception {
+ final TaskExecutor taskExecutor =
+ TaskExecutorBuilder.newBuilder(
+ rpcServiceExtension.getTestingRpcService(),
+ new TestingHighAvailabilityServicesBuilder().build(),
+ WorkingDirectory.create(tempDir))
+ .setConfiguration(configuration)
+ .build();
+ try {
+ taskExecutor.start();
+ return taskExecutor
+ .getSelfGateway(TaskExecutorGateway.class)
+ .requestThreadDump(mode, RPC_TIMEOUT)
+ .get(20, TimeUnit.SECONDS);
+ } finally {
+ RpcUtils.terminateRpcEndpoint(taskExecutor);
+ }
+ }
+}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TestingTaskExecutorGateway.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TestingTaskExecutorGateway.java
index cf1b79c68e643a..77243b8e028cf4 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TestingTaskExecutorGateway.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TestingTaskExecutorGateway.java
@@ -42,6 +42,7 @@
import org.apache.flink.runtime.rest.messages.LogInfo;
import org.apache.flink.runtime.rest.messages.ProfilingInfo;
import org.apache.flink.runtime.rest.messages.ThreadDumpInfo;
+import org.apache.flink.runtime.rest.messages.ThreadDumpMode;
import org.apache.flink.runtime.shuffle.PartitionWithMetrics;
import org.apache.flink.runtime.webmonitor.threadinfo.ThreadInfoSamplesRequest;
import org.apache.flink.types.SerializableOptional;
@@ -371,7 +372,8 @@ public CompletableFuture sendOperatorEventToTask(
}
@Override
- public CompletableFuture requestThreadDump(Duration timeout) {
+ public CompletableFuture requestThreadDump(
+ ThreadDumpMode mode, Duration timeout) {
return requestThreadDumpSupplier.get();
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/TestingRestfulGateway.java b/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/TestingRestfulGateway.java
index ac1b46b1e07be1..f6eac7b17b7b6c 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/TestingRestfulGateway.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/TestingRestfulGateway.java
@@ -40,6 +40,7 @@
import org.apache.flink.runtime.rest.handler.async.OperationResult;
import org.apache.flink.runtime.rest.handler.job.AsynchronousJobOperationKey;
import org.apache.flink.runtime.rest.messages.ThreadDumpInfo;
+import org.apache.flink.runtime.rest.messages.ThreadDumpMode;
import org.apache.flink.runtime.scheduler.ExecutionGraphInfo;
import org.apache.flink.util.SerializedValue;
import org.apache.flink.util.concurrent.FutureUtils;
@@ -406,7 +407,8 @@ public CompletableFuture> requestMetricQueryServiceAddresses(
}
@Override
- public CompletableFuture requestThreadDump(Duration timeout) {
+ public CompletableFuture requestThreadDump(
+ ThreadDumpMode mode, Duration timeout) {
return null;
}