diff --git a/docs/api-reference/service-status-api.md b/docs/api-reference/service-status-api.md
index 1a712603281e..68b5cfa5ee3d 100644
--- a/docs/api-reference/service-status-api.md
+++ b/docs/api-reference/service-status-api.md
@@ -167,6 +167,47 @@ Host: http://ROUTER_IP:ROUTER_PORT
   ```
 </details>
 
+### Get service thread stack traces
+
+Retrieves an on-demand snapshot of the live Java platform threads running in the individual Druid service. Virtual threads are not included. The response includes each thread's state, cumulative CPU times when supported, lock information, deadlock status, and up to `maxStackTraceFrameDepth` stack frames. The `collectedAt` timestamp is recorded when snapshot collection begins, and `stackTrace` is a single jstack-style string. This endpoint requires the same state-read permission as `GET /status`.
+
+#### URL
+
+`GET` `/status/stack[?maxStackTraceFrameDepth=N]`
+
+The optional `maxStackTraceFrameDepth` query parameter controls the maximum number of stack frames returned for each thread. It defaults to `100`, accepts integer values from `10` through `1000`, and returns a `400 Bad Request` response for values outside that range or values that are not integers.
+
+#### Responses
+
+`200 SUCCESS` returns a JSON object containing the UTC collection timestamp and an array of thread snapshots. The `cpuTimeNs` and `userCpuTimeNs` fields are omitted when thread CPU timing is unsupported, disabled, or unavailable for an individual thread. Optional lock-related fields are omitted when the JVM cannot provide them.
+
+#### Sample request
+
+```shell
+curl "http://ROUTER_IP:ROUTER_PORT/status/stack?maxStackTraceFrameDepth=25"
+```
+
+#### Sample response
+
+```json
+{
+  "collectedAt": "2026-08-02T05:00:00.000Z",
+  "threads": [
+    {
+      "threadId": 1,
+      "threadName": "main",
+      "threadState": "RUNNABLE",
+      "daemon": false,
+      "priority": 5,
+      "cpuTimeNs": 123456789,
+      "userCpuTimeNs": 98765432,
+      "deadlocked": false,
+      "stackTrace": "\"main\" prio=5 Id=1 RUNNABLE\n\tat example.Main.run(Main.java:10)\n\n"
+    }
+  ]
+}
+```
+
 ### Get service health
 
 Retrieves the online status of the individual Druid service. It is a simple health check to determine if the service is running and accessible. If online, it will always return a boolean `true` value, indicating that the service can receive API calls. This endpoint is suitable for automated health checks.
diff --git a/docs/querying/sql-metadata-tables.md b/docs/querying/sql-metadata-tables.md
index 055305d81e50..7140ba2dc412 100644
--- a/docs/querying/sql-metadata-tables.md
+++ b/docs/querying/sql-metadata-tables.md
@@ -339,6 +339,53 @@ For example, to retrieve properties for a specific server, use the query
 SELECT * FROM sys.server_properties WHERE server='192.168.1.1:8081'
 ```
 
+### STACK_TRACE table
+
+The `stack_trace` table exposes a live Java platform-thread snapshot collected from explicitly selected Druid servers. Virtual threads are not included. Each thread in a successful snapshot produces one row. The table requires an equality or `IN` filter on `server`, using the same `host:port` value as `sys.servers.server`, to avoid unintentionally collecting stack traces from the entire cluster. If a snapshot cannot be retrieved, the table returns one placeholder row for that server with `server`, `service_name`, `node_roles`, and `error_message` populated; all remaining columns are null.
+
+The optional SQL query context parameter `maxStackTraceFrameDepth` controls the maximum number of stack frames in the `stack` column for each thread. It defaults to `100` and accepts effective values from `10` through `1000`. Conversion follows the standard Druid query context rules. Unquoted JSON numbers are converted using `Number.longValue()` and therefore truncate fractional values toward zero before range validation; for example, `10.9` becomes `10`, while `9.9` becomes `9` and is rejected. Quoted values are parsed exactly: `"10"` and `"10.0"` become `10`, while `"10.9"` is rejected.
+
+|Column|Type|Notes|
+|------|-----|-----|
+|server|VARCHAR|Host and port of the server, in the form `host:port`; aligns with `sys.servers.server`|
+|service_name|VARCHAR|Service name of the server, as defined by `druid.service`|
+|node_roles|VARCHAR|Comma-separated, lexicographically sorted list of roles announced by the process. A process with multiple roles, such as `coordinator` and `overlord`, is represented as `coordinator,overlord`|
+|collected_at|VARCHAR|UTC ISO-8601 timestamp recorded when collection of the server snapshot begins. All successful thread rows from one server snapshot have the same value|
+|thread_id|BIGINT|JVM thread identifier, unique while the thread exists|
+|thread_name|VARCHAR|JVM thread name|
+|thread_state|VARCHAR|JVM state of the live platform thread, such as `RUNNABLE`, `BLOCKED`, `WAITING`, or `TIMED_WAITING`|
+|daemon|BIGINT|Boolean represented as long type where 1 = true and 0 = false|
+|priority|BIGINT|JVM thread priority|
+|cpu_time_ns|BIGINT|Cumulative thread CPU time in nanoseconds at collection time. Null when unavailable or disabled|
+|user_cpu_time_ns|BIGINT|Cumulative thread user CPU time in nanoseconds at collection time. Null when unavailable or disabled|
+|lock_name|VARCHAR|Name of the object or synchronizer on which the thread is blocked or waiting, when available; otherwise null|
+|lock_owner_id|BIGINT|Identifier of the thread owning the lock, when available|
+|lock_owner_name|VARCHAR|Name of the thread owning the lock, when available|
+|is_deadlocked|BIGINT|Boolean represented as long type where 1 = the JVM reports the thread as part of a deadlock and 0 = it is not reported as deadlocked|
+|stack|VARCHAR|One jstack-style string containing the thread header, up to `maxStackTraceFrameDepth` stack frames, and lock or synchronizer annotations. The default is 100 frames. Unlike `ThreadInfo.toString()`, the stack is not truncated to eight frames|
+|error_message|VARCHAR|Describes why the snapshot could not be retrieved, such as an HTTP or connection error. Null for successful thread rows|
+
+The `cpu_time_ns` and `user_cpu_time_ns` values are nullable. Do not assume that either value is present, because availability depends on JVM support and configuration, and a thread can terminate while its snapshot is being collected.
+
+For example, to inspect threads and cumulative CPU time on one Broker:
+
+```sql
+SELECT thread_name, thread_state, cpu_time_ns, user_cpu_time_ns, stack
+FROM sys.stack_trace
+WHERE server = '192.168.1.1:8082'
+```
+
+To request a deeper stack snapshot, include the query context in the SQL request:
+
+```json
+{
+  "query": "SELECT thread_name, stack FROM sys.stack_trace WHERE server = '192.168.1.1:8082'",
+  "context": {
+    "maxStackTraceFrameDepth": 250
+  }
+}
+```
+
 ### QUERIES table
 
 :::info
