This is an automated email from the ASF dual-hosted git repository.
Myasuka pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 7232ad9c2c4 [FLINK-39984][runtime][webUI] Support LITE/FULL thread
dump modes
7232ad9c2c4 is described below
commit 7232ad9c2c4346f92d820d69ef3f192bebb58045
Author: xingsuo-zbz <[email protected]>
AuthorDate: Wed Jul 15 11:00:42 2026 +0800
[FLINK-39984][runtime][webUI] Support LITE/FULL thread dump modes
ThreadMXBean.dumpAllThreads(true, true) enters a single JVM-wide
safepoint to collect monitor/synchronizer state; on busy JVMs the pause
can exceed heartbeat.timeout and cause unnecessary TaskManager failover.
- Introduce ThreadDumpMode {LITE, FULL}: LITE = dumpAllThreads(false,
false), FULL preserves today's (true, true) behavior. Exposed via an
optional query parameter `?mode=lite|full` on the JM/TM thread-dump
endpoints.
- Add config cluster.thread-dump.default-mode (default FULL to preserve
upgrade behavior; LITE recommended for large clusters).
- Add a Lite/Full toggle to both Web UI thread-dump pages; selecting a
mode does not auto-fetch, the download link tracks the selection.
---
.../generated/cluster_configuration.html | 6 ++
.../generated/expert_cluster_section.html | 6 ++
.../shortcodes/generated/rest_v1_dispatcher.html | 20 ++++++
docs/static/generated/rest_v1_dispatcher.yml | 22 ++++++
.../apache/flink/configuration/ClusterOptions.java | 18 +++++
.../apache/flink/configuration/ThreadDumpMode.java | 78 ++++++++++++++++++++++
.../flink/configuration/ClusterOptionsTest.java | 64 ++++++++++++++++++
.../src/test/resources/rest_api_v1.snapshot | 10 ++-
.../job-manager-thread-dump.component.html | 36 ++++++++--
.../job-manager-thread-dump.component.less | 22 ++++++
.../job-manager-thread-dump.component.ts | 40 ++++++++++-
.../task-manager-thread-dump.component.html | 36 ++++++++--
.../task-manager-thread-dump.component.less | 25 +++++++
.../task-manager-thread-dump.component.ts | 52 ++++++++++++++-
.../src/app/services/job-manager.service.ts | 8 ++-
.../src/app/services/task-manager.service.ts | 18 ++---
.../flink/runtime/dispatcher/Dispatcher.java | 11 ++-
.../runtime/resourcemanager/ResourceManager.java | 5 +-
.../resourcemanager/ResourceManagerGateway.java | 7 +-
.../cluster/JobManagerThreadDumpHandler.java | 17 +++--
.../taskmanager/TaskManagerThreadDumpHandler.java | 17 +++--
.../runtime/rest/messages/ThreadDumpInfo.java | 7 +-
.../messages/ThreadDumpModeQueryParameter.java | 56 ++++++++++++++++
.../cluster/JobManagerThreadDumpHeaders.java | 8 +--
.../JobManagerThreadDumpMessageParameters.java | 44 ++++++++++++
.../taskmanager/TaskManagerThreadDumpHeaders.java | 6 +-
.../TaskManagerThreadDumpMessageParameters.java | 26 +++++---
.../flink/runtime/taskexecutor/TaskExecutor.java | 15 +++--
.../runtime/taskexecutor/TaskExecutorGateway.java | 5 +-
.../TaskExecutorGatewayDecoratorBase.java | 6 +-
.../org/apache/flink/runtime/util/JvmUtils.java | 12 ++--
.../NonLeaderRetrievalRestfulGateway.java | 4 +-
.../flink/runtime/webmonitor/RestfulGateway.java | 5 +-
.../DispatcherThreadDumpOffloadTest.java | 3 +-
.../utils/TestingResourceManagerGateway.java | 3 +-
.../messages/ThreadDumpModeQueryParameterTest.java | 68 +++++++++++++++++++
.../TaskExecutorThreadDumpOffloadTest.java | 69 ++++++++++++++-----
.../taskexecutor/TestingTaskExecutorGateway.java | 4 +-
.../runtime/webmonitor/TestingRestfulGateway.java | 4 +-
39 files changed, 767 insertions(+), 96 deletions(-)
diff --git a/docs/layouts/shortcodes/generated/cluster_configuration.html
b/docs/layouts/shortcodes/generated/cluster_configuration.html
index 1d1dbd87d9c..04e605d53e3 100644
--- a/docs/layouts/shortcodes/generated/cluster_configuration.html
+++ b/docs/layouts/shortcodes/generated/cluster_configuration.html
@@ -62,6 +62,12 @@
<td>Duration</td>
<td>The shutdown timeout for cluster services like executors.</td>
</tr>
+ <tr>
+ <td><h5>cluster.thread-dump.default-mode</h5></td>
+ <td style="word-wrap: break-word;">FULL</td>
+ <td><p>Enum</p></td>
+ <td>Default granularity of the JobManager/TaskManager thread-dump
REST endpoint when no explicit <code class="highlighter-rouge">mode</code>
query parameter is supplied. The default is <code
class="highlighter-rouge">FULL</code> to preserve historical behavior;
operators of large clusters are strongly encouraged to switch to <code
class="highlighter-rouge">LITE</code> to avoid heartbeat timeouts caused by
long safepoint pauses.<br /><br />Possible values:<ul><li>"LITE": Stack [...]
+ </tr>
<tr>
<td><h5>cluster.thread-dump.stacktrace-max-depth</h5></td>
<td style="word-wrap: break-word;">50</td>
diff --git a/docs/layouts/shortcodes/generated/expert_cluster_section.html
b/docs/layouts/shortcodes/generated/expert_cluster_section.html
index f32e45d44d0..879d4679003 100644
--- a/docs/layouts/shortcodes/generated/expert_cluster_section.html
+++ b/docs/layouts/shortcodes/generated/expert_cluster_section.html
@@ -26,6 +26,12 @@
<td>Boolean</td>
<td>Whether processes should halt on fatal errors instead of
performing a graceful shutdown. In some environments (e.g. Java 8 with the G1
garbage collector), a regular graceful shutdown can lead to a JVM deadlock. See
<a href="https://issues.apache.org/jira/browse/FLINK-16510";>FLINK-16510</a> for
details.</td>
</tr>
+ <tr>
+ <td><h5>cluster.thread-dump.default-mode</h5></td>
+ <td style="word-wrap: break-word;">FULL</td>
+ <td><p>Enum</p></td>
+ <td>Default granularity of the JobManager/TaskManager thread-dump
REST endpoint when no explicit <code class="highlighter-rouge">mode</code>
query parameter is supplied. The default is <code
class="highlighter-rouge">FULL</code> to preserve historical behavior;
operators of large clusters are strongly encouraged to switch to <code
class="highlighter-rouge">LITE</code> to avoid heartbeat timeouts caused by
long safepoint pauses.<br /><br />Possible values:<ul><li>"LITE": Stack [...]
+ </tr>
<tr>
<td><h5>cluster.thread-dump.stacktrace-max-depth</h5></td>
<td style="word-wrap: break-word;">50</td>
diff --git a/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html
b/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html
index 98c02d7a418..26749b00f69 100644
--- a/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html
+++ b/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html
@@ -1383,6 +1383,16 @@ Using 'curl' you can upload a jar via 'curl -X POST -H
"Expect:" -F "jarfile=@pa
<tr>
<td colspan="2">Returns the thread dump of the JobManager.</td>
</tr>
+ <tr>
+ <td colspan="2">Query parameters</td>
+ </tr>
+ <tr>
+ <td colspan="2">
+ <ul>
+<li><code>mode</code> (optional): Controls how much lock information is
collected. Supported values: [LITE, FULL]. When omitted,
cluster.thread-dump.default-mode is used.</li>
+ </ul>
+ </td>
+ </tr>
<tr>
<td colspan="2">
<label>
@@ -7121,6 +7131,16 @@ Using 'curl' you can upload a jar via 'curl -X POST -H
"Expect:" -F "jarfile=@pa
</ul>
</td>
</tr>
+ <tr>
+ <td colspan="2">Query parameters</td>
+ </tr>
+ <tr>
+ <td colspan="2">
+ <ul>
+<li><code>mode</code> (optional): Controls how much lock information is
collected. Supported values: [LITE, FULL]. When omitted,
cluster.thread-dump.default-mode is used.</li>
+ </ul>
+ </td>
+ </tr>
<tr>
<td colspan="2">
<label>
diff --git a/docs/static/generated/rest_v1_dispatcher.yml
b/docs/static/generated/rest_v1_dispatcher.yml
index 24ecb76ed2e..0c94fafc052 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 4957f486a08..6d03db19229 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<ThreadDumpMode> 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<UncaughtExceptionHandleMode>
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 00000000000..caa58747c14
--- /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 00000000000..40e984b4b38
--- /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 890ac91ef3c..a1dd691a7d2 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 f7572891985..a7fbfc49070 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"
></nz-code-editor>
-<flink-addon-compact
- [downloadHref]="downloadUrl"
- [downloadName]="downloadName"
- (reload)="reload()"
-></flink-addon-compact>
+<div class="thread-dump-toolbar">
+ <nz-space-compact class="thread-dump-mode-group">
+ <a
+ nz-button
+ nzSize="small"
+ [nzType]="mode === 'lite' ? 'primary' : 'default'"
+ nz-tooltip
+ nzTooltipTitle="Stack traces only, no lock info (jstack without -l).
Negligible JVM pause."
+ (click)="selectMode('lite')"
+ >
+ Lite
+ </a>
+ <a
+ nz-button
+ nzSize="small"
+ [nzType]="mode === 'full' ? 'primary' : 'default'"
+ [nzDanger]="true"
+ nz-tooltip
+ nzTooltipTitle="Adds locked monitors and j.u.c. synchronizers (jstack
-l). Pauses the JVM in a safepoint for a duration that scales with heap size
and thread count — seconds on large JMs."
+ (click)="selectMode('full')"
+ >
+ Full
+ <i nz-icon nzType="exclamation-circle" nzTheme="fill"
class="thread-dump-mode-warn"></i>
+ </a>
+ </nz-space-compact>
+ <flink-addon-compact
+ [downloadHref]="downloadUrl"
+ [downloadName]="downloadName"
+ (reload)="reload()"
+ ></flink-addon-compact>
+</div>
diff --git
a/flink-runtime-web/web-dashboard/src/app/pages/job-manager/thread-dump/job-manager-thread-dump.component.less
b/flink-runtime-web/web-dashboard/src/app/pages/job-manager/thread-dump/job-manager-thread-dump.component.less
index d43b64680c6..95d52b7ea35 100644
---
a/flink-runtime-web/web-dashboard/src/app/pages/job-manager/thread-dump/job-manager-thread-dump.component.less
+++
b/flink-runtime-web/web-dashboard/src/app/pages/job-manager/thread-dump/job-manager-thread-dump.component.less
@@ -25,3 +25,25 @@
inset: 0;
}
}
+
+.thread-dump-toolbar {
+ position: absolute;
+ top: 8px;
+ right: 32px;
+ z-index: 1;
+ display: flex;
+ gap: 8px;
+ align-items: center;
+
+ 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/job-manager/thread-dump/job-manager-thread-dump.component.ts
b/flink-runtime-web/web-dashboard/src/app/pages/job-manager/thread-dump/job-manager-thread-dump.component.ts
index 251d5cddf69..1b980be1914 100644
---
a/flink-runtime-web/web-dashboard/src/app/pages/job-manager/thread-dump/job-manager-thread-dump.component.ts
+++
b/flink-runtime-web/web-dashboard/src/app/pages/job-manager/thread-dump/job-manager-thread-dump.component.ts
@@ -29,14 +29,30 @@ import {
JobManagerModuleConfig
} from '@flink-runtime-web/pages/job-manager/job-manager.config';
import { ConfigService, JobManagerService } from '@flink-runtime-web/services';
+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';
+
+/** See ThreadDumpMode in task-manager-thread-dump.component.ts for semantics.
*/
+type ThreadDumpMode = 'lite' | 'full' | undefined;
@Component({
selector: 'flink-job-manager-thread-dump',
templateUrl: './job-manager-thread-dump.component.html',
styleUrls: ['./job-manager-thread-dump.component.less'],
changeDetection: ChangeDetectionStrategy.OnPush,
- imports: [NzCodeEditorModule, AutoResizeDirective, FormsModule,
AddonCompactComponent]
+ imports: [
+ NzCodeEditorModule,
+ AutoResizeDirective,
+ FormsModule,
+ AddonCompactComponent,
+ NzButtonModule,
+ NzIconModule,
+ NzSpaceModule,
+ NzTooltipModule
+ ]
})
export class JobManagerThreadDumpComponent implements OnInit, OnDestroy {
public readonly downloadName = `jobmanager_thread_dump`;
@@ -44,6 +60,8 @@ export class JobManagerThreadDumpComponent implements OnInit,
OnDestroy {
public editorOptions: EditorOptions;
public dump = '';
public loading = true;
+ /** See TaskManagerThreadDumpComponent#mode. */
+ public mode: ThreadDumpMode = undefined;
private readonly destroy$ = new Subject<void>();
@@ -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 30941b13709..d0cc1bb6df3 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)"
></nz-code-editor>
-<flink-addon-compact
- [downloadHref]="downloadUrl"
- [downloadName]="downloadName"
- (reload)="reload()"
-></flink-addon-compact>
+<div class="thread-dump-toolbar">
+ <nz-space-compact class="thread-dump-mode-group">
+ <a
+ nz-button
+ nzSize="small"
+ [nzType]="mode === 'lite' ? 'primary' : 'default'"
+ nz-tooltip
+ nzTooltipTitle="Stack traces only, no lock info (jstack without -l).
Negligible JVM pause."
+ (click)="selectMode('lite')"
+ >
+ Lite
+ </a>
+ <a
+ nz-button
+ nzSize="small"
+ [nzType]="mode === 'full' ? 'primary' : 'default'"
+ [nzDanger]="true"
+ nz-tooltip
+ nzTooltipTitle="Adds locked monitors and j.u.c. synchronizers (jstack
-l). Pauses the JVM in a safepoint for a duration that scales with heap size
and thread count — seconds on large TMs."
+ (click)="selectMode('full')"
+ >
+ Full
+ <i nz-icon nzType="exclamation-circle" nzTheme="fill"
class="thread-dump-mode-warn"></i>
+ </a>
+ </nz-space-compact>
+ <flink-addon-compact
+ [downloadHref]="downloadUrl"
+ [downloadName]="downloadName"
+ (reload)="reload()"
+ ></flink-addon-compact>
+</div>
diff --git
a/flink-runtime-web/web-dashboard/src/app/pages/task-manager/thread-dump/task-manager-thread-dump.component.less
b/flink-runtime-web/web-dashboard/src/app/pages/task-manager/thread-dump/task-manager-thread-dump.component.less
index d43b64680c6..a71ff4989e7 100644
---
a/flink-runtime-web/web-dashboard/src/app/pages/task-manager/thread-dump/task-manager-thread-dump.component.less
+++
b/flink-runtime-web/web-dashboard/src/app/pages/task-manager/thread-dump/task-manager-thread-dump.component.less
@@ -25,3 +25,28 @@
inset: 0;
}
}
+
+.thread-dump-toolbar {
+ position: absolute;
+ top: 8px;
+ right: 32px;
+ z-index: 1;
+ display: flex;
+ gap: 8px;
+ align-items: center;
+
+ // Inner <flink-addon-compact> 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 961a84493da..2474f66928f 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<void>();
@@ -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 52064a87081..c70ae01971b 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<string> {
- return
this.httpClient.get<JobManagerThreadDump>(`${this.configService.BASE_URL}/jobmanager/thread-dump`).pipe(
+ loadThreadDump(mode?: 'lite' | 'full'): Observable<string> {
+ let url = `${this.configService.BASE_URL}/jobmanager/thread-dump`;
+ if (mode) {
+ url += `?mode=${mode}`;
+ }
+ return this.httpClient.get<JobManagerThreadDump>(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 b5ece679d03..3d3ff96172c 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<string> {
- return this.httpClient
-
.get<TaskManagerThreadDump>(`${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<string> {
+ let url =
`${this.configService.BASE_URL}/taskmanagers/${taskManagerId}/thread-dump`;
+ if (mode) {
+ url += `?mode=${mode}`;
+ }
+ return this.httpClient.get<TaskManagerThreadDump>(url).pipe(
+ map(taskManagerThreadDump => {
+ return taskManagerThreadDump.threadInfos.map(threadInfo =>
threadInfo.stringifiedThreadInfo).join('');
+ })
+ );
}
loadLogs(taskManagerId: string): Observable<string> {
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 50bb43285e0..f64645b196b 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.Configuration;
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 abstract class Dispatcher extends
FencedRpcEndpoint<DispatcherId>
}
@Override
- public CompletableFuture<ThreadDumpInfo> requestThreadDump(Duration
timeout) {
- int stackTraceMaxDepth =
configuration.get(ClusterOptions.THREAD_DUMP_STACKTRACE_MAX_DEPTH);
+ public CompletableFuture<ThreadDumpInfo> 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 13af32c8884..5cfb06b3bef 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.annotation.VisibleForTesting;
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 abstract class ResourceManager<WorkerType extends
ResourceIDRetrievable>
@Override
public CompletableFuture<ThreadDumpInfo> requestThreadDump(
- ResourceID taskManagerId, Duration timeout) {
+ ResourceID taskManagerId, ThreadDumpMode mode, Duration timeout) {
final WorkerRegistration<WorkerType> taskExecutor =
taskExecutors.get(taskManagerId);
if (taskExecutor == null) {
@@ -906,7 +907,7 @@ public abstract class ResourceManager<WorkerType extends
ResourceIDRetrievable>
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 938f3c58621..9c722604ebc 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 @@ package org.apache.flink.runtime.resourcemanager;
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 @@ public interface ResourceManagerGateway
/**
* 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<ThreadDumpInfo> 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 15dcbf322d8..368a0a63026 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 @@ import java.util.concurrent.CompletableFuture;
/** 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<? extends RestfulGateway> leaderRetriever,
Duration timeout,
Map<String, String> responseHeaders,
- MessageHeaders<EmptyRequestBody, ThreadDumpInfo,
EmptyMessageParameters>
+ MessageHeaders<EmptyRequestBody, ThreadDumpInfo,
JobManagerThreadDumpMessageParameters>
messageHeaders) {
super(leaderRetriever, timeout, responseHeaders, messageHeaders);
}
@@ -52,6 +58,9 @@ public class JobManagerThreadDumpHandler
protected CompletableFuture<ThreadDumpInfo> handleRequest(
@Nonnull HandlerRequest<EmptyRequestBody> 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 d0dc803262e..cc7472c06f9 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 @@ import java.util.concurrent.CompletableFuture;
/** 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<? extends RestfulGateway> leaderRetriever,
Duration timeout,
Map<String, String> responseHeaders,
- MessageHeaders<EmptyRequestBody, ThreadDumpInfo,
TaskManagerMessageParameters>
+ MessageHeaders<EmptyRequestBody, ThreadDumpInfo,
TaskManagerThreadDumpMessageParameters>
messageHeaders,
GatewayRetriever<ResourceManagerGateway>
resourceManagerGatewayRetriever) {
super(
@@ -64,6 +70,9 @@ public class TaskManagerThreadDumpHandler
@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 e752ac29707..25ddd90f249 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 final class ThreadDumpInfo implements ResponseBody,
Serializable {
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 00000000000..548bc6a70d5
--- /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<ThreadDumpMode> {
+
+ 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 f0dfba9439a..b021d8f5889 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 @@ package org.apache.flink.runtime.rest.messages.cluster;
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 @@ import
org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseSt
/** Headers for the {@link JobManagerThreadDumpHandler}. */
public class JobManagerThreadDumpHeaders
- implements RuntimeMessageHeaders<EmptyRequestBody, ThreadDumpInfo,
EmptyMessageParameters> {
+ implements RuntimeMessageHeaders<
+ EmptyRequestBody, ThreadDumpInfo,
JobManagerThreadDumpMessageParameters> {
private static final JobManagerThreadDumpHeaders INSTANCE = new
JobManagerThreadDumpHeaders();
@@ -43,8 +43,8 @@ public class JobManagerThreadDumpHeaders
}
@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 00000000000..84459686691
--- /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<MessagePathParameter<?>> getPathParameters() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public Collection<MessageQueryParameter<?>> 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 8695abb56fe..22318cde5af 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 @@ import
org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseSt
/** 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 TaskManagerThreadDumpHeaders
}
@Override
- public TaskManagerMessageParameters getUnresolvedMessageParameters() {
- return new TaskManagerMessageParameters();
+ public TaskManagerThreadDumpMessageParameters
getUnresolvedMessageParameters() {
+ return new TaskManagerThreadDumpMessageParameters();
}
@Override
diff --git
a/flink-runtime-web/web-dashboard/src/app/pages/job-manager/thread-dump/job-manager-thread-dump.component.less
b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/taskmanager/TaskManagerThreadDumpMessageParameters.java
similarity index 50%
copy from
flink-runtime-web/web-dashboard/src/app/pages/job-manager/thread-dump/job-manager-thread-dump.component.less
copy to
flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/taskmanager/TaskManagerThreadDumpMessageParameters.java
index d43b64680c6..1079212835a 100644
---
a/flink-runtime-web/web-dashboard/src/app/pages/job-manager/thread-dump/job-manager-thread-dump.component.less
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/taskmanager/TaskManagerThreadDumpMessageParameters.java
@@ -7,7 +7,7 @@
* "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
+ * 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,
@@ -16,12 +16,22 @@
* limitations under the License.
*/
-:host {
- position: relative;
- flex: 1;
+package org.apache.flink.runtime.rest.messages.taskmanager;
- nz-code-editor {
- position: absolute;
- inset: 0;
- }
+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<MessageQueryParameter<?>> 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 a7558bf6d3f..b8609c937bb 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.ApplicationID;
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 class TaskExecutor extends RpcEndpoint
implements TaskExecutorGateway {
}
@Override
- public CompletableFuture<ThreadDumpInfo> requestThreadDump(Duration
timeout) {
- int stacktraceMaxDepth =
- taskManagerConfiguration
- .getConfiguration()
- .get(ClusterOptions.THREAD_DUMP_STACKTRACE_MAX_DEPTH);
+ public CompletableFuture<ThreadDumpInfo> 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 29607c2c2f8..cb13459a47a 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 @@ package org.apache.flink.runtime.taskexecutor;
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 @@ public interface TaskExecutorGateway
/**
* 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<ThreadDumpInfo> requestThreadDump(@RpcTimeout Duration
timeout);
+ CompletableFuture<ThreadDumpInfo> 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 5d33ddd65a0..cb46965f9c9 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 @@ package org.apache.flink.runtime.taskexecutor;
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 class TaskExecutorGatewayDecoratorBase implements
TaskExecutorGateway {
}
@Override
- public CompletableFuture<ThreadDumpInfo> requestThreadDump(Duration
timeout) {
- return originalGateway.requestThreadDump(timeout);
+ public CompletableFuture<ThreadDumpInfo> 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 89a8f8fe0cc..6781b7f02a6 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<ThreadInfo> createThreadDump() {
+ public static Collection<ThreadInfo> 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 e162a2823f9..2455128e50a 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 @@ package org.apache.flink.runtime.webmonitor;
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 class NonLeaderRetrievalRestfulGateway implements
RestfulGateway {
}
@Override
- public CompletableFuture<ThreadDumpInfo> requestThreadDump(Duration
timeout) {
+ public CompletableFuture<ThreadDumpInfo> 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 62cd830a653..d02a3d12c05 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.ApplicationID;
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 @@ public interface RestfulGateway extends RpcGateway {
/**
* 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<ThreadDumpInfo> requestThreadDump(@RpcTimeout Duration
timeout);
+ CompletableFuture<ThreadDumpInfo> 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 111e52a1414..7497a3bba18 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 class DispatcherThreadDumpOffloadTest extends
AbstractDispatcherTest {
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 603eee96625..bd38d99eb7f 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.JobID;
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 class TestingResourceManagerGateway implements
ResourceManagerGateway {
@Override
public CompletableFuture<ThreadDumpInfo> requestThreadDump(
- ResourceID taskManagerId, Duration timeout) {
+ ResourceID taskManagerId, ThreadDumpMode mode, Duration timeout) {
final Function<ResourceID, CompletableFuture<ThreadDumpInfo>> 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 00000000000..4c7ca442e9e
--- /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 5257890563c..9af73c5c1e8 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.Test;
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 java.util.concurrent.TimeUnit;
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 @@ class TaskExecutorThreadDumpOffloadTest {
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 cf1b79c68e6..de1676c90b1 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 @@ package org.apache.flink.runtime.taskexecutor;
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 class TestingTaskExecutorGateway implements
TaskExecutorGateway {
}
@Override
- public CompletableFuture<ThreadDumpInfo> requestThreadDump(Duration
timeout) {
+ public CompletableFuture<ThreadDumpInfo> 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 ac1b46b1e07..a27613679d4 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.ApplicationID;
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 class TestingRestfulGateway implements
RestfulGateway {
}
@Override
- public CompletableFuture<ThreadDumpInfo> requestThreadDump(Duration
timeout) {
+ public CompletableFuture<ThreadDumpInfo> requestThreadDump(
+ ThreadDumpMode mode, Duration timeout) {
return null;
}
