feat: add sys.stack_trace system table to dump druid nodes threads

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.

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

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);
+    }
+  }
+}