diff --git a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/schema/SystemStackTraceTableTest.java b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/schema/SystemStackTraceTableTest.java
new file mode 100644
index 000000000000..d0e033216979
--- /dev/null
+++ b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/schema/SystemStackTraceTableTest.java
@@ -0,0 +1,240 @@
+/*
+ * 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.druid.testing.embedded.schema;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.query.http.ClientSqlQuery;
+import org.apache.druid.rpc.RequestBuilder;
+import org.apache.druid.server.StackTraceCollector;
+import org.apache.druid.sql.http.ResultFormat;
+import org.apache.druid.testing.embedded.EmbeddedBroker;
+import org.apache.druid.testing.embedded.EmbeddedCoordinator;
+import org.apache.druid.testing.embedded.EmbeddedDruidCluster;
+import org.apache.druid.testing.embedded.EmbeddedOverlord;
+import org.apache.druid.testing.embedded.junit5.EmbeddedClusterTestBase;
+import org.jboss.netty.handler.codec.http.HttpMethod;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.Map;
+
+public class SystemStackTraceTableTest extends EmbeddedClusterTestBase
+{
+  private static final String BROKER_PORT = "9082";
+  private static final String BROKER_SERVICE = "test/broker";
+  private static final String OVERLORD_PORT = "9090";
+  private static final String OVERLORD_SERVICE = "test/overlord";
+  private static final String COORDINATOR_PORT = "9081";
+  private static final String COORDINATOR_SERVICE = "test/coordinator";
+
+  private final EmbeddedBroker broker = new EmbeddedBroker()
+      .addProperty("druid.service", BROKER_SERVICE)
+      .addProperty("druid.plaintextPort", BROKER_PORT);
+
+  private final EmbeddedOverlord overlord = new EmbeddedOverlord()
+      .addProperty("druid.service", OVERLORD_SERVICE)
+      .addProperty("druid.plaintextPort", OVERLORD_PORT);
+
+  private final EmbeddedCoordinator coordinator = new EmbeddedCoordinator()
+      .addProperty("druid.service", COORDINATOR_SERVICE)
+      .addProperty("druid.plaintextPort", COORDINATOR_PORT);
+
+  @Override
+  protected EmbeddedDruidCluster createCluster()
+  {
+    return EmbeddedDruidCluster
+        .withZookeeper()
+        .addServer(coordinator)
+        .addServer(overlord)
+        .addServer(broker);
+  }
+
+  @Test
+  public void test_stackTraceEndpoint()
+  {
+    final StackTraceCollector.ThreadStackTraceResponse response = cluster.callApi().serviceClient().onAnyBroker(
+        mapper -> new RequestBuilder(HttpMethod.GET, "/status/stack"),
+        new TypeReference<>(){}
+    );
+
+    Assertions.assertNotNull(response.getCollectedAt());
+    Assertions.assertFalse(response.getThreads().isEmpty());
+    Assertions.assertTrue(
+        response.getThreads().stream().allMatch(thread -> !thread.getThreadName().isEmpty())
+    );
+    Assertions.assertTrue(
+        response.getThreads().stream().anyMatch(thread -> !thread.getStackTrace().isEmpty())
+    );
+    Assertions.assertTrue(
+        response.getThreads().stream().allMatch(thread -> countStackFrames(thread.getStackTrace()) <= 100)
+    );
+  }
+
+  @Test
+  public void test_stackTraceEndpointWithMaxStackTraceFrameDepth()
+  {
+    final StackTraceCollector.ThreadStackTraceResponse response = cluster.callApi().serviceClient().onAnyBroker(
+        mapper -> new RequestBuilder(HttpMethod.GET, "/status/stack?maxStackTraceFrameDepth=10"),
+        new TypeReference<>(){}
+    );
+
+    Assertions.assertTrue(
+        response.getThreads().stream().allMatch(thread -> countStackFrames(thread.getStackTrace()) <= 10)
+    );
+  }
+
+  @Test
+  public void test_stackTraceEndpointRejectsInvalidMaxStackTraceFrameDepth()
+  {
+    for (final String invalidDepth : new String[]{"9", "1001", "10.5"}) {
+      final RuntimeException exception = Assertions.assertThrows(
+          RuntimeException.class,
+          () -> cluster.callApi().serviceClient().onAnyBroker(
+              mapper -> new RequestBuilder(
+                  HttpMethod.GET,
+                  StringUtils.format("/status/stack?maxStackTraceFrameDepth=%s", invalidDepth)
+              ),
+              new TypeReference<StackTraceCollector.ThreadStackTraceResponse>(){}
+          )
+      );
+      Assertions.assertTrue(exception.getMessage().contains("400 Bad Request"), exception.getMessage());
+      Assertions.assertTrue(
+          exception.getMessage().contains(StackTraceCollector.MAX_STACK_TRACE_FRAME_DEPTH_KEY),
+          exception.getMessage()
+      );
+    }
+  }
+
+  @Test
+  public void test_stackTraceTable()
+  {
+    final String brokerHost = StringUtils.format("localhost:%s", BROKER_PORT);
+    final String result = cluster.runSql(
+        "SELECT server, service_name, node_roles, collected_at, thread_id, "
+        + "thread_state, daemon, priority, cpu_time_ns, user_cpu_time_ns, is_deadlocked, error_message "
+        + "FROM sys.stack_trace WHERE server = '%s'",
+        brokerHost
+    );
+
+    Assertions.assertFalse(result.isEmpty(), "The stack trace table should return broker threads");
+    for (final String row : result.split("\\n")) {
+      final String[] columns = row.split(",", -1);
+      Assertions.assertEquals(brokerHost, columns[0]);
+      Assertions.assertEquals(BROKER_SERVICE, columns[1]);
+      Assertions.assertEquals("broker", columns[2]);
+      Assertions.assertFalse(columns[3].isEmpty());
+      assertLong(columns[4]);
+      Assertions.assertFalse(columns[6].isEmpty());
+      Assertions.assertTrue(columns[6].equals("0") || columns[6].equals("1"), row);
+      assertLong(columns[7]);
+      if (!columns[8].isEmpty()) {
+        assertLong(columns[8]);
+      }
+      if (!columns[9].isEmpty()) {
+        assertLong(columns[9]);
+      }
+      Assertions.assertTrue(columns[10].equals("0") || columns[10].equals("1"), row);
+      Assertions.assertTrue(columns[11].isEmpty());
+    }
+
+    Assertions.assertFalse(
+        cluster.runSql("SELECT thread_name FROM sys.stack_trace WHERE server = '%s' LIMIT 1", brokerHost).isEmpty()
+    );
+
+    Assertions.assertFalse(
+        cluster.runSql(
+            "SELECT server FROM sys.stack_trace WHERE server = '%s' AND node_roles = 'broker' LIMIT 1",
+            brokerHost
+        ).isEmpty()
+    );
+
+    Assertions.assertFalse(
+        cluster.runSql("SELECT stack FROM sys.stack_trace WHERE server = '%s' LIMIT 1", brokerHost).isEmpty()
+    );
+  }
+
+  @Test
+  public void test_stackTraceTableWithMaxStackTraceFrameDepth()
+  {
+    final String brokerHost = StringUtils.format("localhost:%s", BROKER_PORT);
+    final String result = cluster.callApi().onAnyBroker(
+        broker -> broker.submitSqlQuery(
+            new ClientSqlQuery(
+                StringUtils.format(
+                    "SELECT stack FROM sys.stack_trace WHERE server = '%s' LIMIT 1",
+                    brokerHost
+                ),
+                ResultFormat.CSV.name(),
+                false,
+                false,
+                false,
+                Map.of(StackTraceCollector.MAX_STACK_TRACE_FRAME_DEPTH_KEY, 10.9),
+                null
+            )
+        )
+    ).trim();
+
+    Assertions.assertFalse(result.isEmpty());
+    Assertions.assertTrue(countStackFrames(result) <= 10);
+  }
+
+  @Test
+  public void test_stackTraceTable_requiresServerFilter()
+  {
+    final RuntimeException exception = Assertions.assertThrows(
+        RuntimeException.class,
+        () -> cluster.runSql("SELECT COUNT(*) FROM sys.stack_trace")
+    );
+    Assertions.assertTrue(exception.getMessage().contains("400 Bad Request"), exception.getMessage());
+    Assertions.assertTrue(exception.getMessage().contains("requires a filter on the server column"));
+  }
+
+  @Test
+  public void test_stackTraceTable_inFilter()
+  {
+    final String brokerHost = StringUtils.format("localhost:%s", BROKER_PORT);
+    final String coordinatorHost = StringUtils.format("localhost:%s", COORDINATOR_PORT);
+    final String result = cluster.runSql(
+        "SELECT DISTINCT server FROM sys.stack_trace WHERE server IN ('%s', '%s')",
+        brokerHost,
+        coordinatorHost
+    );
+
+    Assertions.assertTrue(result.contains(brokerHost));
+    Assertions.assertTrue(result.contains(coordinatorHost));
+    Assertions.assertFalse(result.contains(StringUtils.format("localhost:%s", OVERLORD_PORT)));
+  }
+
+  private static long countStackFrames(final String stackTrace)
+  {
+    return stackTrace.lines().filter(line -> line.startsWith("\tat ")).count();
+  }
+
+  private static void assertLong(final String value)
+  {
+    try {
+      Long.parseLong(value);
+    }
+    catch (NumberFormatException e) {
+      Assertions.fail("Expected a long value but got[" + value + "]", e);
+    }
+  }
+}
diff --git a/processing/src/main/java/org/apache/druid/java/util/http/client/NettyHttpClient.java b/processing/src/main/java/org/apache/druid/java/util/http/client/NettyHttpClient.java
index 76a50a711273..d67258b91f6a 100644
--- a/processing/src/main/java/org/apache/druid/java/util/http/client/NettyHttpClient.java
+++ b/processing/src/main/java/org/apache/druid/java/util/http/client/NettyHttpClient.java
@@ -23,6 +23,7 @@
 import com.google.common.collect.Multimap;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.MoreExecutors;
 import com.google.common.util.concurrent.SettableFuture;
 import org.apache.druid.java.util.common.IAE;
 import org.apache.druid.java.util.common.ISE;
@@ -173,8 +174,25 @@ public <Intermediate, Final> ListenableFuture<Final> go(
 
     // Pipeline can hand us chunks even after exceptionCaught is called. This has the potential to confuse
     // HttpResponseHandler implementations, which expect exceptionCaught to be the final method called. So, we
-    // use this boolean to ensure that handlers do not see any chunks after exceptionCaught fires.
+    // use this boolean to ensure that handlers do not see any chunks after exceptionCaught fires. Cancellation is
+    // tracked by retVal.isCancelled() below for the same reason.
     final AtomicBoolean didEncounterException = new AtomicBoolean();
+    final AtomicBoolean resourceReturned = new AtomicBoolean();
+    final Runnable returnResource = () -> {
+      if (resourceReturned.compareAndSet(false, true)) {
+        channelResourceContainer.returnResource();
+      }
+    };
+
+    retVal.addListener(
+        () -> {
+          if (retVal.isCancelled()) {
+            log.debug("[%s] Request cancelled, closing channel.", requestDesc);
+            channel.close().addListener(future -> returnResource.run());
+          }
+        },
+        MoreExecutors.directExecutor()
+    );
 
     if (readTimeout > 0) {
       channel.getPipeline().addLast(
@@ -209,8 +227,8 @@ public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
               Object msg = e.getMessage();
 
               if (msg instanceof HttpResponse) {
-                if (didEncounterException.get()) {
-                  // Don't process HttpResponse after encountering an exception.
+                if (didEncounterException.get() || retVal.isCancelled()) {
+                  // Don't process HttpResponse after encountering an exception or request cancellation.
                   return;
                 }
 
@@ -259,8 +277,8 @@ public void abort()
                   finishRequest();
                 }
               } else if (msg instanceof HttpChunk) {
-                if (didEncounterException.get()) {
-                  // Don't process HttpChunk after encountering an exception.
+                if (didEncounterException.get() || retVal.isCancelled()) {
+                  // Don't process HttpChunk after encountering an exception or request cancellation.
                   return;
                 }
 
@@ -294,7 +312,7 @@ public void abort()
                 retVal.set(null);
               }
               channel.close();
-              channelResourceContainer.returnResource();
+              returnResource.run();
 
               throw ex;
             }
@@ -332,7 +350,7 @@ private void finishRequest()
             }
             removeHandlers();
             channel.setReadable(true);
-            channelResourceContainer.returnResource();
+            returnResource.run();
           }
 
           @Override
@@ -392,7 +410,7 @@ private void handleExceptionAndCloseChannel(final Throwable t, final boolean clo
               log.warn(e, "[%s] Error while closing channel", requestDesc);
             }
             finally {
-              channelResourceContainer.returnResource();
+              returnResource.run();
             }
           }
 
@@ -414,7 +432,7 @@ public void operationComplete(ChannelFuture future)
           {
             if (!future.isSuccess()) {
               channel.close();
-              channelResourceContainer.returnResource();
+              returnResource.run();
               if (!retVal.isDone()) {
                 retVal.setException(
                     new ChannelException(
diff --git a/processing/src/test/java/org/apache/druid/java/util/http/client/FriendlyServersTest.java b/processing/src/test/java/org/apache/druid/java/util/http/client/FriendlyServersTest.java
index 4df73564389e..4453d8e2196a 100644
--- a/processing/src/test/java/org/apache/druid/java/util/http/client/FriendlyServersTest.java
+++ b/processing/src/test/java/org/apache/druid/java/util/http/client/FriendlyServersTest.java
@@ -49,9 +49,12 @@
 import java.net.Socket;
 import java.net.URL;
 import java.nio.charset.StandardCharsets;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 
@@ -115,6 +118,95 @@ public void run()
     }
   }
 
+  @Test
+  public void testCancelRequestClosesConnection() throws Exception
+  {
+    final CountDownLatch requestReceived = new CountDownLatch(1);
+    final CountDownLatch connectionClosed = new CountDownLatch(1);
+    final ExecutorService exec = Executors.newSingleThreadExecutor();
+    final ExecutorService requestExec = Executors.newSingleThreadExecutor();
+    final ServerSocket serverSocket = new ServerSocket(0);
+    exec.submit(
+        new Runnable()
+        {
+          @Override
+          public void run()
+          {
+            try {
+              try (
+                  Socket clientSocket = serverSocket.accept();
+                  BufferedReader in = new BufferedReader(
+                      new InputStreamReader(clientSocket.getInputStream(), StandardCharsets.UTF_8)
+                  )
+              ) {
+                while (!in.readLine().equals("")) {
+                  // skip lines
+                }
+                requestReceived.countDown();
+                while (in.read() != -1) {
+                  // Wait for the client to close the connection.
+                }
+              }
+              finally {
+                connectionClosed.countDown();
+              }
+
+              try (
+                  Socket clientSocket = serverSocket.accept();
+                  BufferedReader in = new BufferedReader(
+                      new InputStreamReader(clientSocket.getInputStream(), StandardCharsets.UTF_8)
+                  );
+                  OutputStream out = clientSocket.getOutputStream()
+              ) {
+                while (!in.readLine().equals("")) {
+                  // skip lines
+                }
+                out.write(
+                    "HTTP/1.1 200 OK\r\nContent-Length: 6\r\n\r\nhello!".getBytes(StandardCharsets.UTF_8)
+                );
+              }
+            }
+            catch (Exception e) {
+              // Suppress
+            }
+          }
+        }
+    );
+
+    final Lifecycle lifecycle = new Lifecycle();
+    try {
+      final HttpClient client = HttpClientInit.createClient(HttpClientConfig.builder().build(), lifecycle);
+      final ListenableFuture<StatusResponseHolder> future = client.go(
+          new Request(
+              HttpMethod.GET,
+              new URL(StringUtils.format("http://localhost:%d/", serverSocket.getLocalPort()))
+          ),
+          StatusResponseHandler.getInstance()
+      );
+
+      Assert.assertTrue(requestReceived.await(10, TimeUnit.SECONDS));
+      Assert.assertTrue(future.cancel(true));
+      Assert.assertTrue(connectionClosed.await(10, TimeUnit.SECONDS));
+
+      final Future<StatusResponseHolder> secondResponse = requestExec.submit(
+          () -> client.go(
+              new Request(
+                  HttpMethod.GET,
+                  new URL(StringUtils.format("http://localhost:%d/", serverSocket.getLocalPort()))
+              ),
+              StatusResponseHandler.getInstance()
+          ).get()
+      );
+      Assert.assertEquals("hello!", secondResponse.get(10, TimeUnit.SECONDS).getContent());
+    }
+    finally {
+      requestExec.shutdownNow();
+      exec.shutdownNow();
+      serverSocket.close();
+      lifecycle.stop();
+    }
+  }
+
   @Test
   public void testFriendlyProxyHttpServer() throws Exception
   {
diff --git a/server/src/main/java/org/apache/druid/server/StackTraceCollector.java b/server/src/main/java/org/apache/druid/server/StackTraceCollector.java
new file mode 100644
index 000000000000..4968f813f5a7
--- /dev/null
+++ b/server/src/main/java/org/apache/druid/server/StackTraceCollector.java
@@ -0,0 +1,429 @@
+/*
+ * 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.druid.server;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.collect.ImmutableList;
+import org.apache.druid.error.InvalidInput;
+import org.apache.druid.java.util.common.DateTimes;
+
+import javax.annotation.Nullable;
+import java.lang.management.LockInfo;
+import java.lang.management.ManagementFactory;
+import java.lang.management.MonitorInfo;
+import java.lang.management.ThreadInfo;
+import java.lang.management.ThreadMXBean;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * Collects a live snapshot of the Java platform threads running in the current Druid process.
+ *
+ * <p>This class intentionally has no instance state. A new collector can be created for each request.
+ */
+public class StackTraceCollector
+{
+  public static final String MAX_STACK_TRACE_FRAME_DEPTH_KEY = "maxStackTraceFrameDepth";
+  public static final int MIN_ALLOWED_STACK_TRACE_FRAME_DEPTH = 10;
+  public static final int DEFAULT_MAX_STACK_TRACE_FRAME_DEPTH = 100;
+  public static final int MAX_ALLOWED_STACK_TRACE_FRAME_DEPTH = 1000;
+
+  public ThreadStackTraceResponse collect()
+  {
+    return collect(DEFAULT_MAX_STACK_TRACE_FRAME_DEPTH);
+  }
+
+  public ThreadStackTraceResponse collect(final int maxStackTraceFrameDepth)
+  {
+    validateMaxStackTraceFrameDepth(maxStackTraceFrameDepth);
+    final String collectedAt = DateTimes.nowUtc().toString();
+    final ThreadMXBean threadMxBean = ManagementFactory.getThreadMXBean();
+    final boolean cpuTimeEnabled = isCpuTimeEnabled(threadMxBean);
+    final Set<Long> deadlockedThreadIds = findDeadlockedThreadIds(threadMxBean);
+    final long[] threadIds = threadMxBean.getAllThreadIds();
+    final ThreadInfo[] threadInfos = threadMxBean.getThreadInfo(
+        threadIds,
+        threadMxBean.isObjectMonitorUsageSupported(),
+        threadMxBean.isSynchronizerUsageSupported(),
+        maxStackTraceFrameDepth
+    );
+    final List<ThreadStackTrace> threads = new ArrayList<>(threadInfos.length);
+
+    for (final ThreadInfo threadInfo : threadInfos) {
+      if (threadInfo == null) {
+        continue;
+      }
+
+      final long threadId = threadInfo.getThreadId();
+      final long rawLockOwnerId = threadInfo.getLockOwnerId();
+      final Long lockOwnerId = rawLockOwnerId < 0 ? null : rawLockOwnerId;
+      threads.add(
+          new ThreadStackTrace(
+              threadId,
+              threadInfo.getThreadName(),
+              threadInfo.getThreadState().name(),
+              threadInfo.isDaemon(),
+              threadInfo.getPriority(),
+              getThreadCpuTime(threadMxBean, threadId, cpuTimeEnabled, false),
+              getThreadCpuTime(threadMxBean, threadId, cpuTimeEnabled, true),
+              threadInfo.getLockName(),
+              lockOwnerId,
+              threadInfo.getLockOwnerName(),
+              deadlockedThreadIds.contains(threadId),
+              formatThreadInfo(threadInfo)
+          )
+      );
+    }
+
+    return new ThreadStackTraceResponse(collectedAt, threads);
+  }
+
+  public static int parseMaxStackTraceFrameDepth(@Nullable final String value)
+  {
+    if (value == null) {
+      return DEFAULT_MAX_STACK_TRACE_FRAME_DEPTH;
+    }
+
+    try {
+      return validateMaxStackTraceFrameDepth(Long.parseLong(value));
+    }
+    catch (NumberFormatException e) {
+      throw InvalidInput.exception(
+          "Query parameter[%s] must be an integer, but got[%s]",
+          MAX_STACK_TRACE_FRAME_DEPTH_KEY,
+          value
+      );
+    }
+  }
+
+  public static int validateMaxStackTraceFrameDepth(final long maxStackTraceFrameDepth)
+  {
+    InvalidInput.conditionalException(
+        maxStackTraceFrameDepth >= MIN_ALLOWED_STACK_TRACE_FRAME_DEPTH,
+        "[%s] must be greater than or equal to %d, but got[%d]",
+        MAX_STACK_TRACE_FRAME_DEPTH_KEY,
+        MIN_ALLOWED_STACK_TRACE_FRAME_DEPTH,
+        maxStackTraceFrameDepth
+    );
+    InvalidInput.conditionalException(
+        maxStackTraceFrameDepth <= MAX_ALLOWED_STACK_TRACE_FRAME_DEPTH,
+        "[%s] must be less than or equal to %d, but got[%d]",
+        MAX_STACK_TRACE_FRAME_DEPTH_KEY,
+        MAX_ALLOWED_STACK_TRACE_FRAME_DEPTH,
+        maxStackTraceFrameDepth
+    );
+    return Math.toIntExact(maxStackTraceFrameDepth);
+  }
+
+  /**
+   * Formats a thread stack in a jstack-style format based on {@link ThreadInfo#toString()}, but
+   * includes all frames returned by the MX bean. {@code ThreadInfo.toString()} intentionally limits
+   * its output to eight frames and appends an ellipsis.
+   */
+  private static String formatThreadInfo(final ThreadInfo threadInfo)
+  {
+    final StringBuilder builder = new StringBuilder();
+    builder.append('"')
+           .append(threadInfo.getThreadName())
+           .append('"')
+           .append(threadInfo.isDaemon() ? " daemon" : "")
+           .append(" prio=")
+           .append(threadInfo.getPriority())
+           .append(" Id=")
+           .append(threadInfo.getThreadId())
+           .append(' ')
+           .append(threadInfo.getThreadState());
+
+    if (threadInfo.getLockName() != null) {
+      builder.append(" on ").append(threadInfo.getLockName());
+    }
+    if (threadInfo.getLockOwnerName() != null) {
+      builder.append(" owned by \"")
+             .append(threadInfo.getLockOwnerName())
+             .append("\" Id=")
+             .append(threadInfo.getLockOwnerId());
+    }
+    if (threadInfo.isSuspended()) {
+      builder.append(" (suspended)");
+    }
+    if (threadInfo.isInNative()) {
+      builder.append(" (in native)");
+    }
+    builder.append('\n');
+
+    final StackTraceElement[] stackTrace = threadInfo.getStackTrace();
+    final MonitorInfo[] lockedMonitors = threadInfo.getLockedMonitors();
+    for (int i = 0; i < stackTrace.length; i++) {
+      builder.append("\tat ").append(stackTrace[i]);
+
+      if (i == 0) {
+        final LockInfo lockInfo = threadInfo.getLockInfo();
+        if (lockInfo != null) {
+          switch (threadInfo.getThreadState()) {
+            case BLOCKED:
+              builder.append(" - blocked on ").append(lockInfo);
+              break;
+            case WAITING:
+            case TIMED_WAITING:
+              builder.append(" - waiting on ").append(lockInfo);
+              break;
+            default:
+              break;
+          }
+        }
+      }
+
+      builder.append('\n');
+
+      for (final MonitorInfo monitorInfo : lockedMonitors) {
+        if (monitorInfo.getLockedStackDepth() == i) {
+          builder.append("\t-  locked ").append(monitorInfo).append('\n');
+        }
+      }
+    }
+
+    final LockInfo[] lockedSynchronizers = threadInfo.getLockedSynchronizers();
+    if (lockedSynchronizers.length > 0) {
+      builder.append("\n\tNumber of locked synchronizers = ")
+             .append(lockedSynchronizers.length)
+             .append('\n');
+      for (final LockInfo lockedSynchronizer : lockedSynchronizers) {
+        builder.append("\t- ").append(lockedSynchronizer).append('\n');
+      }
+    }
+
+    return builder.append('\n').toString();
+  }
+
+  private static boolean isCpuTimeEnabled(final ThreadMXBean threadMxBean)
+  {
+    try {
+      return threadMxBean.isThreadCpuTimeSupported() && threadMxBean.isThreadCpuTimeEnabled();
+    }
+    catch (UnsupportedOperationException | SecurityException e) {
+      return false;
+    }
+  }
+
+  @Nullable
+  private static Long getThreadCpuTime(
+      final ThreadMXBean threadMxBean,
+      final long threadId,
+      final boolean cpuTimeEnabled,
+      final boolean userTime
+  )
+  {
+    if (!cpuTimeEnabled) {
+      return null;
+    }
+
+    try {
+      final long cpuTime = userTime
+                           ? threadMxBean.getThreadUserTime(threadId)
+                           : threadMxBean.getThreadCpuTime(threadId);
+      return cpuTime < 0 ? null : cpuTime;
+    }
+    catch (UnsupportedOperationException | SecurityException e) {
+      return null;
+    }
+  }
+
+  private static Set<Long> findDeadlockedThreadIds(final ThreadMXBean threadMxBean)
+  {
+    try {
+      final long[] threadIds = threadMxBean.findDeadlockedThreads();
+      if (threadIds == null) {
+        return Collections.emptySet();
+      }
+
+      final Set<Long> deadlockedThreadIds = new HashSet<>();
+      for (final long threadId : threadIds) {
+        deadlockedThreadIds.add(threadId);
+      }
+      return deadlockedThreadIds;
+    }
+    catch (UnsupportedOperationException | SecurityException e) {
+      return Collections.emptySet();
+    }
+  }
+
+  @JsonInclude(JsonInclude.Include.NON_NULL)
+  public static class ThreadStackTraceResponse
+  {
+    private final String collectedAt;
+    private final List<ThreadStackTrace> threads;
+
+    @JsonCreator
+    public ThreadStackTraceResponse(
+        @JsonProperty("collectedAt") final String collectedAt,
+        @JsonProperty("threads") final List<ThreadStackTrace> threads
+    )
+    {
+      this.collectedAt = collectedAt;
+      this.threads = threads == null ? ImmutableList.of() : ImmutableList.copyOf(threads);
+    }
+
+    @JsonProperty
+    public String getCollectedAt()
+    {
+      return collectedAt;
+    }
+
+    @JsonProperty
+    public List<ThreadStackTrace> getThreads()
+    {
+      return threads;
+    }
+  }
+
+  @JsonInclude(JsonInclude.Include.NON_NULL)
+  public static class ThreadStackTrace
+  {
+    private final long threadId;
+    private final String threadName;
+    private final String threadState;
+    private final boolean daemon;
+    private final int priority;
+    @Nullable
+    private final Long cpuTimeNs;
+    @Nullable
+    private final Long userCpuTimeNs;
+    @Nullable
+    private final String lockName;
+    @Nullable
+    private final Long lockOwnerId;
+    @Nullable
+    private final String lockOwnerName;
+    private final boolean deadlocked;
+    private final String stackTrace;
+
+    @JsonCreator
+    public ThreadStackTrace(
+        @JsonProperty("threadId") final long threadId,
+        @JsonProperty("threadName") final String threadName,
+        @JsonProperty("threadState") final String threadState,
+        @JsonProperty("daemon") final boolean daemon,
+        @JsonProperty("priority") final int priority,
+        @JsonProperty("cpuTimeNs") @Nullable final Long cpuTimeNs,
+        @JsonProperty("userCpuTimeNs") @Nullable final Long userCpuTimeNs,
+        @JsonProperty("lockName") @Nullable final String lockName,
+        @JsonProperty("lockOwnerId") @Nullable final Long lockOwnerId,
+        @JsonProperty("lockOwnerName") @Nullable final String lockOwnerName,
+        @JsonProperty("deadlocked") final boolean deadlocked,
+        @JsonProperty("stackTrace") final String stackTrace
+    )
+    {
+      this.threadId = threadId;
+      this.threadName = threadName;
+      this.threadState = threadState;
+      this.daemon = daemon;
+      this.priority = priority;
+      this.cpuTimeNs = cpuTimeNs;
+      this.userCpuTimeNs = userCpuTimeNs;
+      this.lockName = lockName;
+      this.lockOwnerId = lockOwnerId;
+      this.lockOwnerName = lockOwnerName;
+      this.deadlocked = deadlocked;
+      this.stackTrace = stackTrace;
+    }
+
+    @JsonProperty
+    public long getThreadId()
+    {
+      return threadId;
+    }
+
+    @JsonProperty
+    public String getThreadName()
+    {
+      return threadName;
+    }
+
+    @JsonProperty
+    public String getThreadState()
+    {
+      return threadState;
+    }
+
+    @JsonProperty
+    public boolean isDaemon()
+    {
+      return daemon;
+    }
+
+    @JsonProperty
+    public int getPriority()
+    {
+      return priority;
+    }
+
+    @Nullable
+    @JsonProperty
+    public Long getCpuTimeNs()
+    {
+      return cpuTimeNs;
+    }
+
+    @Nullable
+    @JsonProperty
+    public Long getUserCpuTimeNs()
+    {
+      return userCpuTimeNs;
+    }
+
+    @Nullable
+    @JsonProperty
+    public String getLockName()
+    {
+      return lockName;
+    }
+
+    @Nullable
+    @JsonProperty
+    public Long getLockOwnerId()
+    {
+      return lockOwnerId;
+    }
+
+    @Nullable
+    @JsonProperty
+    public String getLockOwnerName()
+    {
+      return lockOwnerName;
+    }
+
+    @JsonProperty
+    public boolean isDeadlocked()
+    {
+      return deadlocked;
+    }
+
+    @JsonProperty
+    public String getStackTrace()
+    {
+      return stackTrace;
+    }
+  }
+}
diff --git a/server/src/main/java/org/apache/druid/server/StatusResource.java b/server/src/main/java/org/apache/druid/server/StatusResource.java
index 4c971eae6b77..901b501e3a94 100644
--- a/server/src/main/java/org/apache/druid/server/StatusResource.java
+++ b/server/src/main/java/org/apache/druid/server/StatusResource.java
@@ -28,16 +28,19 @@
 import org.apache.druid.guice.ExtensionsLoader;
 import org.apache.druid.initialization.DruidModule;
 import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.server.http.ServletResourceUtils;
 import org.apache.druid.server.http.security.ConfigResourceFilter;
 import org.apache.druid.server.http.security.StateResourceFilter;
 import org.apache.druid.utils.RuntimeInfo;
 
 import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
 import javax.inject.Inject;
 import javax.servlet.http.HttpServletRequest;
 import javax.ws.rs.GET;
 import javax.ws.rs.Path;
 import javax.ws.rs.Produces;
