From 4efeefc9540216e416536e03b4cca0855e610d9e Mon Sep 17 00:00:00 2001
From: xingsuo-zbz <zhao_abc_123@163.com>
Date: Wed, 15 Jul 2026 11:00:42 +0800
Subject: [PATCH] [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 ++
 .../generated/rest_v1_dispatcher.html         | 20 +++++
 docs/static/generated/rest_v1_dispatcher.yml  | 22 ++++++
 .../flink/configuration/ClusterOptions.java   | 18 +++++
 .../flink/configuration/ThreadDumpMode.java   | 78 +++++++++++++++++++
 .../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 ++-
 .../resourcemanager/ResourceManager.java      |  5 +-
 .../ResourceManagerGateway.java               |  7 +-
 .../cluster/JobManagerThreadDumpHandler.java  | 17 +++-
 .../TaskManagerThreadDumpHandler.java         | 17 +++-
 .../runtime/rest/messages/ThreadDumpInfo.java |  7 +-
 .../ThreadDumpModeQueryParameter.java         | 56 +++++++++++++
 .../cluster/JobManagerThreadDumpHeaders.java  |  8 +-
 ...JobManagerThreadDumpMessageParameters.java | 44 +++++++++++
 .../TaskManagerThreadDumpHeaders.java         |  6 +-
 ...askManagerThreadDumpMessageParameters.java | 37 +++++++++
 .../runtime/taskexecutor/TaskExecutor.java    | 15 ++--
 .../taskexecutor/TaskExecutorGateway.java     |  5 +-
 .../TaskExecutorGatewayDecoratorBase.java     |  6 +-
 .../apache/flink/runtime/util/JvmUtils.java   | 12 +--
 .../NonLeaderRetrievalRestfulGateway.java     |  4 +-
 .../runtime/webmonitor/RestfulGateway.java    |  5 +-
 .../DispatcherThreadDumpOffloadTest.java      |  3 +-
 .../utils/TestingResourceManagerGateway.java  |  3 +-
 .../ThreadDumpModeQueryParameterTest.java     | 68 ++++++++++++++++
 .../TaskExecutorThreadDumpOffloadTest.java    | 69 ++++++++++++----
 .../TestingTaskExecutorGateway.java           |  4 +-
 .../webmonitor/TestingRestfulGateway.java     |  4 +-
 39 files changed, 786 insertions(+), 88 deletions(-)
 create mode 100644 flink-core/src/main/java/org/apache/flink/configuration/ThreadDumpMode.java
 create mode 100644 flink-core/src/test/java/org/apache/flink/configuration/ClusterOptionsTest.java
 create mode 100644 flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ThreadDumpModeQueryParameter.java
 create mode 100644 flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/cluster/JobManagerThreadDumpMessageParameters.java
 create mode 100644 flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/taskmanager/TaskManagerThreadDumpMessageParameters.java
 create mode 100644 flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/ThreadDumpModeQueryParameterTest.java

diff --git a/docs/layouts/shortcodes/generated/cluster_configuration.html b/docs/layouts/shortcodes/generated/cluster_configuration.html
index 1d1dbd87d9cf1d..04e605d53e3510 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 traces only, without lock information. Negligible JVM pause.</li><li>"FULL": 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.</li></ul></td>
+        </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 f32e45d44d0093..879d46790035a0 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 traces only, without lock information. Negligible JVM pause.</li><li>"FULL": 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.</li></ul></td>
+        </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 98c02d7a418fc5..26749b00f691b4 100644
--- a/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html
+++ b/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html
@@ -1383,6 +1383,16 @@
     <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 @@
         </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 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<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 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"
 ></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 d43b64680c6ee8..95d52b7ea35048 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 251d5cddf695f4..1b980be1914fe4 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 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)"
 ></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 d43b64680c6ee8..a71ff4989e782a 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 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<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 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<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 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<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 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<Collection<String>> requestMetricQueryServiceAddresses(
     }
 
     @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 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<Map<IntermediateDataSetID, DataSetMetaInfo>> listDataSe
 
     @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 CompletableFuture<ThreadDumpInfo> 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<Collection<LogInfo>> 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<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 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<? 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 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 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<? 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 @@ protected CompletableFuture<ThreadDumpInfo> 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<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 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<EmptyRequestBody, ThreadDumpInfo, EmptyMessageParameters> {
+        implements RuntimeMessageHeaders<
+                EmptyRequestBody, ThreadDumpInfo, JobManagerThreadDumpMessageParameters> {
 
     private static final JobManagerThreadDumpHeaders INSTANCE = new JobManagerThreadDumpHeaders();
 
@@ -43,8 +43,8 @@ public Class<EmptyRequestBody> 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<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 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<EmptyRequestBody> 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<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 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<Acknowledge> sendOperatorEventToTask(
     }
 
     @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 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<Acknowledge> 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<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 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<Acknowledge> sendOperatorEventToTask(
     }
 
     @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 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<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 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<Collection<String>> requestMetricQueryServiceAddresses(
     }
 
     @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 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<Collection<String>> 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<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 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<Collection<LogInfo>> requestTaskManagerLogList(
 
     @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 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<Acknowledge> sendOperatorEventToTask(
     }
 
     @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 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<Collection<String>> requestMetricQueryServiceAddresses(
     }
 
     @Override
-    public CompletableFuture<ThreadDumpInfo> requestThreadDump(Duration timeout) {
+    public CompletableFuture<ThreadDumpInfo> requestThreadDump(
+            ThreadDumpMode mode, Duration timeout) {
         return null;
     }