From 86388ac0cadaffdcf05bc08da4c48917fc38bbe3 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.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
---
 .../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   |  25 +++++
 .../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  |  13 ++-
 .../resourcemanager/ResourceManager.java      |   5 +-
 .../ResourceManagerGateway.java               |   7 +-
 .../cluster/JobManagerThreadDumpHandler.java  |  17 ++-
 .../TaskManagerThreadDumpHandler.java         |  17 ++-
 .../runtime/rest/messages/ThreadDumpInfo.java |   6 +-
 .../runtime/rest/messages/ThreadDumpMode.java |  83 ++++++++++++++
 .../ThreadDumpModeQueryParameter.java         |  53 +++++++++
 .../cluster/JobManagerThreadDumpHeaders.java  |   8 +-
 ...JobManagerThreadDumpMessageParameters.java |  44 ++++++++
 .../TaskManagerThreadDumpHeaders.java         |   6 +-
 ...askManagerThreadDumpMessageParameters.java |  37 ++++++
 .../runtime/taskexecutor/TaskExecutor.java    |  16 ++-
 .../taskexecutor/TaskExecutorGateway.java     |   5 +-
 .../TaskExecutorGatewayDecoratorBase.java     |   6 +-
 .../apache/flink/runtime/util/JvmUtils.java   |  12 +-
 .../NonLeaderRetrievalRestfulGateway.java     |   4 +-
 .../runtime/webmonitor/RestfulGateway.java    |   5 +-
 .../utils/TestingResourceManagerGateway.java  |   3 +-
 .../ThreadDumpModeQueryParameterTest.java     |  66 +++++++++++
 .../TaskExecutorThreadDumpTest.java           | 105 ++++++++++++++++++
 .../TestingTaskExecutorGateway.java           |   4 +-
 .../webmonitor/TestingRestfulGateway.java     |   4 +-
 37 files changed, 780 insertions(+), 72 deletions(-)
 create mode 100644 flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ThreadDumpMode.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
 create mode 100644 flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorThreadDumpTest.java