+import javax.ws.rs.QueryParam;
 import javax.ws.rs.core.Context;
 import javax.ws.rs.core.MediaType;
 import javax.ws.rs.core.Response;
@@ -92,6 +95,27 @@ public Map<String, String> getProperties()
     return new TreeMap<>(filtered);
   }
 
+  /**
+   * Returns a live thread-stack snapshot for this Druid process.
+   *
+   * <p>{@link StateResourceFilter} authorizes this endpoint as a state-read operation before this
+   * method is invoked.
+   */
+  @GET
+  @Path("/stack")
+  @ResourceFilters(StateResourceFilter.class)
+  @Produces(MediaType.APPLICATION_JSON)
+  public Response getStackTrace(
+      @QueryParam(StackTraceCollector.MAX_STACK_TRACE_FRAME_DEPTH_KEY) @Nullable final String maxStackTraceFrameDepth
+  )
+  {
+    return ServletResourceUtils.buildReadResponse(
+        () -> new StackTraceCollector().collect(
+            StackTraceCollector.parseMaxStackTraceFrameDepth(maxStackTraceFrameDepth)
+        )
+    );
+  }
+
   /**
    * filter out entries from allProperties with key containing elements in hiddenProperties (case insensitive)
    *
diff --git a/server/src/test/java/org/apache/druid/server/StatusResourceTest.java b/server/src/test/java/org/apache/druid/server/StatusResourceTest.java
index c677d48ef283..81bfb41df70b 100644
--- a/server/src/test/java/org/apache/druid/server/StatusResourceTest.java
+++ b/server/src/test/java/org/apache/druid/server/StatusResourceTest.java
@@ -23,6 +23,8 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.collect.ImmutableList;
 import com.google.inject.Injector;
+import org.apache.druid.error.DruidException;
+import org.apache.druid.error.ErrorResponse;
 import org.apache.druid.guice.PropertiesModule;
 import org.apache.druid.guice.StartupInjectorBuilder;
 import org.apache.druid.guice.TestDruidModule;
@@ -31,15 +33,21 @@
 import org.apache.druid.segment.loading.SegmentLoaderConfig;
 import org.apache.druid.utils.JvmUtils;
 import org.junit.Assert;
+import org.junit.Assume;
 import org.junit.Test;
 
 import javax.ws.rs.core.Response;
+import java.lang.management.ManagementFactory;
+import java.lang.management.ThreadMXBean;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
 import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.ReentrantLock;
 import java.util.stream.Collectors;
 
 public class StatusResourceTest
@@ -101,6 +109,175 @@ public void testGetReadyReturns503WhenNotReady()
     Assert.assertEquals(false, response.getEntity());
   }
 
+  @Test
+  public void testStackTrace()
+  {
+    final StatusResource resource = new StatusResource(new Properties(), null, null, null, null);
+    final Response httpResponse = resource.getStackTrace(null);
+    Assert.assertEquals(Response.Status.OK.getStatusCode(), httpResponse.getStatus());
+    final StackTraceCollector.ThreadStackTraceResponse response =
+        (StackTraceCollector.ThreadStackTraceResponse) httpResponse.getEntity();
+
+    Assert.assertNotNull(response.getCollectedAt());
+    Assert.assertFalse(response.getThreads().isEmpty());
+
+    final long currentThreadId = Thread.currentThread().threadId();
+    final StackTraceCollector.ThreadStackTrace currentThread = response.getThreads()
+        .stream()
+        .filter(thread -> thread.getThreadId() == currentThreadId)
+        .findFirst()
+        .orElse(null);
+    Assert.assertNotNull(currentThread);
+    Assert.assertEquals(Thread.currentThread().getName(), currentThread.getThreadName());
+    Assert.assertEquals(Thread.currentThread().getState().name(), currentThread.getThreadState());
+    Assert.assertFalse(currentThread.getStackTrace().isEmpty());
+    Assert.assertTrue(currentThread.getStackTrace().contains("\n\tat "));
+    Assert.assertTrue(
+        currentThread.getStackTrace().lines().filter(line -> line.startsWith("\tat ")).count() > 8
+    );
+    Assert.assertFalse(currentThread.getStackTrace().contains("\t...\n"));
+    if (JvmUtils.isThreadCpuTimeEnabled()) {
+      Assert.assertNotNull(currentThread.getCpuTimeNs());
+      Assert.assertNotNull(currentThread.getUserCpuTimeNs());
+    }
+  }
+
+  @Test
+  public void testStackTraceWithMaxStackTraceFrameDepth()
+  {
+    final StatusResource resource = new StatusResource(new Properties(), null, null, null, null);
+    final Response httpResponse = resource.getStackTrace("10");
+    Assert.assertEquals(Response.Status.OK.getStatusCode(), httpResponse.getStatus());
+    final StackTraceCollector.ThreadStackTraceResponse response =
+        (StackTraceCollector.ThreadStackTraceResponse) httpResponse.getEntity();
+
+    Assert.assertTrue(
+        response.getThreads()
+                .stream()
+                .allMatch(
+                    thread -> thread.getStackTrace()
+                                   .lines()
+                                   .filter(line -> line.startsWith("\tat "))
+                                   .count() <= 10
+                )
+    );
+  }
+
+  @Test
+  public void testStackTraceRejectsInvalidMaxStackTraceFrameDepth()
+  {
+    final StatusResource resource = new StatusResource(new Properties(), null, null, null, null);
+
+    for (final String invalidDepth : ImmutableList.of("-1", "0", "9", "1001", "10.5", "not-an-integer")) {
+      final Response response = resource.getStackTrace(invalidDepth);
+      Assert.assertEquals(Response.Status.BAD_REQUEST.getStatusCode(), response.getStatus());
+      Assert.assertTrue(response.getEntity() instanceof ErrorResponse);
+      final DruidException exception = ((ErrorResponse) response.getEntity()).getUnderlyingException();
+      Assert.assertEquals(DruidException.Category.INVALID_INPUT, exception.getCategory());
+      Assert.assertTrue(exception.getMessage().contains(StackTraceCollector.MAX_STACK_TRACE_FRAME_DEPTH_KEY));
+    }
+  }
+
+  @Test
+  public void testStackTraceFormatsWaitingLockOnStackFrame() throws Exception
+  {
+    final Object monitor = new Object();
+    final CountDownLatch enteredMonitor = new CountDownLatch(1);
+    final Thread waitingThread = new Thread(
+        () -> {
+          synchronized (monitor) {
+            enteredMonitor.countDown();
+            try {
+              monitor.wait();
+            }
+            catch (InterruptedException e) {
+              Thread.currentThread().interrupt();
+            }
+          }
+        },
+        "stack-trace-waiting-thread"
+    );
+    waitingThread.start();
+
+    try {
+      Assert.assertTrue(enteredMonitor.await(5, TimeUnit.SECONDS));
+      final long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
+      while (waitingThread.getState() != Thread.State.WAITING && System.nanoTime() < deadline) {
+        Thread.yield();
+      }
+
+      final StackTraceCollector.ThreadStackTrace thread = new StackTraceCollector().collect()
+          .getThreads()
+          .stream()
+          .filter(stackTrace -> stackTrace.getThreadId() == waitingThread.threadId())
+          .findFirst()
+          .orElse(null);
+      Assert.assertNotNull(thread);
+      Assert.assertEquals(Thread.State.WAITING.name(), thread.getThreadState());
+      Assert.assertTrue(
+          thread.getStackTrace().contains(" - waiting on " + thread.getLockName() + "\n")
+      );
+      Assert.assertFalse(thread.getStackTrace().contains("\n\t-  waiting on "));
+    }
+    finally {
+      waitingThread.interrupt();
+      waitingThread.join(TimeUnit.SECONDS.toMillis(5));
+    }
+  }
+
+  @Test
+  public void testStackTraceFormatsHeldMonitorAndSynchronizer() throws Exception
+  {
+    final ThreadMXBean threadMxBean = ManagementFactory.getThreadMXBean();
+    Assume.assumeTrue(threadMxBean.isObjectMonitorUsageSupported());
+    Assume.assumeTrue(threadMxBean.isSynchronizerUsageSupported());
+
+    final Object monitor = new Object();
+    final ReentrantLock synchronizer = new ReentrantLock();
+    final CountDownLatch locksHeld = new CountDownLatch(1);
+    final CountDownLatch releaseLocks = new CountDownLatch(1);
+    final Thread lockHolder = new Thread(
+        () -> {
+          synchronizer.lock();
+          try {
+            synchronized (monitor) {
+              locksHeld.countDown();
+              try {
+                releaseLocks.await();
+              }
+              catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+              }
+            }
+          }
+          finally {
+            synchronizer.unlock();
+          }
+        },
+        "stack-trace-lock-holder"
+    );
+    lockHolder.start();
+
+    try {
+      Assert.assertTrue(locksHeld.await(5, TimeUnit.SECONDS));
+      final StackTraceCollector.ThreadStackTrace thread = new StackTraceCollector().collect()
+          .getThreads()
+          .stream()
+          .filter(stackTrace -> stackTrace.getThreadId() == lockHolder.threadId())
+          .findFirst()
+          .orElse(null);
+      Assert.assertNotNull(thread);
+      Assert.assertTrue(thread.getStackTrace().contains("\t-  locked " + monitor));
+      Assert.assertTrue(thread.getStackTrace().contains("\tNumber of locked synchronizers = 1\n"));
+      Assert.assertTrue(thread.getStackTrace().contains("java.util.concurrent.locks.ReentrantLock$"));
+    }
+    finally {
+      releaseLocks.countDown();
+      lockHolder.interrupt();
+      lockHolder.join(TimeUnit.SECONDS.toMillis(5));
+    }
+  }
+
   private void testHiddenPropertiesWithPropertyFileName(String fileName) throws Exception
   {
     Injector injector = new StartupInjectorBuilder()
diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/planner/PlannerContext.java b/sql/src/main/java/org/apache/druid/sql/calcite/planner/PlannerContext.java
index 859d1d5715c4..12e84d373eb2 100644
--- a/sql/src/main/java/org/apache/druid/sql/calcite/planner/PlannerContext.java
+++ b/sql/src/main/java/org/apache/druid/sql/calcite/planner/PlannerContext.java
@@ -48,6 +48,7 @@
 import org.apache.druid.query.lookup.LookupExtractorFactoryContainerProvider;
 import org.apache.druid.query.lookup.RegisteredLookupExtractionFn;
 import org.apache.druid.segment.join.JoinableFactoryWrapper;
+import org.apache.druid.server.StackTraceCollector;
 import org.apache.druid.server.lookup.cache.LookupLoadingSpec;
 import org.apache.druid.server.security.AuthenticationResult;
 import org.apache.druid.server.security.AuthorizationResult;
@@ -500,6 +501,12 @@ DataContext.Variable.LOCAL_TIMESTAMP.camelName, new Interval(
         if (authenticationResult != null) {
           builder.put(DATA_CTX_AUTHENTICATION_RESULT, authenticationResult);
         }
+        // Query contexts are not copied wholesale into the table-scan DataContext. Propagate
+        // this execution parameter explicitly because sys.stack_trace reads it during scanning.
+        final Object maxStackTraceFrameDepth = queryContext().get(StackTraceCollector.MAX_STACK_TRACE_FRAME_DEPTH_KEY);
+        if (maxStackTraceFrameDepth != null) {
+          builder.put(StackTraceCollector.MAX_STACK_TRACE_FRAME_DEPTH_KEY, maxStackTraceFrameDepth);
+        }
         context = builder.build();
       }
 
diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/schema/SystemSchema.java b/sql/src/main/java/org/apache/druid/sql/calcite/schema/SystemSchema.java
index 56d6bb593d3a..46821bc2b65c 100644
--- a/sql/src/main/java/org/apache/druid/sql/calcite/schema/SystemSchema.java
+++ b/sql/src/main/java/org/apache/druid/sql/calcite/schema/SystemSchema.java
@@ -294,6 +294,10 @@ public SystemSchema(
         SystemServerPropertiesTable.TABLE_NAME,
         new SystemServerPropertiesTable(druidNodeDiscoveryProvider, authorizerMapper, httpClient, jsonMapper)
     );
+    builder.put(
+        SystemStackTraceTable.TABLE_NAME,
+        new SystemStackTraceTable(druidNodeDiscoveryProvider, authorizerMapper, httpClient, jsonMapper)
+    );
 
     if (plannerConfig.isEnableSysQueriesTable()) {
       builder.put(QUERIES_TABLE, new QueriesTable(sqlEngineRegistryProvider, jsonMapper, authorizerMapper));
diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/schema/SystemStackTraceTable.java b/sql/src/main/java/org/apache/druid/sql/calcite/schema/SystemStackTraceTable.java
new file mode 100644
index 000000000000..37cff94f577c
--- /dev/null
+++ b/sql/src/main/java/org/apache/druid/sql/calcite/schema/SystemStackTraceTable.java
@@ -0,0 +1,380 @@
+/*
+ * 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.druid.sql.calcite.schema;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Preconditions;
+import org.apache.calcite.DataContext;
+import org.apache.calcite.linq4j.Enumerable;
+import org.apache.calcite.linq4j.Linq4j;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeFactory;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.schema.ProjectableFilterableTable;
+import org.apache.calcite.schema.Schema;
+import org.apache.calcite.schema.impl.AbstractTable;
+import org.apache.druid.common.guava.FutureUtils;
+import org.apache.druid.discovery.DiscoveryDruidNode;
+import org.apache.druid.discovery.DruidNodeDiscoveryProvider;
+import org.apache.druid.error.InvalidInput;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.logger.Logger;
+import org.apache.druid.java.util.http.client.HttpClient;
+import org.apache.druid.java.util.http.client.Request;
+import org.apache.druid.java.util.http.client.response.StringFullResponseHandler;
+import org.apache.druid.java.util.http.client.response.StringFullResponseHolder;
+import org.apache.druid.query.QueryContexts;
+import org.apache.druid.segment.column.ColumnType;
+import org.apache.druid.segment.column.RowSignature;
+import org.apache.druid.server.DruidNode;
+import org.apache.druid.server.StackTraceCollector;
+import org.apache.druid.server.security.AuthenticationResult;
+import org.apache.druid.server.security.AuthorizerMapper;
+import org.apache.druid.sql.calcite.planner.PlannerContext;
+import org.apache.druid.sql.calcite.table.RowSignatures;
+import org.jboss.netty.handler.codec.http.HttpMethod;
+
+import javax.annotation.Nullable;
+import javax.servlet.http.HttpServletResponse;
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * System schema table {@code sys.stack_trace} that contains a live Java thread-stack snapshot for
+ * explicitly selected Druid servers.
+ */
+public class SystemStackTraceTable extends AbstractTable implements ProjectableFilterableTable
+{
+  private static final Logger log = new Logger(SystemStackTraceTable.class);
+
+  public static final String TABLE_NAME = "stack_trace";
+
+  static final RowSignature ROW_SIGNATURE = RowSignature
+      .builder()
+      .add("server", ColumnType.STRING)
+      .add("service_name", ColumnType.STRING)
+      .add("node_roles", ColumnType.STRING)
+      .add("collected_at", ColumnType.STRING)
+      .add("thread_id", ColumnType.LONG)
+      .add("thread_name", ColumnType.STRING)
+      .add("thread_state", ColumnType.STRING)
+      .add("daemon", ColumnType.LONG)
+      .add("priority", ColumnType.LONG)
+      .add("cpu_time_ns", ColumnType.LONG)
+      .add("user_cpu_time_ns", ColumnType.LONG)
+      .add("lock_name", ColumnType.STRING)
+      .add("lock_owner_id", ColumnType.LONG)
+      .add("lock_owner_name", ColumnType.STRING)
+      .add("is_deadlocked", ColumnType.LONG)
+      .add("stack", ColumnType.STRING)
+      .add("error_message", ColumnType.STRING)
+      .build();
+
+  private static final int SERVER_INDEX = ROW_SIGNATURE.indexOf("server");
+  private static final int SERVICE_NAME_INDEX = ROW_SIGNATURE.indexOf("service_name");
+  private static final int NODE_ROLES_INDEX = ROW_SIGNATURE.indexOf("node_roles");
+  private static final int COLLECTED_AT_INDEX = ROW_SIGNATURE.indexOf("collected_at");
+  private static final int THREAD_ID_INDEX = ROW_SIGNATURE.indexOf("thread_id");
+  private static final int THREAD_NAME_INDEX = ROW_SIGNATURE.indexOf("thread_name");
+  private static final int THREAD_STATE_INDEX = ROW_SIGNATURE.indexOf("thread_state");
+  private static final int DAEMON_INDEX = ROW_SIGNATURE.indexOf("daemon");
+  private static final int PRIORITY_INDEX = ROW_SIGNATURE.indexOf("priority");
+  private static final int CPU_TIME_NS_INDEX = ROW_SIGNATURE.indexOf("cpu_time_ns");
+  private static final int USER_CPU_TIME_NS_INDEX = ROW_SIGNATURE.indexOf("user_cpu_time_ns");
+  private static final int LOCK_NAME_INDEX = ROW_SIGNATURE.indexOf("lock_name");
+  private static final int LOCK_OWNER_ID_INDEX = ROW_SIGNATURE.indexOf("lock_owner_id");
+  private static final int LOCK_OWNER_NAME_INDEX = ROW_SIGNATURE.indexOf("lock_owner_name");
+  private static final int IS_DEADLOCKED_INDEX = ROW_SIGNATURE.indexOf("is_deadlocked");
+  private static final int STACK_INDEX = ROW_SIGNATURE.indexOf("stack");
+  private static final int ERROR_MESSAGE_INDEX = ROW_SIGNATURE.indexOf("error_message");
+
+  private final DruidNodeDiscoveryProvider druidNodeDiscoveryProvider;
+  private final AuthorizerMapper authorizerMapper;
+  private final HttpClient httpClient;
+  private final ObjectMapper jsonMapper;
+
+  public SystemStackTraceTable(
+      final DruidNodeDiscoveryProvider druidNodeDiscoveryProvider,
+      final AuthorizerMapper authorizerMapper,
+      final HttpClient httpClient,
+      final ObjectMapper jsonMapper
+  )
+  {
+    this.druidNodeDiscoveryProvider = druidNodeDiscoveryProvider;
+    this.authorizerMapper = authorizerMapper;
+    this.httpClient = httpClient;
+    this.jsonMapper = jsonMapper;
+  }
+
+  @Override
+  public RelDataType getRowType(final RelDataTypeFactory typeFactory)
+  {
+    return RowSignatures.toRelDataType(ROW_SIGNATURE, typeFactory);
+  }
+
+  @Override
+  public Schema.TableType getJdbcTableType()
+  {
+    return Schema.TableType.SYSTEM_TABLE;
+  }
+
+  @Override
+  public Enumerable<Object[]> scan(
+      final DataContext root,
+      final List<RexNode> filters,
+      @Nullable final int[] projects
+  )
+  {
+    final AuthenticationResult authenticationResult = (AuthenticationResult) Preconditions.checkNotNull(
+        root.get(PlannerContext.DATA_CTX_AUTHENTICATION_RESULT),
+        "authenticationResult in dataContext"
+    );
+    SystemSchema.checkStateReadAccessForServers(authenticationResult, authorizerMapper);
+
+    final Set<String> serverFilter = SystemSchemaFilters.extractColumnValues(filters, SERVER_INDEX);
+    InvalidInput.conditionalException(
+        serverFilter != null,
+        "sys.stack_trace requires a filter on the server column using '=' or 'IN'"
+    );
+    final int maxStackTraceFrameDepth = getMaxStackTraceFrameDepth(
+        root.get(StackTraceCollector.MAX_STACK_TRACE_FRAME_DEPTH_KEY)
+    );
+
+    final Iterator<DiscoveryDruidNode> druidServers = SystemSchema.getDruidServers(druidNodeDiscoveryProvider);
+    final Map<String, ServerStackTraceTarget> serverToTargetMap = new HashMap<>();
+    druidServers.forEachRemaining(discoveryDruidNode -> {
+      final DruidNode druidNode = discoveryDruidNode.getDruidNode();
+      final String server = druidNode.getHostAndPortToUse();
+      if (!serverFilter.contains(server)) {
+        return;
+      }
+
+      final String nodeRole = discoveryDruidNode.getNodeRole().getJsonName();
+      final ServerStackTraceTarget target = serverToTargetMap.get(server);
+      if (target == null) {
+        serverToTargetMap.put(
+            server,
+            new ServerStackTraceTarget(
+                server,
+                druidNode.getServiceName(),
+                new ArrayList<>(Collections.singletonList(nodeRole)),
+                druidNode
+            )
+        );
+      } else {
+        target.addNodeRole(nodeRole);
+      }
+    });
+
+    final List<Object[]> rows = new ArrayList<>();
+    for (final ServerStackTraceTarget target : serverToTargetMap.values()) {
+      rows.addAll(target.buildRows(this, projects, maxStackTraceFrameDepth));
+    }
+    return Linq4j.asEnumerable(rows);
+  }
+
+  static int getMaxStackTraceFrameDepth(@Nullable final Object value)
+  {
+    return StackTraceCollector.validateMaxStackTraceFrameDepth(
+        QueryContexts.getAsLong(
+            StackTraceCollector.MAX_STACK_TRACE_FRAME_DEPTH_KEY,
+            value,
+            StackTraceCollector.DEFAULT_MAX_STACK_TRACE_FRAME_DEPTH
+        )
+    );
+  }
+
+  private static Object[] projectRow(final Object[] row, @Nullable final int[] projects)
+  {
+    if (projects == null) {
+      return row;
+    }
+    final Object[] projectedRow = new Object[projects.length];
+    for (int i = 0; i < projects.length; i++) {
+      projectedRow[i] = row[projects[i]];
+    }
+    return projectedRow;
+  }
+
+  private StackTraceResult getStackTrace(
+      final DruidNode druidNode,
+      final int maxStackTraceFrameDepth
+  )
+  {
+    final String url = druidNode.getUriToUse().resolve(
+        StringUtils.format(
+            "/status/stack?%s=%d",
+            StackTraceCollector.MAX_STACK_TRACE_FRAME_DEPTH_KEY,
+            maxStackTraceFrameDepth
+        )
+    ).toString();
+    try {
+      final Request request = new Request(HttpMethod.GET, URI.create(url).toURL());
+      final StringFullResponseHolder response = FutureUtils.get(
+          httpClient.go(request, new StringFullResponseHandler(StandardCharsets.UTF_8)),
+          true
+      );
+
+      if (response.getStatus().getCode() != HttpServletResponse.SC_OK) {
+        final String errorMessage = StringUtils.format(
+            "HTTP %d: %s",
+            response.getStatus().getCode(),
+            response.getStatus().getReasonPhrase()
+        );
+        log.warn("Failed to get stack trace from node[%s]: error[%s]", url, errorMessage);
+        return new StackTraceResult(null, errorMessage);
+      }
+
+      final StackTraceCollector.ThreadStackTraceResponse stackTraceResponse =
+          jsonMapper.readValue(response.getContent(), StackTraceCollector.ThreadStackTraceResponse.class);
+      return stackTraceResponse == null
+             ? new StackTraceResult(null, "Empty stack trace response")
+             : new StackTraceResult(stackTraceResponse, null);
+    }
+    catch (InterruptedException e) {
+      Thread.currentThread().interrupt();
+      throw new RuntimeException(StringUtils.format("Interrupted while fetching stack trace from node[%s]", url), e);
+    }
+    catch (Exception e) {
+      final String errorMessage = e.getMessage() != null ? e.getMessage() : e.getClass().getSimpleName();
+      log.warn(e, "Failed to get stack trace from node[%s]", url);
+      return new StackTraceResult(null, errorMessage);
+    }
+  }
+
+  private static class StackTraceResult
+  {
+    @Nullable
+    final StackTraceCollector.ThreadStackTraceResponse response;
+    @Nullable
+    final String error;
+
+    StackTraceResult(
+        @Nullable final StackTraceCollector.ThreadStackTraceResponse response,
+        @Nullable final String error
+    )
+    {
+      this.response = response;
+      this.error = error;
+    }
+  }
+
+  private static class ServerStackTraceTarget
+  {
+    final String server;
+    final String serviceName;
+    final List<String> nodeRoles;
+    final DruidNode druidNode;
+
+    ServerStackTraceTarget(
+        final String server,
+        final String serviceName,
+        final List<String> nodeRoles,
+        final DruidNode druidNode
+    )
+    {
+      this.server = server;
+      this.serviceName = serviceName;
+      this.nodeRoles = nodeRoles;
+      this.druidNode = druidNode;
+    }
+
+    void addNodeRole(final String nodeRole)
+    {
+      if (!nodeRoles.contains(nodeRole)) {
+        nodeRoles.add(nodeRole);
+      }
+    }
+
+    String nodeRolesString()
+    {
+      return nodeRoles.stream().sorted().collect(Collectors.joining(","));
+    }
+
+    List<Object[]> buildRows(
+        final SystemStackTraceTable table,
+        @Nullable final int[] projects,
+        final int maxStackTraceFrameDepth
+    )
+    {
+      final StackTraceResult result = table.getStackTrace(druidNode, maxStackTraceFrameDepth);
+      if (result.error != null || result.response == null) {
+        return Collections.singletonList(table.buildErrorRow(this, result.error, projects));
+      }
+
+      return result.response.getThreads()
+          .stream()
+          .map(thread -> {
+            final Object[] row = table.buildThreadRow(this, result.response, thread);
+            return projectRow(row, projects);
+          })
+          .collect(Collectors.toList());
+    }
+  }
+
+  private Object[] buildThreadRow(
+      final ServerStackTraceTarget target,
+      final StackTraceCollector.ThreadStackTraceResponse response,
+      final StackTraceCollector.ThreadStackTrace thread
+  )
+  {
+    final Object[] row = new Object[ROW_SIGNATURE.size()];
+    row[SERVER_INDEX] = target.server;
+    row[SERVICE_NAME_INDEX] = target.serviceName;
+    row[NODE_ROLES_INDEX] = target.nodeRolesString();
+    row[COLLECTED_AT_INDEX] = response.getCollectedAt();
+    row[THREAD_ID_INDEX] = thread.getThreadId();
+    row[THREAD_NAME_INDEX] = thread.getThreadName();
+    row[THREAD_STATE_INDEX] = thread.getThreadState();
+    row[DAEMON_INDEX] = thread.isDaemon() ? 1L : 0L;
+    row[PRIORITY_INDEX] = (long) thread.getPriority();
+    row[CPU_TIME_NS_INDEX] = thread.getCpuTimeNs();
+    row[USER_CPU_TIME_NS_INDEX] = thread.getUserCpuTimeNs();
+    row[LOCK_NAME_INDEX] = thread.getLockName();
+    row[LOCK_OWNER_ID_INDEX] = thread.getLockOwnerId();
+    row[LOCK_OWNER_NAME_INDEX] = thread.getLockOwnerName();
+    row[IS_DEADLOCKED_INDEX] = thread.isDeadlocked() ? 1L : 0L;
+    row[STACK_INDEX] = thread.getStackTrace();
+    row[ERROR_MESSAGE_INDEX] = null;
+    return row;
+  }
+
+  private Object[] buildErrorRow(
+      final ServerStackTraceTarget target,
+      @Nullable final String errorMessage,
+      @Nullable final int[] projects
+  )
+  {
+    final Object[] row = new Object[ROW_SIGNATURE.size()];
+    row[SERVER_INDEX] = target.server;
+    row[SERVICE_NAME_INDEX] = target.serviceName;
+    row[NODE_ROLES_INDEX] = target.nodeRolesString();
+    row[ERROR_MESSAGE_INDEX] = errorMessage;
+    return projectRow(row, projects);
+  }
+}
diff --git a/sql/src/test/java/org/apache/druid/sql/calcite/CalciteQueryTest.java b/sql/src/test/java/org/apache/druid/sql/calcite/CalciteQueryTest.java
index 808b06e94119..4c2967c2981a 100644
--- a/sql/src/test/java/org/apache/druid/sql/calcite/CalciteQueryTest.java
+++ b/sql/src/test/java/org/apache/druid/sql/calcite/CalciteQueryTest.java
@@ -211,6 +211,7 @@ public void testInformationSchemaTables()
                      .add(new Object[]{"sys", "server_properties", "SYSTEM_TABLE", "NO", "NO"})
                      .add(new Object[]{"sys", "server_segments", "SYSTEM_TABLE", "NO", "NO"})
                      .add(new Object[]{"sys", "servers", "SYSTEM_TABLE", "NO", "NO"})
