diff --git a/docs/static/generated/rest_v1_dispatcher.yml b/docs/static/generated/rest_v1_dispatcher.yml
index 24ecb76ed2e961..0c94fafc052e56 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:\
+ \ [LITE, 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:\
+ \ [LITE, 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:
+ - LITE
+ - 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..6d03db19229fb0 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,24 @@ 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")
+ .enumType(ThreadDumpMode.class)
+ .defaultValue(ThreadDumpMode.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(
+ "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("LITE"))
+ .build());
+
@Documentation.Section(Documentation.Sections.EXPERT_CLUSTER)
public static final ConfigOption UNCAUGHT_EXCEPTION_HANDLING =
ConfigOptions.key("cluster.uncaught-exception-handling")
diff --git a/flink-core/src/main/java/org/apache/flink/configuration/ThreadDumpMode.java b/flink-core/src/main/java/org/apache/flink/configuration/ThreadDumpMode.java
new file mode 100644
index 00000000000000..caa58747c144f6
--- /dev/null
+++ b/flink-core/src/main/java/org/apache/flink/configuration/ThreadDumpMode.java
@@ -0,0 +1,78 @@
+/*
+ * 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.configuration;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.description.InlineElement;
+
+import static org.apache.flink.configuration.description.TextElement.text;
+
+/**
+ * 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.
+ *
+ * @see ClusterOptions#THREAD_DUMP_DEFAULT_MODE
+ */
+@PublicEvolving
+public enum ThreadDumpMode implements DescribedEnum {
+
+ /**
+ * {@code dumpAllThreads(false, false)}: stack traces only, no lock info (jstack without {@code
+ * -l}). Negligible JVM pause.
+ */
+ LITE(false, false, text("Stack traces only, without lock information. Negligible JVM pause.")),
+
+ /**
+ * {@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,
+ text(
+ "Additionally collects 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, which can take seconds on large TaskManagers."));
+
+ private final boolean lockedMonitors;
+ private final boolean lockedSynchronizers;
+ private final InlineElement description;
+
+ ThreadDumpMode(boolean lockedMonitors, boolean lockedSynchronizers, InlineElement description) {
+ this.lockedMonitors = lockedMonitors;
+ this.lockedSynchronizers = lockedSynchronizers;
+ this.description = description;
+ }
+
+ public boolean isLockedMonitors() {
+ return lockedMonitors;
+ }
+
+ public boolean isLockedSynchronizers() {
+ return lockedSynchronizers;
+ }
+
+ @Override
+ public InlineElement getDescription() {
+ return description;
+ }
+}
diff --git a/flink-core/src/test/java/org/apache/flink/configuration/ClusterOptionsTest.java b/flink-core/src/test/java/org/apache/flink/configuration/ClusterOptionsTest.java
new file mode 100644
index 00000000000000..40e984b4b38ff6
--- /dev/null
+++ b/flink-core/src/test/java/org/apache/flink/configuration/ClusterOptionsTest.java
@@ -0,0 +1,64 @@
+/*
+ * 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.configuration;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link ClusterOptions}. */
+class ClusterOptionsTest {
+
+ @Test
+ void testThreadDumpDefaultModeDefaultsToFull() {
+ assertThat(new Configuration().get(ClusterOptions.THREAD_DUMP_DEFAULT_MODE))
+ .isEqualTo(ThreadDumpMode.FULL);
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"LITE", "lite", "Lite"})
+ void testThreadDumpDefaultModeIsCaseInsensitive(String value) {
+ final Configuration configuration = new Configuration();
+ configuration.setString(ClusterOptions.THREAD_DUMP_DEFAULT_MODE.key(), value);
+
+ assertThat(configuration.get(ClusterOptions.THREAD_DUMP_DEFAULT_MODE))
+ .isEqualTo(ThreadDumpMode.LITE);
+ }
+
+ /**
+ * An unparsable value must fail fast instead of silently falling back to {@link
+ * ThreadDumpMode#FULL}, which is the mode operators configure this option to avoid.
+ */
+ @ParameterizedTest
+ @ValueSource(strings = {"Ltie", "", " ", "none"})
+ void testThreadDumpDefaultModeRejectsUnknownValue(String value) {
+ final Configuration configuration = new Configuration();
+ configuration.setString(ClusterOptions.THREAD_DUMP_DEFAULT_MODE.key(), value);
+
+ assertThatThrownBy(() -> configuration.get(ClusterOptions.THREAD_DUMP_DEFAULT_MODE))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining(ClusterOptions.THREAD_DUMP_DEFAULT_MODE.key())
+ .cause()
+ .hasMessageContaining(ThreadDumpMode.LITE.name())
+ .hasMessageContaining(ThreadDumpMode.FULL.name());
+ }
+}
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..a7fbfc490705e0 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: 'lite' | '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..d0cc1bb6df35bf 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..2474f66928f84a 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 = 'lite' | '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 'lite' 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: 'lite' | '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 52064a870810be..c70ae01971bc4b 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
@@ -85,8 +85,12 @@ export class JobManagerService {
);
}
- loadThreadDump(): Observable {
- return this.httpClient.get(`${this.configService.BASE_URL}/jobmanager/thread-dump`).pipe(
+ loadThreadDump(mode?: 'lite' | '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 b5ece679d031ee..3d3ff96172c4cd 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
@@ -77,14 +77,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?: 'lite' | '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 50bb43285e0c8b..f64645b196bef9 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
@@ -34,6 +34,7 @@
import org.apache.flink.configuration.DeploymentOptions;
import org.apache.flink.configuration.HighAvailabilityOptions;
import org.apache.flink.configuration.PipelineOptions;
+import org.apache.flink.configuration.ThreadDumpMode;
import org.apache.flink.configuration.WebOptions;
import org.apache.flink.core.execution.CheckpointType;
import org.apache.flink.core.execution.SavepointFormatType;
@@ -1864,10 +1865,14 @@ public CompletableFuture> requestMetricQueryServiceAddresses(
}
@Override
- public CompletableFuture requestThreadDump(Duration timeout) {
- int stackTraceMaxDepth = configuration.get(ClusterOptions.THREAD_DUMP_STACKTRACE_MAX_DEPTH);
+ public CompletableFuture requestThreadDump(
+ ThreadDumpMode mode, Duration timeout) {
+ final int stackTraceMaxDepth =
+ configuration.get(ClusterOptions.THREAD_DUMP_STACKTRACE_MAX_DEPTH);
+ final ThreadDumpMode resolvedMode =
+ mode != null ? mode : configuration.get(ClusterOptions.THREAD_DUMP_DEFAULT_MODE);
return CompletableFuture.supplyAsync(
- () -> ThreadDumpInfo.dumpAndCreate(stackTraceMaxDepth), ioExecutor);
+ () -> ThreadDumpInfo.dumpAndCreate(stackTraceMaxDepth, resolvedMode), 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..5cfb06b3bef50c 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
@@ -22,6 +22,7 @@
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.JobStatus;
import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.ThreadDumpMode;
import org.apache.flink.runtime.blob.TransientBlobKey;
import org.apache.flink.runtime.blocklist.BlockedNode;
import org.apache.flink.runtime.blocklist.BlocklistContext;
@@ -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..9c722604ebc74e 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
@@ -21,6 +21,7 @@
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.JobStatus;
import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.ThreadDumpMode;
import org.apache.flink.runtime.blob.BlobServer;
import org.apache.flink.runtime.blob.TransientBlobKey;
import org.apache.flink.runtime.blocklist.BlocklistListener;
@@ -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..368a0a6302609d 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
@@ -18,13 +18,16 @@
package org.apache.flink.runtime.rest.handler.cluster;
+import org.apache.flink.configuration.ThreadDumpMode;
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.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..cc7472c06f9772 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
@@ -18,16 +18,19 @@
package org.apache.flink.runtime.rest.handler.taskmanager;
+import org.apache.flink.configuration.ThreadDumpMode;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.resourcemanager.ResourceManagerGateway;
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.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..25ddd90f249ed7 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
@@ -19,6 +19,7 @@
package org.apache.flink.runtime.rest.messages;
import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.configuration.ThreadDumpMode;
import org.apache.flink.runtime.util.JvmUtils;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
@@ -54,9 +55,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/ThreadDumpModeQueryParameter.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ThreadDumpModeQueryParameter.java
new file mode 100644
index 00000000000000..548bc6a70d5412
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ThreadDumpModeQueryParameter.java
@@ -0,0 +1,56 @@
+/*
+ * 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.apache.flink.configuration.ThreadDumpMode;
+
+import java.util.Arrays;
+import java.util.Locale;
+
+/**
+ * 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(Locale.ROOT));
+ }
+
+ @Override
+ public String convertValueToString(ThreadDumpMode value) {
+ return value.name().toLowerCase(Locale.ROOT);
+ }
+
+ @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 a7558bf6d3f1fb..b8609c937bb3ab 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,8 @@
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.configuration.ThreadDumpMode;
import org.apache.flink.management.jmx.JMXService;
import org.apache.flink.runtime.accumulators.AccumulatorSnapshot;
import org.apache.flink.runtime.blob.JobPermanentBlobService;
@@ -1466,13 +1468,14 @@ public CompletableFuture sendOperatorEventToTask(
}
@Override
- public CompletableFuture requestThreadDump(Duration timeout) {
- int stacktraceMaxDepth =
- taskManagerConfiguration
- .getConfiguration()
- .get(ClusterOptions.THREAD_DUMP_STACKTRACE_MAX_DEPTH);
+ 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 =
+ mode != null ? mode : config.get(ClusterOptions.THREAD_DUMP_DEFAULT_MODE);
return CompletableFuture.supplyAsync(
- () -> ThreadDumpInfo.dumpAndCreate(stacktraceMaxDepth), ioExecutor);
+ () -> ThreadDumpInfo.dumpAndCreate(stacktraceMaxDepth, resolvedMode), 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..cb13459a47a98d 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
@@ -20,6 +20,7 @@
import org.apache.flink.api.common.ApplicationID;
import org.apache.flink.api.common.JobID;
+import org.apache.flink.configuration.ThreadDumpMode;
import org.apache.flink.runtime.blob.BlobServer;
import org.apache.flink.runtime.blob.TransientBlobKey;
import org.apache.flink.runtime.checkpoint.CheckpointOptions;
@@ -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..cb46965f9c9c76 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
@@ -20,6 +20,7 @@
import org.apache.flink.api.common.ApplicationID;
import org.apache.flink.api.common.JobID;
+import org.apache.flink.configuration.ThreadDumpMode;
import org.apache.flink.runtime.blob.TransientBlobKey;
import org.apache.flink.runtime.checkpoint.CheckpointOptions;
import org.apache.flink.runtime.clusterframework.types.AllocationID;
@@ -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..2455128e50a62d 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
@@ -21,6 +21,7 @@
import org.apache.flink.api.common.ApplicationID;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.ThreadDumpMode;
import org.apache.flink.runtime.application.ArchivedApplication;
import org.apache.flink.runtime.checkpoint.CheckpointStatsSnapshot;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
@@ -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..d02a3d12c05d49 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
@@ -22,6 +22,7 @@
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.JobStatus;
import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.ThreadDumpMode;
import org.apache.flink.core.execution.CheckpointType;
import org.apache.flink.core.execution.SavepointFormatType;
import org.apache.flink.runtime.application.ArchivedApplication;
@@ -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/dispatcher/DispatcherThreadDumpOffloadTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherThreadDumpOffloadTest.java
index 111e52a1414e25..7497a3bba18313 100644
--- 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
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.dispatcher;
+import org.apache.flink.configuration.ThreadDumpMode;
import org.apache.flink.runtime.rest.messages.ThreadDumpInfo;
import org.apache.flink.runtime.rpc.RpcUtils;
@@ -62,7 +63,7 @@ public void requestThreadDumpRunsOnIoExecutor() throws Exception {
final ThreadDumpInfo dump =
dispatcher
.getSelfGateway(DispatcherGateway.class)
- .requestThreadDump(TIMEOUT)
+ .requestThreadDump(ThreadDumpMode.FULL, TIMEOUT)
.get(20, TimeUnit.SECONDS);
assertThat(dump.getThreadInfos())
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..bd38d99eb7f55f 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
@@ -22,6 +22,7 @@
import org.apache.flink.api.common.JobStatus;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.configuration.ThreadDumpMode;
import org.apache.flink.runtime.blob.TransientBlobKey;
import org.apache.flink.runtime.blocklist.BlockedNode;
import org.apache.flink.runtime.clusterframework.ApplicationStatus;
@@ -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..4c7ca442e9e561
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/ThreadDumpModeQueryParameterTest.java
@@ -0,0 +1,68 @@
+/*
+ * 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.apache.flink.configuration.ThreadDumpMode;
+
+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 parsesLiteCaseInsensitively() {
+ assertThat(param.convertStringToValue("lite")).isEqualTo(ThreadDumpMode.LITE);
+ assertThat(param.convertStringToValue("LITE")).isEqualTo(ThreadDumpMode.LITE);
+ assertThat(param.convertStringToValue(" Lite ")).isEqualTo(ThreadDumpMode.LITE);
+ }
+
+ @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.LITE)).isEqualTo("lite");
+ assertThat(param.convertValueToString(ThreadDumpMode.FULL)).isEqualTo("full");
+ }
+}
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
index 5257890563c053..9af73c5c1e8d71 100644
--- 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
@@ -18,6 +18,9 @@
package org.apache.flink.runtime.taskexecutor;
+import org.apache.flink.configuration.ClusterOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.ThreadDumpMode;
import org.apache.flink.core.testutils.EachCallbackWrapper;
import org.apache.flink.runtime.entrypoint.WorkingDirectory;
import org.apache.flink.runtime.highavailability.TestingHighAvailabilityServicesBuilder;
@@ -31,6 +34,8 @@
import org.junit.jupiter.api.extension.ExtendWith;
import org.junit.jupiter.api.extension.RegisterExtension;
import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
import java.io.File;
import java.time.Duration;
@@ -40,7 +45,12 @@
import static org.assertj.core.api.Assertions.assertThat;
-/** Tests that {@link TaskExecutor#requestThreadDump(Duration)} is offloaded to the ioExecutor. */
+/**
+ * Tests for {@link TaskExecutor#requestThreadDump(ThreadDumpMode, Duration)}: the dump is offloaded
+ * to the ioExecutor instead of blocking the main thread, returns a non-empty result for every
+ * supported mode, and falls back to {@link ClusterOptions#THREAD_DUMP_DEFAULT_MODE} when the
+ * request omits the mode.
+ */
@ExtendWith(TestLoggerExtension.class)
class TaskExecutorThreadDumpOffloadTest {
@@ -55,8 +65,44 @@ class TaskExecutorThreadDumpOffloadTest {
@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 ThreadDumpInfo dump = requestDump(tempDir, new Configuration(), ThreadDumpMode.FULL);
+
+ assertThat(dump.getThreadInfos())
+ .as("dump must include ioExecutor thread '%s' (proves offload)", IO_THREAD_NAME)
+ .anyMatch(t -> IO_THREAD_NAME.equals(t.getThreadName()));
+ }
+
+ @ParameterizedTest
+ @EnumSource(ThreadDumpMode.class)
+ void requestThreadDumpReturnsNonEmptyDumpForEachMode(ThreadDumpMode mode, @TempDir File tempDir)
+ throws Exception {
+ 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 LITE so the assertion below distinguishes "config was
+ // honored" from "hardcoded FULL". LITE omits the "Number of locked synchronizers" section
+ // that FULL always emits.
+ final Configuration configuration = new Configuration();
+ configuration.set(ClusterOptions.THREAD_DUMP_DEFAULT_MODE, ThreadDumpMode.LITE);
+
+ 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 {
+ // 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 {
@@ -72,22 +118,15 @@ void requestThreadDumpRunsOnIoExecutor(@TempDir File tempDir) throws Exception {
rpcServiceExtension.getTestingRpcService(),
new TestingHighAvailabilityServicesBuilder().build(),
WorkingDirectory.create(tempDir))
+ .setConfiguration(configuration)
.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()));
+ 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..de1676c90b1909 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
@@ -21,6 +21,7 @@
import org.apache.flink.api.common.ApplicationID;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.java.tuple.Tuple7;
+import org.apache.flink.configuration.ThreadDumpMode;
import org.apache.flink.runtime.blob.TransientBlobKey;
import org.apache.flink.runtime.checkpoint.CheckpointOptions;
import org.apache.flink.runtime.clusterframework.types.AllocationID;
@@ -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..a27613679d422f 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
@@ -22,6 +22,7 @@
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.JobStatus;
import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.ThreadDumpMode;
import org.apache.flink.core.execution.CheckpointType;
import org.apache.flink.core.execution.SavepointFormatType;
import org.apache.flink.runtime.application.ArchivedApplication;
@@ -406,7 +407,8 @@ public CompletableFuture> requestMetricQueryServiceAddresses(
}
@Override
- public CompletableFuture requestThreadDump(Duration timeout) {
+ public CompletableFuture requestThreadDump(
+ ThreadDumpMode mode, Duration timeout) {
return null;
}