[KYUUBI #2402] [Improvement] addTimeoutMonitor for trino engine when it run query async

### _Why are the changes needed?_

### _How was this patch tested?_
- [ ] Add some test cases that check the changes thoroughly including negative and positive cases if possible

- [ ] Add screenshots for manual tests if appropriate

- [ ] [Run test](https://kyuubi.apache.org/docs/latest/develop_tools/testing.html#running-tests) locally before make a pull request

Closes #2404 from zhaomin1423/trino_async_addTimeoutMonitor.

Closes #2402

5e08e67 [Min Zhao] [KYUUBI #2402] [Improvement] addTimeoutMonitor for trino engine when it run query async

Authored-by: Min Zhao <zhaomin1423@163.com>
Signed-off-by: ulysses-you <ulyssesyou@apache.org>

Author: zhaomin1423

externals/kyuubi-trino-engine/src/main/scala/org/apache/kyuubi/engine/trino/operation/ExecuteStatement.scala

@@ -17,7 +17,7 @@
 
 package org.apache.kyuubi.engine.trino.operation
 
-import java.util.concurrent.RejectedExecutionException
+import java.util.concurrent.{RejectedExecutionException, ScheduledExecutorService, TimeUnit}
 
 import org.apache.kyuubi.KyuubiSQLException
 import org.apache.kyuubi.Logging
@@ -28,14 +28,18 @@ import org.apache.kyuubi.operation.OperationState
 import org.apache.kyuubi.operation.OperationType
 import org.apache.kyuubi.operation.log.OperationLog
 import org.apache.kyuubi.session.Session
+import org.apache.kyuubi.util.ThreadUtils
 
 class ExecuteStatement(
     session: Session,
     override val statement: String,
     override val shouldRunAsync: Boolean,
+    queryTimeout: Long,
     incrementalCollect: Boolean)
   extends TrinoOperation(OperationType.EXECUTE_STATEMENT, session) with Logging {
 
+  private var statementTimeoutCleaner: Option[ScheduledExecutorService] = None
+
   private val operationLog: OperationLog = OperationLog.createOperationLog(session, getHandle)
   override def getOperationLog: Option[OperationLog] = Option(operationLog)
 
@@ -50,6 +54,7 @@ class ExecuteStatement(
   }
 
   override protected def runInternal(): Unit = {
+    addTimeoutMonitor()
     val trinoStatement = TrinoStatement(trinoContext, session.sessionManager.getConf, statement)
     trino = trinoStatement.getTrinoClient
     if (shouldRunAsync) {
@@ -93,6 +98,18 @@ class ExecuteStatement(
       setState(OperationState.FINISHED)
     } catch {
       onError(cancel = true)
+    } finally {
+      statementTimeoutCleaner.foreach(_.shutdown())
+    }
+  }
+
+  private def addTimeoutMonitor(): Unit = {
+    if (queryTimeout > 0) {
+      val timeoutExecutor =
+        ThreadUtils.newDaemonSingleThreadScheduledExecutor("query-timeout-thread")
+      val action: Runnable = () => cleanup(OperationState.TIMEOUT)
+      timeoutExecutor.schedule(action, queryTimeout, TimeUnit.SECONDS)
+      statementTimeoutCleaner = Some(timeoutExecutor)
     }
   }
 }

externals/kyuubi-trino-engine/src/main/scala/org/apache/kyuubi/engine/trino/operation/TrinoOperationManager.scala

@@ -35,7 +35,8 @@ class TrinoOperationManager extends OperationManager("TrinoOperationManager") {
       runAsync: Boolean,
       queryTimeout: Long): Operation = {
     val incrementalCollect = session.sessionManager.getConf.get(OPERATION_INCREMENTAL_COLLECT)
-    val operation = new ExecuteStatement(session, statement, runAsync, incrementalCollect)
+    val operation =
+      new ExecuteStatement(session, statement, runAsync, queryTimeout, incrementalCollect)
     addOperation(operation)
   }