+                     .add(new Object[]{"sys", "stack_trace", "SYSTEM_TABLE", "NO", "NO"})
                      .add(new Object[]{"sys", "supervisors", "SYSTEM_TABLE", "NO", "NO"})
                      .add(new Object[]{"sys", "tasks", "SYSTEM_TABLE", "NO", "NO"})
                      .add(new Object[]{"view", "aview", "VIEW", "NO", "NO"})
@@ -257,6 +258,7 @@ public void testInformationSchemaTables()
                      .add(new Object[]{"sys", "server_properties", "SYSTEM_TABLE", "NO", "NO"})
                      .add(new Object[]{"sys", "server_segments", "SYSTEM_TABLE", "NO", "NO"})
                      .add(new Object[]{"sys", "servers", "SYSTEM_TABLE", "NO", "NO"})
+                     .add(new Object[]{"sys", "stack_trace", "SYSTEM_TABLE", "NO", "NO"})
                      .add(new Object[]{"sys", "supervisors", "SYSTEM_TABLE", "NO", "NO"})
                      .add(new Object[]{"sys", "tasks", "SYSTEM_TABLE", "NO", "NO"})
                      .add(new Object[]{"view", "aview", "VIEW", "NO", "NO"})
diff --git a/sql/src/test/java/org/apache/druid/sql/calcite/schema/SystemSchemaTest.java b/sql/src/test/java/org/apache/druid/sql/calcite/schema/SystemSchemaTest.java
index 94167e669703..32c433ecbe55 100644
--- a/sql/src/test/java/org/apache/druid/sql/calcite/schema/SystemSchemaTest.java
+++ b/sql/src/test/java/org/apache/druid/sql/calcite/schema/SystemSchemaTest.java
@@ -59,6 +59,7 @@
 import org.apache.druid.discovery.DruidNodeDiscovery;
 import org.apache.druid.discovery.DruidNodeDiscoveryProvider;
 import org.apache.druid.discovery.NodeRole;
