refactor: Enhance ThreadDumpUtils with new features and tests

- Made the ThreadSortBy object public for better accessibility.
- Added a helper method to find specific thread sections in the dump for targeted assertions.
- Implemented tests to verify stack depth limitation, thread sorting by ID, and deadlock detection in thread dumps.
- Removed outdated test for stack depth configuration to streamline the test suite.

Author: wangzhigang1999

kyuubi-common/src/main/scala/org/apache/kyuubi/util/ThreadDumpUtils.scala

@@ -59,7 +59,7 @@ object ThreadDumpUtils {
    * Thread sorting options
    */
   sealed trait ThreadSortBy
-  private object ThreadSortBy {
+  object ThreadSortBy {
     case object Id extends ThreadSortBy
     case object Name extends ThreadSortBy
     case object State extends ThreadSortBy

kyuubi-common/src/test/scala/org/apache/kyuubi/util/ThreadDumpUtilsSuite.scala

@@ -18,6 +18,7 @@
 package org.apache.kyuubi.util
 
 import java.io.{ByteArrayOutputStream, PrintStream}
+import java.util.concurrent.{CountDownLatch, TimeUnit}
 
 import org.apache.kyuubi.KyuubiFunSuite
 
@@ -30,12 +31,6 @@ class ThreadDumpUtilsSuite extends KyuubiFunSuite {
     assert(dump.contains("Thread Dump End"))
   }
 
-  test("dumpToString should respect stackDepth config") {
-    val config = ThreadDumpUtils.DumpConfig(stackDepth = 1)
-    val dump = ThreadDumpUtils.dumpToString(config)
-    assert(dump.linesIterator.exists(_.trim.startsWith("at ")))
-  }
-
   test("dumpToConsole should print to provided stream") {
     val baos = new ByteArrayOutputStream()
     val ps = new PrintStream(baos)
@@ -59,4 +54,141 @@ class ThreadDumpUtilsSuite extends KyuubiFunSuite {
     assert(dump.contains("Thread Dump End"))
     assert(!dump.contains("====================== Daemon Threads ======================"))
   }
+
+  /**
+   * Helper to find a specific thread's section in the dump.
+   * This is useful for targeted assertions on a single thread's output.
+   */
+  private def findThreadSection(dump: String, threadName: String): Option[String] = {
+    val pattern = s"""(?s)Thread: "$threadName".*?(?=\\n\\nThread:|\\n\\n================)""".r
+    pattern.findFirstIn(dump)
+  }
+
+  test("dumpToString should correctly limit stack depth and show truncation message") {
+    val readyLatch = new CountDownLatch(1)
+    var deepStackThread: Thread = null
+    val stackDepth = 100
+    val limit = 10
+
+    // Recursive function to create a deep stack
+    def deepStack(n: Int): Unit = {
+      if (n > 0) {
+        deepStack(n - 1)
+      } else {
+        readyLatch.countDown() // Signal that the thread is ready
+        try Thread.sleep(5000) // Keep thread alive for dumping
+        catch { case _: InterruptedException => Thread.currentThread().interrupt() }
+      }
+    }
+
+    try {
+      deepStackThread = new Thread(() => deepStack(stackDepth), "deep-stack-test-thread")
+      deepStackThread.setDaemon(true)
+      deepStackThread.start()
+
+      // Wait for the thread to reach its deepest point
+      assert(readyLatch.await(5, TimeUnit.SECONDS), "Test thread did not initialize in time")
+
+      val config = ThreadDumpUtils.DumpConfig(stackDepth = limit)
+      val dump = ThreadDumpUtils.dumpToString(config)
+      val section = findThreadSection(dump, "deep-stack-test-thread")
+
+      assert(section.isDefined, "Thread 'deep-stack-test-thread' not found in dump")
+
+      // Verify that the stack trace is truncated and the message is correct
+      // Total frames = stackDepth + 1 (for the call to deepStack(0)),
+      // so we expect (101 - 10) more frames
+      val expectedMoreFrames = stackDepth + 1 - limit
+      assert(section.get.contains(s"... (${expectedMoreFrames} more stack frames)"))
+
+      // Verify the number of 'at' lines matches the limit
+      val stackTraceLines = section.get.linesIterator.count(_.trim.startsWith("at "))
+      assert(stackTraceLines == limit)
+
+    } finally {
+      if (deepStackThread != null) deepStackThread.interrupt()
+    }
+  }
+
+  test("dumpToString should sort threads by ID when configured") {
+    val config = ThreadDumpUtils.DumpConfig(sortThreadsBy = ThreadDumpUtils.ThreadSortBy.Id)
+    val dump = ThreadDumpUtils.dumpToString(config)
+
+    // Extract the summary table for easier parsing
+    val summarySection =
+      dump.substring(dump.indexOf("======================== Summary ========================"))
+
+    // Regex to extract thread IDs from the summary table lines
+    val idPattern = """^\s*(\d+)\s*\|.*""".r
+    val ids = summarySection.linesIterator.flatMap { line =>
+      idPattern.findFirstMatchIn(line).map(_.group(1).toLong)
+    }.toList
+
+    assert(ids.nonEmpty, "No thread IDs found in the summary")
+    // Verify that the list of IDs is sorted, which proves the sorting logic worked
+    assert(ids == ids.sorted, s"Thread IDs are not sorted: $ids")
+  }
+
+  test("dumpToString should detect and report deadlocks") {
+    val lock1 = new Object()
+    val lock2 = new Object()
+    // Latch to ensure both threads are in a deadlock state before we take the dump
+    val deadlockSetupLatch = new CountDownLatch(2)
+
+    val thread1 = new Thread(
+      () => {
+        lock1.synchronized {
+          deadlockSetupLatch.countDown()
+          Thread.sleep(200) // Wait for thread2 to acquire lock2
+          lock2.synchronized {
+            // This line will never be reached
+          }
+        }
+      },
+      "kyuubi-deadlock-thread-1")
+
+    val thread2 = new Thread(
+      () => {
+        lock2.synchronized {
+          deadlockSetupLatch.countDown()
+          Thread.sleep(200) // Wait for thread1 to acquire lock1
+          lock1.synchronized {
+            // This line will never be reached
+          }
+        }
+      },
+      "kyuubi-deadlock-thread-2")
+
+    // Use daemon threads so they don't block JVM exit if the test fails
+    thread1.setDaemon(true)
+    thread2.setDaemon(true)
+
+    try {
+      thread1.start()
+      thread2.start()
+
+      // Wait for both threads to acquire their first lock
+      assert(
+        deadlockSetupLatch.await(5, TimeUnit.SECONDS),
+        "Deadlock condition was not met in time")
+
+      // Give the JVM time to officially recognize the deadlock state
+      Thread.sleep(500)
+
+      val dump = ThreadDumpUtils.dumpToString()
+
+      assert(dump.contains("*** DEADLOCK DETECTED ***"))
+      // Check that both threads involved in the deadlock are mentioned
+      assert(dump.contains(""""kyuubi-deadlock-thread-1""""))
+      assert(dump.contains(""""kyuubi-deadlock-thread-2""""))
+      // Check for lock details which are crucial for debugging deadlocks
+      assert(dump.contains("Waiting on:"))
+      assert(dump.contains("Lock owned by"))
+
+    } finally {
+      // Clean up the threads
+      thread1.interrupt()
+      thread2.interrupt()
+    }
+  }
 }