feat(kyuubi-common):增加操作超时监控线程池配置并实现相关功能

- 在 KyuubiConf 中添加 OPERATION_TIMEOUT_MONITOR_POOL_SIZE 配置项
- 在 SessionManager 中使用新配置初始化操作超时监控线程池
- 在 ThreadUtils 中新增 newDaemonScheduledThreadPool 方法
- 在 ThreadUtilsSuite 中添加新方法的测试用例

Author: wangzhigang1999

kyuubi-common/src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala

@@ -336,7 +336,12 @@ object KyuubiConf {
     .version("1.0.0")
     .timeConf
     .createWithDefault(Duration.ofHours(3).toMillis)
-
+  val OPERATION_TIMEOUT_MONITOR_POOL_SIZE: ConfigEntry[Int] =
+    buildConf("kyuubi.operation.timeout.monitor.pool.size")
+      .doc("Number of threads in the operation timeout monitor thread pool")
+      .version("1.8.0")
+      .intConf
+      .createWithDefault(10)
   val CREDENTIALS_RENEWAL_INTERVAL: ConfigEntry[Long] =
     buildConf("kyuubi.credentials.renewal.interval")
       .doc("How often Kyuubi renews one user's delegation tokens")

kyuubi-common/src/main/scala/org/apache/kyuubi/session/SessionManager.scala

@@ -288,7 +288,8 @@ abstract class SessionManager(name: String) extends CompositeService(name) {
       s"$name-exec-pool")
 
     // Initialize timeout monitor pool
-    operationTimeoutMonitor = ThreadUtils.newDaemonSingleThreadScheduledExecutor(
+    operationTimeoutMonitor = ThreadUtils.newDaemonScheduledThreadPool(
+      conf.get(OPERATION_TIMEOUT_MONITOR_POOL_SIZE),
       s"$name-operation-timeout-monitor-pool")
 
     super.initialize(conf)

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

@@ -37,6 +37,15 @@ object ThreadUtils extends Logging {
     executor
   }
 
+  def newDaemonScheduledThreadPool(
+      poolSize: Int,
+      threadName: String): ScheduledExecutorService = {
+    val threadFactory = new NamedThreadFactory(threadName, daemon = true)
+    val executor = new ScheduledThreadPoolExecutor(poolSize, threadFactory)
+    executor.setRemoveOnCancelPolicy(true)
+    executor
+  }
+
   def newDaemonQueuedThreadPool(
       poolSize: Int,
       poolQueueSize: Int,

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

@@ -17,12 +17,20 @@
 
 package org.apache.kyuubi.util
 
-import java.util.concurrent.TimeUnit
+import java.lang.Thread.sleep
+import java.util.concurrent._
+
+import scala.collection.concurrent.TrieMap
+
+import org.scalatest.time.{Millis, Seconds, Span}
 
 import org.apache.kyuubi.KyuubiFunSuite
 
 class ThreadUtilsSuite extends KyuubiFunSuite {
-
+  // Configure Eventually patience for retries/waits
+  implicit override val patienceConfig: PatienceConfig = PatienceConfig(
+    timeout = scaled(Span(5, Seconds)),
+    interval = scaled(Span(10, Millis)))
   test("New daemon single thread scheduled executor for shutdown") {
     val service = ThreadUtils.newDaemonSingleThreadScheduledExecutor("ThreadUtilsTest")
     @volatile var threadName = ""
@@ -61,4 +69,181 @@ class ThreadUtilsSuite extends KyuubiFunSuite {
     service.awaitTermination(10, TimeUnit.SECONDS)
     assert(threadName startsWith "")
   }
+
+  // Helper function for cleanup
+  private def shutdownAndAwaitTermination(
+      service: ExecutorService,
+      timeoutSeconds: Long = 5): Unit = {
+    service.shutdown() // Disable new tasks from being submitted
+    try {
+      // Wait a while for existing tasks to terminate
+      if (!service.awaitTermination(timeoutSeconds / 2, TimeUnit.SECONDS)) {
+        service.shutdownNow() // Cancel currently executing tasks
+        // Wait a while for tasks to respond to being cancelled
+        if (!service.awaitTermination(timeoutSeconds / 2, TimeUnit.SECONDS)) {
+          throw new IllegalStateException(
+            s"Thread pool did not terminate within $timeoutSeconds seconds")
+        }
+      }
+    } catch {
+      case _: InterruptedException =>
+        // (Re-)Cancel if current thread also interrupted
+        service.shutdownNow()
+        // Preserve interrupt status
+        Thread.currentThread().interrupt()
+    }
+  }
+  test("newDaemonScheduledThreadPool - thread naming and daemon status") {
+    val poolSize = 2
+    val threadNamePrefix = "test-pool-thread"
+    val executor = ThreadUtils.newDaemonScheduledThreadPool(poolSize, threadNamePrefix)
+    val latch = new CountDownLatch(poolSize)
+    val threadNames = TrieMap[String, Boolean]() // Thread-safe map
+    try {
+      for (_ <- 0 until poolSize) {
+        executor.submit(new Runnable {
+          override def run(): Unit = {
+            val currentThread = Thread.currentThread()
+            threadNames.put(currentThread.getName, currentThread.isDaemon)
+            latch.countDown()
+          }
+        })
+      }
+      // Wait for tasks to complete
+      assert(latch.await(5, TimeUnit.SECONDS), "Tasks did not complete in time")
+      // Verify thread names and daemon status
+      assert(threadNames.size === poolSize)
+      threadNames.foreach { case (name, isDaemon) =>
+        assert(
+          name.startsWith(threadNamePrefix),
+          s"Thread name '$name' should start with '$threadNamePrefix'")
+        assert(isDaemon, s"Thread '$name' should be a daemon thread")
+      }
+    } finally {
+      shutdownAndAwaitTermination(executor)
+    }
+  }
+  test("newDaemonScheduledThreadPool - schedule and execute tasks") {
+    val executor = ThreadUtils.newDaemonScheduledThreadPool(1, "test-schedule")
+    val taskRan = new java.util.concurrent.atomic.AtomicBoolean(false)
+    val latch = new CountDownLatch(1)
+    try {
+      val future: ScheduledFuture[_] = executor.schedule(
+        new Runnable {
+          override def run(): Unit = {
+            taskRan.set(true)
+            latch.countDown()
+          }
+        },
+        50,
+        TimeUnit.MILLISECONDS
+      ) // Schedule with a small delay
+      // Wait for the task to execute
+      assert(latch.await(2, TimeUnit.SECONDS), "Scheduled task did not run in time")
+      assert(taskRan.get(), "Scheduled task flag should be true")
+      assert(future.isDone, "Future should be done after task completion")
+    } finally {
+      shutdownAndAwaitTermination(executor)
+    }
+  }
+  test("newDaemonScheduledThreadPool - removeOnCancelPolicy works for scheduled tasks") {
+    // We need the specific ScheduledThreadPoolExecutor type to access the queue
+    val executor = ThreadUtils.newDaemonScheduledThreadPool(1, "test-cancel")
+      .asInstanceOf[ScheduledThreadPoolExecutor]
+    val taskRan = new java.util.concurrent.atomic.AtomicBoolean(false)
+    try {
+      // Schedule a task far enough in the future that we can cancel it
+      val future: ScheduledFuture[_] = executor.schedule(
+        new Runnable {
+          override def run(): Unit = {
+            taskRan.set(true)
+          }
+        },
+        5,
+        TimeUnit.SECONDS
+      ) // Long delay
+      // Verify the task is in the queue initially
+      eventually {
+        assert(executor.getQueue.size() === 1, "Task should be in the queue initially")
+      }
+      // Cancel the task
+      val cancelled = future.cancel(false) // false = don't interrupt if running (it shouldn't be)
+      assert(cancelled, "Future.cancel() should return true")
+      // Verify the task is removed from the queue due to the policy
+      eventually {
+        assert(
+          executor.getQueue.isEmpty,
+          "Task should be removed from the queue after cancellation")
+      }
+      // Wait a bit and verify the task never ran
+      Thread.sleep(100) // Give some time just in case
+      assert(!taskRan.get(), "Cancelled task should not have run")
+    } finally {
+      shutdownAndAwaitTermination(executor)
+      // Final check after shutdown
+      assert(executor.getQueue.isEmpty, "Queue should be empty after shutdown")
+    }
+  }
+  test("newDaemonScheduledThreadPool - shutdown rejects new tasks") {
+    val executor = ThreadUtils.newDaemonScheduledThreadPool(1, "test-shutdown")
+    try {
+      // Submit one task to ensure the pool is active
+      val future = executor.submit(new Runnable { override def run(): Unit = Thread.sleep(50) })
+      future.get(1, TimeUnit.SECONDS) // Wait for it to finish
+      executor.shutdown()
+      assert(executor.isShutdown, "Executor should be shutdown")
+      // Try submitting after shutdown
+      assertThrows[RejectedExecutionException] {
+        executor.submit(new Runnable { override def run(): Unit = sleep(5) })
+      }
+      assertThrows[RejectedExecutionException] {
+        executor.schedule(
+          new Runnable { override def run(): Unit = sleep(5) },
+          10,
+          TimeUnit.MILLISECONDS)
+      }
+    } finally {
+      // Ensure termination even if already shut down
+      shutdownAndAwaitTermination(executor)
+      assert(executor.isTerminated, "Executor should be terminated")
+    }
+  }
+  test("newDaemonScheduledThreadPool - concurrent execution with poolSize > 1") {
+    val poolSize = 3
+    val taskCount = 5
+    val executor = ThreadUtils.newDaemonScheduledThreadPool(poolSize, "test-concurrent")
+    val latch = new CountDownLatch(taskCount)
+    val runningThreads = TrieMap[String, Long]() // Thread Name -> Start Time
+    val executionTimes = TrieMap[Int, Long]() // Task Index -> Completion Time
+    try {
+      val startTime = System.nanoTime()
+      for (i <- 0 until taskCount) {
+        executor.submit(new Runnable {
+          override def run(): Unit = {
+            val threadName = Thread.currentThread().getName
+            runningThreads.put(threadName, System.nanoTime())
+            try {
+              // Simulate work
+              Thread.sleep(100)
+            } finally {
+              executionTimes.put(i, System.nanoTime() - startTime)
+              latch.countDown()
+            }
+          }
+        })
+      }
+      // Wait for all tasks to complete
+      assert(latch.await(5, TimeUnit.SECONDS), s"All $taskCount tasks did not complete in time")
+      // Verify that multiple threads were used (likely up to poolSize)
+      assert(
+        runningThreads.size > 1,
+        s"Expected more than 1 thread to be used, but found ${runningThreads.size}")
+      assert(
+        runningThreads.size <= poolSize,
+        s"Used ${runningThreads.size} threads, which should not exceed poolSize $poolSize")
+    } finally {
+      shutdownAndAwaitTermination(executor)
+    }
+  }
+
 }