+import org.apache.druid.error.DruidException;
 import org.apache.druid.indexer.TaskStatusPlus;
 import org.apache.druid.indexer.granularity.GranularitySpec;
 import org.apache.druid.indexer.partitions.DynamicPartitionsSpec;
@@ -76,6 +77,7 @@
 import org.apache.druid.java.util.http.client.response.InputStreamFullResponseHolder;
 import org.apache.druid.java.util.http.client.response.StringFullResponseHandler;
 import org.apache.druid.java.util.http.client.response.StringFullResponseHolder;
+import org.apache.druid.query.BadQueryContextException;
 import org.apache.druid.query.QueryRunnerFactoryConglomerate;
 import org.apache.druid.query.aggregation.CountAggregatorFactory;
 import org.apache.druid.query.aggregation.DoubleSumAggregatorFactory;
@@ -95,6 +97,7 @@
 import org.apache.druid.server.QueryStackTests;
 import org.apache.druid.server.SegmentManager;
 import org.apache.druid.server.SpecificSegmentsQuerySegmentWalker;
+import org.apache.druid.server.StackTraceCollector;
 import org.apache.druid.server.coordination.DruidServerMetadata;
 import org.apache.druid.server.coordination.ServerType;
 import org.apache.druid.server.coordinator.BytesAccumulatingResponseHandler;
