refactor(operation): Optimize operation timeout handling mechanism

- Removed ScheduledExecutorService from AbstractOperation and replaced it with Future[_]
- Added a dedicated operationTimeoutMonitor thread in SessionManager
- Updated timeout monitoring logic to use the new thread for unified management
- Optimized shutdownTimeoutMonitor method by using Future.cancel instead of ScheduledExecutorService.shutdown

Author: wangzhigang1999

kyuubi-common/src/main/scala/org/apache/kyuubi/operation/AbstractOperation.scala

@@ -18,7 +18,7 @@
 package org.apache.kyuubi.operation
 
 import java.io.IOException
-import java.util.concurrent.{Future, ScheduledExecutorService, TimeUnit}
+import java.util.concurrent.{Future, TimeUnit}
 import java.util.concurrent.locks.ReentrantLock
 
 import scala.collection.JavaConverters._
@@ -31,8 +31,7 @@ import org.apache.kyuubi.operation.FetchOrientation.FetchOrientation
 import org.apache.kyuubi.operation.OperationState._
 import org.apache.kyuubi.operation.log.OperationLog
 import org.apache.kyuubi.session.Session
-import org.apache.kyuubi.shaded.hive.service.rpc.thrift.{TFetchResultsResp, TGetResultSetMetadataResp, TProgressUpdateResp, TProtocolVersion, TStatus, TStatusCode}
-import org.apache.kyuubi.util.ThreadUtils
+import org.apache.kyuubi.shaded.hive.service.rpc.thrift._
 
 abstract class AbstractOperation(session: Session) extends Operation with Logging {
 
@@ -45,7 +44,7 @@ abstract class AbstractOperation(session: Session) extends Operation with Loggin
 
   final private[kyuubi] val statementId = handle.identifier.toString
 
-  private var statementTimeoutCleaner: Option[ScheduledExecutorService] = None
+  private var statementTimeoutCleaner: Option[Future[_]] = None
 
   private val lock: ReentrantLock = new ReentrantLock()
 
@@ -60,8 +59,6 @@ abstract class AbstractOperation(session: Session) extends Operation with Loggin
 
   protected def addTimeoutMonitor(queryTimeout: Long): Unit = {
     if (queryTimeout > 0) {
-      val timeoutExecutor =
-        ThreadUtils.newDaemonSingleThreadScheduledExecutor("query-timeout-thread", false)
       val action: Runnable = () =>
         // Clients less than version 2.1 have no HIVE-4924 Patch,
         // no queryTimeout parameter and no TIMEOUT status.
@@ -74,13 +71,13 @@ abstract class AbstractOperation(session: Session) extends Operation with Loggin
         } else {
           cleanup(OperationState.TIMEOUT)
         }
-      timeoutExecutor.schedule(action, queryTimeout, TimeUnit.SECONDS)
-      statementTimeoutCleaner = Some(timeoutExecutor)
+      statementTimeoutCleaner = Some(
+        session.sessionManager.submitTimeoutMonitor(action, queryTimeout, TimeUnit.SECONDS))
     }
   }
 
   protected def shutdownTimeoutMonitor(): Unit = {
-    statementTimeoutCleaner.foreach(_.shutdown())
+    statementTimeoutCleaner.foreach(_.cancel(true))
   }
 
   override def getOperationLog: Option[OperationLog] = None

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

@@ -19,7 +19,7 @@ package org.apache.kyuubi.session
 
 import java.io.IOException
 import java.nio.file.{Files, Paths}
-import java.util.concurrent.{ConcurrentHashMap, Future, ThreadPoolExecutor, TimeUnit}
+import java.util.concurrent.{ConcurrentHashMap, Future, ScheduledExecutorService, ThreadPoolExecutor, TimeUnit}
 
 import scala.collection.JavaConverters._
 import scala.concurrent.duration.Duration
@@ -74,12 +74,17 @@ abstract class SessionManager(name: String) extends CompositeService(name) {
   private val timeoutChecker =
     ThreadUtils.newDaemonSingleThreadScheduledExecutor(s"$name-timeout-checker")
 
+  private var operationTimeoutMonitor: ScheduledExecutorService = _
+
   protected def isServer: Boolean
 
   private var execPool: ThreadPoolExecutor = _
 
   def submitBackgroundOperation(r: Runnable): Future[_] = execPool.submit(r)
 
+  def submitTimeoutMonitor(r: Runnable, delay: Long, unit: TimeUnit): Future[_] =
+    operationTimeoutMonitor.schedule(r, delay, unit)
+
   def operationManager: OperationManager
 
   protected def createSession(
@@ -281,6 +286,11 @@ abstract class SessionManager(name: String) extends CompositeService(name) {
       waitQueueSize,
       keepAliveMs,
       s"$name-exec-pool")
+
+    // Initialize timeout monitor pool
+    operationTimeoutMonitor = ThreadUtils.newDaemonSingleThreadScheduledExecutor(
+      s"$name-operation-timeout-monitor-pool")
+
     super.initialize(conf)
   }
 
@@ -301,6 +311,7 @@ abstract class SessionManager(name: String) extends CompositeService(name) {
 
     ThreadUtils.shutdown(timeoutChecker, Duration(shutdownTimeout, TimeUnit.MILLISECONDS))
     ThreadUtils.shutdown(execPool, Duration(shutdownTimeout, TimeUnit.MILLISECONDS))
+    ThreadUtils.shutdown(operationTimeoutMonitor, Duration(shutdownTimeout, TimeUnit.MILLISECONDS))
   }
 
   private def startTimeoutChecker(): Unit = {