diff --git a/docs/layouts/shortcodes/generated/cluster_configuration.html b/docs/layouts/shortcodes/generated/cluster_configuration.html
index 1d1dbd87d9cf1d..ada54985eadf2f 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>String</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. Allowed values (case-insensitive): <code class="highlighter-rouge">LITE</code> (stack traces only, negligible JVM pause) and <code class="highlighter-rouge">FULL</code> (adds locked monitors and j.u.c. synchronizers, equivalent to jstack -l; pauses the JVM in a safepoint for a duration that scales with heap size and thread count -- seconds on large TaskManagers). 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.</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..82b6b9364a403c 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>String</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. Allowed values (case-insensitive): <code class="highlighter-rouge">LITE</code> (stack traces only, negligible JVM pause) and <code class="highlighter-rouge">FULL</code> (adds locked monitors and j.u.c. synchronizers, equivalent to jstack -l; pauses the JVM in a safepoint for a duration that scales with heap size and thread count -- seconds on large TaskManagers). 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.</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..6566cfd4b6d4e4 100644
--- a/flink-core/src/main/java/org/apache/flink/configuration/ClusterOptions.java
+++ b/flink-core/src/main/java/org/apache/flink/configuration/ClusterOptions.java
@@ -140,6 +140,31 @@ public class ClusterOptions {
                     .withDescription(
                             "The maximum stacktrace depth of TaskManager and JobManager's thread dump web-frontend displayed.");
 
+    @Documentation.Section(Documentation.Sections.EXPERT_CLUSTER)
+    public static final ConfigOption<String> THREAD_DUMP_DEFAULT_MODE =
+            key("cluster.thread-dump.default-mode")
+                    .stringType()
+                    .defaultValue("FULL")
+                    .withDescription(
+                            Description.builder()
+                                    .text(
+                                            "Default granularity of the JobManager/TaskManager thread-dump REST endpoint "
+                                                    + "when no explicit %s query parameter is supplied. ",
+                                            code("mode"))
+                                    .text(
+                                            "Allowed values (case-insensitive): %s (stack traces only, negligible JVM pause) "
+                                                    + "and %s (adds locked monitors and j.u.c. synchronizers, equivalent "
+                                                    + "to jstack -l; pauses the JVM in a safepoint for a duration that "
+                                                    + "scales with heap size and thread count -- seconds on large "
+                                                    + "TaskManagers). ",
+                                            code("LITE"), code("FULL"))
+                                    .text(
+                                            "The default is %s to preserve historical behavior; operators of large "
+                                                    + "clusters are strongly encouraged to switch to %s to avoid "
+                                                    + "heartbeat timeouts caused by long safepoint pauses.",
+                                            code("FULL"), code("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-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 fb011b9457dca4..450c179b3254b9 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
@@ -28,3 +28,25 @@
     left: 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 fb011b9457dca4..313cc726de2646 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
@@ -28,3 +28,28 @@
     left: 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 7a3525b6eb0872..950186a071b6ae 100644
--- a/flink-runtime-web/web-dashboard/src/app/services/job-manager.service.ts
+++ b/flink-runtime-web/web-dashboard/src/app/services/job-manager.service.ts
@@ -82,8 +82,12 @@ export class JobManagerService {
       );
   }
 
-  loadThreadDump(): Observable<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 3a8c3227ddcf23..26a50b06e1a695 100644
--- a/flink-runtime-web/web-dashboard/src/app/services/task-manager.service.ts
+++ b/flink-runtime-web/web-dashboard/src/app/services/task-manager.service.ts
@@ -74,14 +74,16 @@ export class TaskManagerService {
       );
   }
 
-  loadThreadDump(taskManagerId: string): Observable<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 6d53fa0e2216be..9a86a2df45a584 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java
@@ -107,6 +107,7 @@
 import org.apache.flink.runtime.rest.handler.async.OperationResult;
 import org.apache.flink.runtime.rest.handler.job.AsynchronousJobOperationKey;
 import org.apache.flink.runtime.rest.messages.ThreadDumpInfo;
+import org.apache.flink.runtime.rest.messages.ThreadDumpMode;
 import org.apache.flink.runtime.rpc.FatalErrorHandler;
 import org.apache.flink.runtime.rpc.FencedRpcEndpoint;
 import org.apache.flink.runtime.rpc.RpcService;
@@ -1855,9 +1856,15 @@ public CompletableFuture<Collection<String>> requestMetricQueryServiceAddresses(
     }
 
     @Override
-    public CompletableFuture<ThreadDumpInfo> requestThreadDump(Duration timeout) {
-        int stackTraceMaxDepth = configuration.get(ClusterOptions.THREAD_DUMP_STACKTRACE_MAX_DEPTH);
-        return CompletableFuture.completedFuture(ThreadDumpInfo.dumpAndCreate(stackTraceMaxDepth));
+    public CompletableFuture<ThreadDumpInfo> requestThreadDump(
+            ThreadDumpMode mode, Duration timeout) {
+        final int stackTraceMaxDepth =
+                configuration.get(ClusterOptions.THREAD_DUMP_STACKTRACE_MAX_DEPTH);
+        final ThreadDumpMode resolvedMode =
+                ThreadDumpMode.resolve(
+                        mode, configuration.get(ClusterOptions.THREAD_DUMP_DEFAULT_MODE));
+        return CompletableFuture.completedFuture(
+                ThreadDumpInfo.dumpAndCreate(stackTraceMaxDepth, resolvedMode));
     }
 
     @Override
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java
index 13af32c8884422..d1f2c2a0a03d8f 100755
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java
@@ -62,6 +62,7 @@
 import org.apache.flink.runtime.rest.messages.LogInfo;
 import org.apache.flink.runtime.rest.messages.ProfilingInfo;
 import org.apache.flink.runtime.rest.messages.ThreadDumpInfo;
+import org.apache.flink.runtime.rest.messages.ThreadDumpMode;
 import org.apache.flink.runtime.rest.messages.taskmanager.TaskManagerInfo;
 import org.apache.flink.runtime.rpc.FatalErrorHandler;
 import org.apache.flink.runtime.rpc.FencedRpcEndpoint;
@@ -896,7 +897,7 @@ public CompletableFuture<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..ae147c9393c394 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerGateway.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerGateway.java
@@ -39,6 +39,7 @@
 import org.apache.flink.runtime.rest.messages.ProfilingInfo;
 import org.apache.flink.runtime.rest.messages.ProfilingInfo.ProfilingMode;
 import org.apache.flink.runtime.rest.messages.ThreadDumpInfo;
+import org.apache.flink.runtime.rest.messages.ThreadDumpMode;
 import org.apache.flink.runtime.rest.messages.taskmanager.TaskManagerInfo;
 import org.apache.flink.runtime.rpc.FencedRpcGateway;
 import org.apache.flink.runtime.rpc.RpcTimeout;
@@ -257,13 +258,13 @@ CompletableFuture<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..d4e043a9dfef94 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/cluster/JobManagerThreadDumpHandler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/cluster/JobManagerThreadDumpHandler.java
@@ -21,10 +21,13 @@
 import org.apache.flink.runtime.rest.handler.AbstractRestHandler;
 import org.apache.flink.runtime.rest.handler.HandlerRequest;
 import org.apache.flink.runtime.rest.handler.RestHandlerException;
-import org.apache.flink.runtime.rest.messages.EmptyMessageParameters;
+import org.apache.flink.runtime.rest.handler.util.HandlerRequestUtils;
 import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
 import org.apache.flink.runtime.rest.messages.MessageHeaders;
 import org.apache.flink.runtime.rest.messages.ThreadDumpInfo;
+import org.apache.flink.runtime.rest.messages.ThreadDumpMode;
+import org.apache.flink.runtime.rest.messages.ThreadDumpModeQueryParameter;
+import org.apache.flink.runtime.rest.messages.cluster.JobManagerThreadDumpMessageParameters;
 import org.apache.flink.runtime.webmonitor.RestfulGateway;
 import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever;
 
@@ -37,13 +40,16 @@
 /** Rest handler which serves the thread dump info from the JobManager. */
 public class JobManagerThreadDumpHandler
         extends AbstractRestHandler<
-                RestfulGateway, EmptyRequestBody, ThreadDumpInfo, EmptyMessageParameters> {
+                RestfulGateway,
+                EmptyRequestBody,
+                ThreadDumpInfo,
+                JobManagerThreadDumpMessageParameters> {
 
     public JobManagerThreadDumpHandler(
             GatewayRetriever<? 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..109c9511ec5fc9 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/taskmanager/TaskManagerThreadDumpHandler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/taskmanager/TaskManagerThreadDumpHandler.java
@@ -23,11 +23,14 @@
 import org.apache.flink.runtime.rest.handler.HandlerRequest;
 import org.apache.flink.runtime.rest.handler.RestHandlerException;
 import org.apache.flink.runtime.rest.handler.resourcemanager.AbstractResourceManagerHandler;
+import org.apache.flink.runtime.rest.handler.util.HandlerRequestUtils;
 import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
 import org.apache.flink.runtime.rest.messages.MessageHeaders;
 import org.apache.flink.runtime.rest.messages.ThreadDumpInfo;
+import org.apache.flink.runtime.rest.messages.ThreadDumpMode;
+import org.apache.flink.runtime.rest.messages.ThreadDumpModeQueryParameter;
 import org.apache.flink.runtime.rest.messages.taskmanager.TaskManagerIdPathParameter;
-import org.apache.flink.runtime.rest.messages.taskmanager.TaskManagerMessageParameters;
+import org.apache.flink.runtime.rest.messages.taskmanager.TaskManagerThreadDumpMessageParameters;
 import org.apache.flink.runtime.taskexecutor.TaskExecutor;
 import org.apache.flink.runtime.webmonitor.RestfulGateway;
 import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever;
@@ -41,13 +44,16 @@
 /** Rest handler which serves the thread dump info from a {@link TaskExecutor}. */
 public class TaskManagerThreadDumpHandler
         extends AbstractResourceManagerHandler<
-                RestfulGateway, EmptyRequestBody, ThreadDumpInfo, TaskManagerMessageParameters> {
+                RestfulGateway,
+                EmptyRequestBody,
+                ThreadDumpInfo,
+                TaskManagerThreadDumpMessageParameters> {
 
     public TaskManagerThreadDumpHandler(
             GatewayRetriever<? 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..3409deadc4edd2 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ThreadDumpInfo.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ThreadDumpInfo.java
@@ -54,9 +54,11 @@ public static ThreadDumpInfo create(
         return new ThreadDumpInfo(threadInfos);
     }
 
-    public static ThreadDumpInfo dumpAndCreate(int stacktraceMaxDepth) {
+    /** Dumps all threads of the current JVM at the granularity indicated by {@code mode}. */
+    public static ThreadDumpInfo dumpAndCreate(int stacktraceMaxDepth, ThreadDumpMode mode) {
         return create(
-                JvmUtils.createThreadDump().stream()
+                JvmUtils.createThreadDump(mode.isLockedMonitors(), mode.isLockedSynchronizers())
+                        .stream()
                         .map(
                                 threadInfo ->
                                         ThreadDumpInfo.ThreadInfo.create(
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ThreadDumpMode.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ThreadDumpMode.java
new file mode 100644
index 00000000000000..d3c44186cdd7b4
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ThreadDumpMode.java
@@ -0,0 +1,83 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.rest.messages;
+
+/**
+ * Granularity of the thread dump collected via {@link
+ * java.lang.management.ThreadMXBean#dumpAllThreads(boolean, boolean)}. Information about the lock
+ * each thread is currently waiting on ({@link java.lang.management.ThreadInfo#getLockInfo()}) is
+ * populated in both modes.
+ */
+public enum ThreadDumpMode {
+
+    /**
+     * {@code dumpAllThreads(false, false)}: stack traces only, no lock info (jstack without {@code
+     * -l}). Negligible JVM pause.
+     */
+    LITE(false, false),
+
+    /**
+     * {@code dumpAllThreads(true, true)}: also collects locked monitors and j.u.c. synchronizers
+     * (equivalent to {@code jstack -l}). Pauses the JVM in a safepoint for a duration that scales
+     * with heap size and thread count -- seconds on large TaskManagers.
+     */
+    FULL(true, true);
+
+    private final boolean lockedMonitors;
+    private final boolean lockedSynchronizers;
+
+    ThreadDumpMode(boolean lockedMonitors, boolean lockedSynchronizers) {
+        this.lockedMonitors = lockedMonitors;
+        this.lockedSynchronizers = lockedSynchronizers;
+    }
+
+    public boolean isLockedMonitors() {
+        return lockedMonitors;
+    }
+
+    public boolean isLockedSynchronizers() {
+        return lockedSynchronizers;
+    }
+
+    /**
+     * Resolves a mode name (case-insensitive), returning {@code fallback} when the input is
+     * null/blank or does not match a known mode.
+     */
+    public static ThreadDumpMode fromStringOrDefault(String name, ThreadDumpMode fallback) {
+        if (name == null || name.trim().isEmpty()) {
+            return fallback;
+        }
+        for (ThreadDumpMode m : values()) {
+            if (m.name().equalsIgnoreCase(name.trim())) {
+                return m;
+            }
+        }
+        return fallback;
+    }
+
+    /**
+     * Returns the effective mode for a REST request: {@code explicit} takes precedence; otherwise
+     * {@code clusterDefaultName} (typically the value of {@code cluster.thread-dump.default-mode})
+     * is parsed via {@link #fromStringOrDefault(String, ThreadDumpMode)}, falling back to {@link
+     * #FULL} for null/blank/unknown values (matching the historical behavior).
+     */
+    public static ThreadDumpMode resolve(ThreadDumpMode explicit, String clusterDefaultName) {
+        return explicit != null ? explicit : fromStringOrDefault(clusterDefaultName, FULL);
+    }
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ThreadDumpModeQueryParameter.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ThreadDumpModeQueryParameter.java
new file mode 100644
index 00000000000000..926747f443bb52
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ThreadDumpModeQueryParameter.java
@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.rest.messages;
+
+import java.util.Arrays;
+
+/**
+ * Optional {@code mode} query parameter for the thread-dump REST endpoints. When omitted, the
+ * cluster default ({@code cluster.thread-dump.default-mode}) is used.
+ */
+public class ThreadDumpModeQueryParameter extends MessageQueryParameter<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());
+    }
+
+    @Override
+    public String convertValueToString(ThreadDumpMode value) {
+        return value.name().toLowerCase();
+    }
+
+    @Override
+    public String getDescription() {
+        return "Controls how much lock information is collected. Supported values: "
+                + Arrays.toString(ThreadDumpMode.values())
+                + ". When omitted, cluster.thread-dump.default-mode is used.";
+    }
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/cluster/JobManagerThreadDumpHeaders.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/cluster/JobManagerThreadDumpHeaders.java
index f0dfba9439a749..b021d8f5889acb 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/cluster/JobManagerThreadDumpHeaders.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/cluster/JobManagerThreadDumpHeaders.java
@@ -20,7 +20,6 @@
 
 import org.apache.flink.runtime.rest.HttpMethodWrapper;
 import org.apache.flink.runtime.rest.handler.cluster.JobManagerThreadDumpHandler;
-import org.apache.flink.runtime.rest.messages.EmptyMessageParameters;
 import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
 import org.apache.flink.runtime.rest.messages.RuntimeMessageHeaders;
 import org.apache.flink.runtime.rest.messages.ThreadDumpInfo;
@@ -29,7 +28,8 @@
 
 /** Headers for the {@link JobManagerThreadDumpHandler}. */
 public class JobManagerThreadDumpHeaders
-        implements RuntimeMessageHeaders<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 e5b9fc9b79d4ca..6b3da4b2cf29a1 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java
@@ -23,6 +23,7 @@
 import org.apache.flink.api.common.JobID;
 import org.apache.flink.configuration.BatchExecutionOptions;
 import org.apache.flink.configuration.ClusterOptions;
+import org.apache.flink.configuration.Configuration;
 import org.apache.flink.management.jmx.JMXService;
 import org.apache.flink.runtime.accumulators.AccumulatorSnapshot;
 import org.apache.flink.runtime.blob.JobPermanentBlobService;
@@ -95,6 +96,7 @@
 import org.apache.flink.runtime.rest.messages.LogInfo;
 import org.apache.flink.runtime.rest.messages.ProfilingInfo;
 import org.apache.flink.runtime.rest.messages.ThreadDumpInfo;
+import org.apache.flink.runtime.rest.messages.ThreadDumpMode;
 import org.apache.flink.runtime.rpc.FatalErrorHandler;
 import org.apache.flink.runtime.rpc.RpcEndpoint;
 import org.apache.flink.runtime.rpc.RpcService;
@@ -1460,12 +1462,14 @@ public CompletableFuture<Acknowledge> sendOperatorEventToTask(
     }
 
     @Override
-    public CompletableFuture<ThreadDumpInfo> requestThreadDump(Duration timeout) {
-        int stacktraceMaxDepth =
-                taskManagerConfiguration
-                        .getConfiguration()
-                        .get(ClusterOptions.THREAD_DUMP_STACKTRACE_MAX_DEPTH);
-        return CompletableFuture.completedFuture(ThreadDumpInfo.dumpAndCreate(stacktraceMaxDepth));
+    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 =
+                ThreadDumpMode.resolve(mode, config.get(ClusterOptions.THREAD_DUMP_DEFAULT_MODE));
+        return CompletableFuture.completedFuture(
+                ThreadDumpInfo.dumpAndCreate(stacktraceMaxDepth, resolvedMode));
     }
 
     @Override
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorGateway.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorGateway.java
index 29607c2c2f8130..a330682f6a2af0 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorGateway.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorGateway.java
@@ -41,6 +41,7 @@
 import org.apache.flink.runtime.rest.messages.LogInfo;
 import org.apache.flink.runtime.rest.messages.ProfilingInfo;
 import org.apache.flink.runtime.rest.messages.ThreadDumpInfo;
+import org.apache.flink.runtime.rest.messages.ThreadDumpMode;
 import org.apache.flink.runtime.rpc.RpcGateway;
 import org.apache.flink.runtime.rpc.RpcTimeout;
 import org.apache.flink.runtime.shuffle.PartitionWithMetrics;
@@ -306,10 +307,12 @@ CompletableFuture<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..1f40bfaa61a6bf 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorGatewayDecoratorBase.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorGatewayDecoratorBase.java
@@ -41,6 +41,7 @@
 import org.apache.flink.runtime.rest.messages.LogInfo;
 import org.apache.flink.runtime.rest.messages.ProfilingInfo;
 import org.apache.flink.runtime.rest.messages.ThreadDumpInfo;
+import org.apache.flink.runtime.rest.messages.ThreadDumpMode;
 import org.apache.flink.runtime.webmonitor.threadinfo.ThreadInfoSamplesRequest;
 import org.apache.flink.types.SerializableOptional;
 import org.apache.flink.util.SerializedValue;
@@ -238,8 +239,9 @@ public CompletableFuture<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..d5f4114018a9cf 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/NonLeaderRetrievalRestfulGateway.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/NonLeaderRetrievalRestfulGateway.java
@@ -30,6 +30,7 @@
 import org.apache.flink.runtime.messages.webmonitor.MultipleApplicationsDetails;
 import org.apache.flink.runtime.messages.webmonitor.MultipleJobsDetails;
 import org.apache.flink.runtime.rest.messages.ThreadDumpInfo;
+import org.apache.flink.runtime.rest.messages.ThreadDumpMode;
 import org.apache.flink.runtime.scheduler.ExecutionGraphInfo;
 
 import java.time.Duration;
@@ -125,7 +126,8 @@ public CompletableFuture<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..5191941a07ff3b 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/RestfulGateway.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/RestfulGateway.java
@@ -44,6 +44,7 @@
 import org.apache.flink.runtime.rest.handler.async.OperationResult;
 import org.apache.flink.runtime.rest.handler.job.AsynchronousJobOperationKey;
 import org.apache.flink.runtime.rest.messages.ThreadDumpInfo;
+import org.apache.flink.runtime.rest.messages.ThreadDumpMode;
 import org.apache.flink.runtime.rpc.RpcGateway;
 import org.apache.flink.runtime.rpc.RpcTimeout;
 import org.apache.flink.runtime.scheduler.ExecutionGraphInfo;
@@ -188,10 +189,12 @@ CompletableFuture<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/resourcemanager/utils/TestingResourceManagerGateway.java b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/utils/TestingResourceManagerGateway.java
index 603eee966255e2..f7c691ae8a7aea 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/utils/TestingResourceManagerGateway.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/utils/TestingResourceManagerGateway.java
@@ -46,6 +46,7 @@
 import org.apache.flink.runtime.rest.messages.LogInfo;
 import org.apache.flink.runtime.rest.messages.ProfilingInfo;
 import org.apache.flink.runtime.rest.messages.ThreadDumpInfo;
+import org.apache.flink.runtime.rest.messages.ThreadDumpMode;
 import org.apache.flink.runtime.rest.messages.taskmanager.TaskManagerInfo;
 import org.apache.flink.runtime.shuffle.ShuffleDescriptor;
 import org.apache.flink.runtime.slots.ResourceRequirements;
@@ -475,7 +476,7 @@ public CompletableFuture<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..d8f16baade9377
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/ThreadDumpModeQueryParameterTest.java
@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.rest.messages;
+
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link ThreadDumpModeQueryParameter}. */
+class ThreadDumpModeQueryParameterTest {
+
+    private final ThreadDumpModeQueryParameter param = new ThreadDumpModeQueryParameter();
+
+    @Test
+    void keyIsMode() {
+        assertThat(param.getKey()).isEqualTo("mode");
+    }
+
+    @Test
+    void parameterIsOptional() {
+        assertThat(param.isMandatory()).isFalse();
+    }
+
+    @Test
+    void 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/TaskExecutorThreadDumpTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorThreadDumpTest.java
new file mode 100644
index 00000000000000..66114c937ff1e1
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorThreadDumpTest.java
@@ -0,0 +1,105 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.taskexecutor;
+
+import org.apache.flink.configuration.ClusterOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.testutils.EachCallbackWrapper;
+import org.apache.flink.runtime.entrypoint.WorkingDirectory;
+import org.apache.flink.runtime.highavailability.TestingHighAvailabilityServicesBuilder;
+import org.apache.flink.runtime.rest.messages.ThreadDumpInfo;
+import org.apache.flink.runtime.rest.messages.ThreadDumpMode;
+import org.apache.flink.runtime.rpc.RpcUtils;
+import org.apache.flink.runtime.rpc.TestingRpcServiceExtension;
+import org.apache.flink.util.TestLoggerExtension;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.File;
+import java.time.Duration;
+import java.util.concurrent.TimeUnit;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests for {@link TaskExecutor#requestThreadDump(ThreadDumpMode, Duration)}: verifies that the RPC
+ * returns a non-empty dump for each supported mode and honors the {@code
+ * cluster.thread-dump.default-mode} fallback.
+ */
+@ExtendWith(TestLoggerExtension.class)
+class TaskExecutorThreadDumpTest {
+
+    private static final Duration RPC_TIMEOUT = Duration.ofSeconds(10);
+
+    private final TestingRpcServiceExtension rpcServiceExtension = new TestingRpcServiceExtension();
+
+    @RegisterExtension
+    private final EachCallbackWrapper<TestingRpcServiceExtension> eachWrapper =
+            new EachCallbackWrapper<>(rpcServiceExtension);
+
+    @Test
+    void requestThreadDumpReturnsNonEmptyDumpForEachMode(@TempDir File tempDir) throws Exception {
+        for (ThreadDumpMode mode :
+                new ThreadDumpMode[] {null, ThreadDumpMode.LITE, ThreadDumpMode.FULL}) {
+            final ThreadDumpInfo dump = requestDump(tempDir, new Configuration(), mode);
+            assertThat(dump.getThreadInfos()).isNotEmpty();
+        }
+    }
+
+    @Test
+    void requestThreadDumpFallsBackToClusterDefaultWhenModeOmitted(@TempDir File tempDir)
+            throws Exception {
+        // Override the cluster default to 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, "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 {
+        final TaskExecutor taskExecutor =
+                TaskExecutorBuilder.newBuilder(
+                                rpcServiceExtension.getTestingRpcService(),
+                                new TestingHighAvailabilityServicesBuilder().build(),
+                                WorkingDirectory.create(tempDir))
+                        .setConfiguration(configuration)
+                        .build();
+        try {
+            taskExecutor.start();
+            return taskExecutor
+                    .getSelfGateway(TaskExecutorGateway.class)
+                    .requestThreadDump(mode, RPC_TIMEOUT)
+                    .get(20, TimeUnit.SECONDS);
+        } finally {
+            RpcUtils.terminateRpcEndpoint(taskExecutor);
+        }
+    }
+}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TestingTaskExecutorGateway.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TestingTaskExecutorGateway.java
index cf1b79c68e643a..77243b8e028cf4 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TestingTaskExecutorGateway.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TestingTaskExecutorGateway.java
@@ -42,6 +42,7 @@
 import org.apache.flink.runtime.rest.messages.LogInfo;
 import org.apache.flink.runtime.rest.messages.ProfilingInfo;
 import org.apache.flink.runtime.rest.messages.ThreadDumpInfo;
+import org.apache.flink.runtime.rest.messages.ThreadDumpMode;
 import org.apache.flink.runtime.shuffle.PartitionWithMetrics;
 import org.apache.flink.runtime.webmonitor.threadinfo.ThreadInfoSamplesRequest;
 import org.apache.flink.types.SerializableOptional;
@@ -371,7 +372,8 @@ public CompletableFuture<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..f6eac7b17b7b6c 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/TestingRestfulGateway.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/TestingRestfulGateway.java
@@ -40,6 +40,7 @@
 import org.apache.flink.runtime.rest.handler.async.OperationResult;
 import org.apache.flink.runtime.rest.handler.job.AsynchronousJobOperationKey;
 import org.apache.flink.runtime.rest.messages.ThreadDumpInfo;
+import org.apache.flink.runtime.rest.messages.ThreadDumpMode;
 import org.apache.flink.runtime.scheduler.ExecutionGraphInfo;
 import org.apache.flink.util.SerializedValue;
 import org.apache.flink.util.concurrent.FutureUtils;
@@ -406,7 +407,8 @@ public CompletableFuture<Collection<String>> requestMetricQueryServiceAddresses(
     }
 
     @Override
-    public CompletableFuture<ThreadDumpInfo> requestThreadDump(Duration timeout) {
+    public CompletableFuture<ThreadDumpInfo> requestThreadDump(
+            ThreadDumpMode mode, Duration timeout) {
         return null;
     }