@@ -571,13 +574,29 @@ DataNodeService.DISCOVERY_SERVICE_KEY, new DataNodeService("tier", 1000, null, S
   public void testGetTableMap()
   {
     Assert.assertEquals(
-        ImmutableSet.of("segments", "servers", "server_segments", "tasks", "supervisors", "server_properties"),
+        ImmutableSet.of(
+            "segments",
+            "servers",
+            "server_segments",
+            "tasks",
+            "supervisors",
+            "server_properties",
+            "stack_trace"
+        ),
         schema.getTableNames()
     );
 
     final Map<String, Table> tableMap = schema.getTableMap();
     Assert.assertEquals(
-        ImmutableSet.of("segments", "servers", "server_segments", "tasks", "supervisors", "server_properties"),
+        ImmutableSet.of(
+            "segments",
+            "servers",
+            "server_segments",
+            "tasks",
+            "supervisors",
+            "server_properties",
+            "stack_trace"
+        ),
         tableMap.keySet()
     );
     final SystemSchema.SegmentsTable segmentsTable = (SystemSchema.SegmentsTable) schema.getTableMap().get("segments");
@@ -606,6 +625,14 @@ public void testGetTableMap()
     final RelDataType propertiesRowType = propertiesTable.getRowType(new JavaTypeFactoryImpl());
     final List<RelDataTypeField> propertiesFields = propertiesRowType.getFieldList();
     Assert.assertEquals(6, propertiesFields.size());
+
+    final SystemStackTraceTable stackTraceTable = (SystemStackTraceTable) schema.getTableMap().get("stack_trace");
+    final RelDataType stackTraceRowType = stackTraceTable.getRowType(new JavaTypeFactoryImpl());
+    final List<RelDataTypeField> stackTraceFields = stackTraceRowType.getFieldList();
+    Assert.assertEquals(17, stackTraceFields.size());
+    Assert.assertEquals(SqlTypeName.VARCHAR, stackTraceFields.get(2).getType().getSqlTypeName());
+    Assert.assertEquals("stack", stackTraceFields.get(15).getName());
+    Assert.assertEquals(SqlTypeName.VARCHAR, stackTraceFields.get(15).getType().getSqlTypeName());
   }
 
   @Test
@@ -2211,6 +2238,323 @@ public void testPropertiesTable_exceptionWithNullMessage()
     EasyMock.verify(druidNodeDiscoveryProvider, httpClient);
   }
 
+  @Test
+  public void testStackTraceTable() throws Exception
+  {
+    final SystemStackTraceTable stackTraceTable = new SystemStackTraceTable(
+        druidNodeDiscoveryProvider,
+        authMapper,
+        httpClient,
+        MAPPER
+    );
+
+    mockAllNodeRolesWithCoordinator(coordinator);
+
+    final StackTraceCollector.ThreadStackTrace thread = new StackTraceCollector.ThreadStackTrace(
+        42L,
+        "test-thread",
+        "RUNNABLE",
+        false,
+        7,
+        100L,
+        80L,
+        null,
+        null,
+        null,
+        false,
+        "\"test-thread\" Id=42 RUNNABLE\n\tat TestClass.testMethod(TestClass.java:42)\n"
+    );
+    final StackTraceCollector.ThreadStackTraceResponse stackTraceResponse = new StackTraceCollector.ThreadStackTraceResponse(
+        "2026-08-02T05:00:00.000Z",
+        ImmutableList.of(thread)
+    );
+    final HttpResponse httpResponse = new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK);
+    final StringFullResponseHolder responseHolder = new StringFullResponseHolder(httpResponse, StandardCharsets.UTF_8);
+    responseHolder.addChunk(MAPPER.writeValueAsString(stackTraceResponse));
+
+    EasyMock.expect(
+        httpClient.go(EasyMock.isA(Request.class), EasyMock.isA(StringFullResponseHandler.class))
+    ).andAnswer(() -> {
+      final Request request = (Request) EasyMock.getCurrentArguments()[0];
+      Assert.assertEquals(
+          coordinator.getDruidNode().getUriToUse()
+                    .resolve("/status/stack?maxStackTraceFrameDepth=10")
+                    .toURL(),
+          request.getUrl()
+      );
+      return Futures.immediateFuture(responseHolder);
+    }).once();
+    EasyMock.replay(druidNodeDiscoveryProvider, httpClient);
+
+    final List<Object[]> rows = stackTraceTable.scan(
+        createDataContext(Users.SUPER, 10),
+        ImmutableList.of(
+            createStackTraceServerEquality(stackTraceTable, coordinator.getDruidNode().getHostAndPortToUse())
+        ),
+        null
+    ).toList();
+
+    Assert.assertEquals(1, rows.size());
+    final Object[] row = rows.get(0);
+    Assert.assertEquals(coordinator.getDruidNode().getHostAndPortToUse(), row[0]);
+    Assert.assertEquals(coordinator.getDruidNode().getServiceName(), row[1]);
+    Assert.assertEquals(NodeRole.COORDINATOR.getJsonName(), row[2]);
+    Assert.assertEquals("2026-08-02T05:00:00.000Z", row[3]);
+    Assert.assertEquals(42L, row[4]);
+    Assert.assertEquals("test-thread", row[5]);
+    Assert.assertEquals("RUNNABLE", row[6]);
+    Assert.assertEquals(0L, row[7]);
+    Assert.assertEquals(7L, row[8]);
+    Assert.assertEquals(100L, row[9]);
+    Assert.assertEquals(80L, row[10]);
+    Assert.assertNull(row[11]);
+    Assert.assertNull(row[12]);
+    Assert.assertNull(row[13]);
+    Assert.assertEquals(0L, row[14]);
+    Assert.assertEquals(thread.getStackTrace(), row[15]);
+    Assert.assertNull(row[16]);
+
+    EasyMock.verify(druidNodeDiscoveryProvider, httpClient);
+  }
+
+  @Test
+  public void testStackTraceTable_multiRoleAndProjection() throws Exception
+  {
+    final SystemStackTraceTable stackTraceTable = new SystemStackTraceTable(
+        druidNodeDiscoveryProvider,
+        authMapper,
+        httpClient,
+        MAPPER
+    );
+    final DiscoveryDruidNode coordinatorRole = new DiscoveryDruidNode(
+        coordinator.getDruidNode(),
+        NodeRole.COORDINATOR,
+        ImmutableMap.of(),
+        startTime
+    );
+    final DiscoveryDruidNode overlordRole = new DiscoveryDruidNode(
+        coordinator.getDruidNode(),
+        NodeRole.OVERLORD,
+        ImmutableMap.of(),
+        startTime
+    );
+
+    mockNodeDiscovery(NodeRole.BROKER);
+    mockNodeDiscovery(NodeRole.ROUTER);
+    mockNodeDiscovery(NodeRole.HISTORICAL);
+    mockNodeDiscovery(NodeRole.OVERLORD, overlordRole);
+    mockNodeDiscovery(NodeRole.PEON);
+    mockNodeDiscovery(NodeRole.INDEXER);
+    mockNodeDiscovery(NodeRole.MIDDLE_MANAGER);
+    mockNodeDiscovery(NodeRole.COORDINATOR, coordinatorRole);
+
+    final StackTraceCollector.ThreadStackTrace thread = new StackTraceCollector.ThreadStackTrace(
+        43L,
+        "multi-role-thread",
+        "WAITING",
+        true,
+        5,
+        null,
+        null,
+        "java.lang.Object@1",
+        7L,
+        "owner-thread",
+        true,
+        "\"multi-role-thread\" Id=43 WAITING\n"
+    );
+    final StackTraceCollector.ThreadStackTraceResponse stackTraceResponse = new StackTraceCollector.ThreadStackTraceResponse(
+        "2026-08-02T05:01:00.000Z",
+        ImmutableList.of(thread)
+    );
+    final HttpResponse httpResponse = new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK);
+    final StringFullResponseHolder responseHolder = new StringFullResponseHolder(httpResponse, StandardCharsets.UTF_8);
+    responseHolder.addChunk(MAPPER.writeValueAsString(stackTraceResponse));
+
+    EasyMock.expect(
+        httpClient.go(EasyMock.isA(Request.class), EasyMock.isA(StringFullResponseHandler.class))
+    ).andReturn(Futures.immediateFuture(responseHolder)).once();
+    EasyMock.replay(druidNodeDiscoveryProvider, httpClient);
+
+    final int[] projects = new int[]{0, 2, 3, 4, 15, 16};
+    final List<Object[]> rows = stackTraceTable.scan(
+        createDataContext(Users.SUPER),
+        ImmutableList.of(
+            createStackTraceServerEquality(stackTraceTable, coordinator.getDruidNode().getHostAndPortToUse())
+        ),
+        projects
+    ).toList();
+
+    Assert.assertEquals(1, rows.size());
+    Assert.assertEquals(6, rows.get(0).length);
+    Assert.assertEquals(coordinator.getDruidNode().getHostAndPortToUse(), rows.get(0)[0]);
+    Assert.assertEquals("coordinator,overlord", rows.get(0)[1]);
+    Assert.assertEquals("2026-08-02T05:01:00.000Z", rows.get(0)[2]);
+    Assert.assertEquals(43L, rows.get(0)[3]);
+    Assert.assertEquals(thread.getStackTrace(), rows.get(0)[4]);
+    Assert.assertNull(rows.get(0)[5]);
+
+    EasyMock.verify(druidNodeDiscoveryProvider, httpClient);
+  }
+
+  @Test
+  public void testStackTraceTable_httpError()
+  {
+    final SystemStackTraceTable stackTraceTable = new SystemStackTraceTable(
+        druidNodeDiscoveryProvider,
+        authMapper,
+        httpClient,
+        MAPPER
+    );
+    mockAllNodeRolesWithCoordinator(coordinator);
+
+    final HttpResponse httpResponse = new DefaultHttpResponse(
+        HttpVersion.HTTP_1_1,
+        HttpResponseStatus.SERVICE_UNAVAILABLE
+    );
+    final StringFullResponseHolder responseHolder = new StringFullResponseHolder(httpResponse, StandardCharsets.UTF_8);
+    responseHolder.addChunk("service unavailable");
+    EasyMock.expect(
+        httpClient.go(EasyMock.isA(Request.class), EasyMock.isA(StringFullResponseHandler.class))
+    ).andReturn(Futures.immediateFuture(responseHolder)).once();
+    EasyMock.replay(druidNodeDiscoveryProvider, httpClient);
+
+    final List<Object[]> rows = stackTraceTable.scan(
+        createDataContext(Users.SUPER),
+        ImmutableList.of(
+            createStackTraceServerEquality(stackTraceTable, coordinator.getDruidNode().getHostAndPortToUse())
+        ),
+        null
+    ).toList();
+
+    Assert.assertEquals(1, rows.size());
+    Assert.assertEquals(coordinator.getDruidNode().getHostAndPortToUse(), rows.get(0)[0]);
+    Assert.assertNull(rows.get(0)[4]);
+    Assert.assertTrue(((String) rows.get(0)[16]).contains("HTTP 503"));
+
+    EasyMock.verify(druidNodeDiscoveryProvider, httpClient);
+  }
+
+  @Test
+  public void testStackTraceTable_withInterruptedException()
+  {
+    final SystemStackTraceTable stackTraceTable = new SystemStackTraceTable(
+        druidNodeDiscoveryProvider,
+        authMapper,
+        httpClient,
+        MAPPER
+    );
+
+    mockAllNodeRolesWithCoordinator(coordinator);
+
+    final SettableFuture<StringFullResponseHolder> interruptingFuture = SettableFuture.create();
+    EasyMock.expect(
+        httpClient.go(EasyMock.isA(Request.class), EasyMock.isA(StringFullResponseHandler.class))
+    ).andReturn(interruptingFuture).once();
+    EasyMock.replay(druidNodeDiscoveryProvider, httpClient);
+
+    try {
+      Thread.currentThread().interrupt();
+      final RuntimeException exception = Assert.assertThrows(
+          RuntimeException.class,
+          () -> stackTraceTable.scan(
+              createDataContext(Users.SUPER),
+              ImmutableList.of(
+                  createStackTraceServerEquality(stackTraceTable, coordinator.getDruidNode().getHostAndPortToUse())
+              ),
+              null
+          ).toList()
+      );
+      Assert.assertTrue(exception.getMessage().contains("Interrupted"));
+      Assert.assertTrue(interruptingFuture.isCancelled());
+      Assert.assertTrue(Thread.currentThread().isInterrupted());
+    }
+    finally {
+      Thread.interrupted();
+    }
+
+    EasyMock.verify(druidNodeDiscoveryProvider, httpClient);
+  }
+
+  @Test
+  public void testStackTraceTable_requiresServerFilter()
+  {
+    final SystemStackTraceTable stackTraceTable = new SystemStackTraceTable(
+        druidNodeDiscoveryProvider,
+        authMapper,
+        httpClient,
+        MAPPER
+    );
+
+    final DruidException exception = Assert.assertThrows(
+        DruidException.class,
+        () -> stackTraceTable.scan(createDataContext(Users.SUPER), Collections.emptyList(), null).toList()
+    );
+    Assert.assertEquals(DruidException.Category.INVALID_INPUT, exception.getCategory());
+    Assert.assertTrue(exception.getMessage().contains("requires a filter on the server column"));
+
+    final RexBuilder rexBuilder = new RexBuilder(new JavaTypeFactoryImpl());
+    final int serverIndex = SystemStackTraceTable.ROW_SIGNATURE.indexOf("server");
+    final RelDataType rowType = stackTraceTable.getRowType(new JavaTypeFactoryImpl());
+    final RexNode serverRef = rexBuilder.makeInputRef(
+        rowType.getFieldList().get(serverIndex).getType(),
+        serverIndex
+    );
+    final RexNode notEquals = rexBuilder.makeCall(
+        SqlStdOperatorTable.NOT_EQUALS,
+        serverRef,
+        rexBuilder.makeLiteral(coordinator.getDruidNode().getHostAndPortToUse())
+    );
+    final DruidException notEqualsException = Assert.assertThrows(
+        DruidException.class,
+        () -> stackTraceTable.scan(createDataContext(Users.SUPER), ImmutableList.of(notEquals), null).toList()
+    );
+    Assert.assertEquals(DruidException.Category.INVALID_INPUT, notEqualsException.getCategory());
+  }
+
+  @Test
+  public void testStackTraceTable_rejectsInvalidMaxStackTraceFrameDepth()
+  {
+    final SystemStackTraceTable stackTraceTable = new SystemStackTraceTable(
+        druidNodeDiscoveryProvider,
+        authMapper,
+        httpClient,
+        MAPPER
+    );
+
+    for (final long invalidDepth : new long[]{9, StackTraceCollector.MAX_ALLOWED_STACK_TRACE_FRAME_DEPTH + 1L}) {
+      final DruidException exception = Assert.assertThrows(
+          DruidException.class,
+          () -> stackTraceTable.scan(
+              createDataContext(Users.SUPER, invalidDepth),
+              ImmutableList.of(
+                  createStackTraceServerEquality(stackTraceTable, coordinator.getDruidNode().getHostAndPortToUse())
+              ),
+              null
+          ).toList()
+      );
+      Assert.assertEquals(DruidException.Category.INVALID_INPUT, exception.getCategory());
+      Assert.assertTrue(exception.getMessage().contains(StackTraceCollector.MAX_STACK_TRACE_FRAME_DEPTH_KEY));
+    }
+  }
+
+  @Test
+  public void testStackTraceTable_maxStackTraceFrameDepthContextConversion()
+  {
+    Assert.assertEquals(100, SystemStackTraceTable.getMaxStackTraceFrameDepth(null));
+    Assert.assertEquals(10, SystemStackTraceTable.getMaxStackTraceFrameDepth(10.9));
+    Assert.assertEquals(10, SystemStackTraceTable.getMaxStackTraceFrameDepth("10"));
+    Assert.assertEquals(10, SystemStackTraceTable.getMaxStackTraceFrameDepth("10.0"));
+
+    Assert.assertThrows(
+        BadQueryContextException.class,
+        () -> SystemStackTraceTable.getMaxStackTraceFrameDepth("10.9")
+    );
+    Assert.assertThrows(
+        DruidException.class,
+        () -> SystemStackTraceTable.getMaxStackTraceFrameDepth(9.9)
+    );
+  }
+
   @Test
   public void testQueriesTable()
   {
@@ -2380,6 +2724,21 @@ private String getStatusPropertiesUrl(DiscoveryDruidNode discoveryDruidNode)
     return discoveryDruidNode.getDruidNode().getUriToUse().resolve("/status/properties").toString();
   }
 
+  private RexNode createStackTraceServerEquality(
+      final SystemStackTraceTable stackTraceTable,
+      final String server
+  )
+  {
+    final RexBuilder rexBuilder = new RexBuilder(new JavaTypeFactoryImpl());
+    final int serverIndex = SystemStackTraceTable.ROW_SIGNATURE.indexOf("server");
+    final RelDataType rowType = stackTraceTable.getRowType(new JavaTypeFactoryImpl());
+    return rexBuilder.makeCall(
+        SqlStdOperatorTable.EQUALS,
+        rexBuilder.makeInputRef(rowType.getFieldList().get(serverIndex).getType(), serverIndex),
+        rexBuilder.makeLiteral(server)
+    );
+  }
+
   /**
    * Creates a response holder that contains the given json.
    */
@@ -2400,7 +2759,12 @@ private InputStreamFullResponseHolder createFullResponseHolder(
   /**
    * Creates a DataContext for the given username.
    */
-  private DataContext createDataContext(String username)
+  private DataContext createDataContext(final String username)
+  {
+    return createDataContext(username, null);
+  }
+
+  private DataContext createDataContext(final String username, final Object maxStackTraceFrameDepth)
   {
     return new DataContext()
     {
@@ -2423,8 +2787,11 @@ public QueryProvider getQueryProvider()
       }
 
       @Override
-      public Object get(String authorizerName)
+      public Object get(final String authorizerName)
       {
+        if (StackTraceCollector.MAX_STACK_TRACE_FRAME_DEPTH_KEY.equals(authorizerName)) {
+          return maxStackTraceFrameDepth;
+        }
         return CalciteTests.TEST_SUPERUSER_NAME.equals(username)
                ? CalciteTests.SUPER_USER_AUTH_RESULT
                : new AuthenticationResult(username, authorizerName, null, null);
diff --git a/website/.spelling b/website/.spelling
index a74372f7dc1c..aac12d42326b 100644
--- a/website/.spelling
+++ b/website/.spelling
@@ -56,11 +56,13 @@ blocklist
 bottlenecked
 build_revision
 cartesian
+collected_at
 concat
 CIDR
 CORS
 CNF
 CPUs
+cpu_time_ns
 CSVs
 CTEs
 CentralizedDatasourceSchema
@@ -163,6 +165,7 @@ JDK8
 JKS
 jks
 JMX
+jstack
 JRE
 JS
 JSON
@@ -425,6 +428,9 @@ laning
 lifecycle
 lineage
 localhost
+lock_name
+lock_owner_id
+lock_owner_name
 log4j
 log4j2
 log4j2.xml
@@ -573,6 +579,7 @@ smoosh
 smooshed
 snapshotting
 splittable
+stack_trace
 ssl
 sslmode
 start_time
@@ -599,6 +606,9 @@ syncs
 syntaxes
 systemFields
 tablePath
+thread_id
+thread_name
+thread_state
 tiering
 timeseries
 Timeseries
@@ -619,6 +629,7 @@ uncomment
 uncompacted
 underutilization
 unintuitive
+user_cpu_time_ns
 unioned
 unmergeable
 unmerged
@@ -683,6 +694,7 @@ DEFAULT_CHARACTER_SET_NAME
 DEFAULT_CHARACTER_SET_SCHEMA
 ISODOW
 ISOYEAR
+is_deadlocked
 IS_NULLABLE
 JDBC_TYPE
 MIDDLE_MANAGER