[KYUUBI #2927] Fix the thread in ScheduleThreadExecutorPool can't be shutdown immediately

fix #2927

### _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 #2928 from iodone/dev-2.

Closes #2927

a5cb190 [odone] [KYUUBI #2927] fixed

Authored-by: odone <odone.zhang@gmail.com>
Signed-off-by: ulysses-you <ulyssesyou@apache.org>

Author: iodone

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

@@ -54,7 +54,7 @@ abstract class AbstractOperation(opType: OperationType, session: Session)
   protected def addTimeoutMonitor(queryTimeout: Long): Unit = {
     if (queryTimeout > 0) {
       val timeoutExecutor =
-        ThreadUtils.newDaemonSingleThreadScheduledExecutor("query-timeout-thread")
+        ThreadUtils.newDaemonSingleThreadScheduledExecutor("query-timeout-thread", false)
       val action: Runnable = () => cleanup(OperationState.TIMEOUT)
       timeoutExecutor.schedule(action, queryTimeout, TimeUnit.SECONDS)
       statementTimeoutCleaner = Some(timeoutExecutor)

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

@@ -26,11 +26,15 @@ import org.apache.kyuubi.{KyuubiException, Logging}
 
 object ThreadUtils extends Logging {
 
-  def newDaemonSingleThreadScheduledExecutor(threadName: String): ScheduledExecutorService = {
+  def newDaemonSingleThreadScheduledExecutor(
+      threadName: String,
+      executeExistingDelayedTasksAfterShutdown: Boolean = true): ScheduledExecutorService = {
     val threadFactory = new NamedThreadFactory(threadName, daemon = true)
     val executor = new ScheduledThreadPoolExecutor(1, threadFactory)
     executor.setRemoveOnCancelPolicy(true)
     executor
+      .setExecuteExistingDelayedTasksAfterShutdownPolicy(executeExistingDelayedTasksAfterShutdown)
+    executor
   }
 
   def newDaemonQueuedThreadPool(

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

@@ -23,7 +23,7 @@ import org.apache.kyuubi.KyuubiFunSuite
 
 class ThreadUtilsSuite extends KyuubiFunSuite {
 
-  test("New daemon single thread scheduled executor") {
+  test("New daemon single thread scheduled executor for shutdown") {
     val service = ThreadUtils.newDaemonSingleThreadScheduledExecutor("ThreadUtilsTest")
     @volatile var threadName = ""
     service.submit(new Runnable {
@@ -35,4 +35,30 @@ class ThreadUtilsSuite extends KyuubiFunSuite {
     service.awaitTermination(10, TimeUnit.SECONDS)
     assert(threadName startsWith "ThreadUtilsTest")
   }
+
+  test("New daemon single thread scheduled executor for shutdownNow") {
+    val service = ThreadUtils.newDaemonSingleThreadScheduledExecutor("ThreadUtilsTest")
+    @volatile var threadName = ""
+    service.submit(new Runnable {
+      override def run(): Unit = {
+        threadName = Thread.currentThread().getName
+      }
+    })
+    service.shutdownNow()
+    service.awaitTermination(10, TimeUnit.SECONDS)
+    assert(threadName startsWith "")
+  }
+
+  test("New daemon single thread scheduled executor for cancel delayed tasks") {
+    val service = ThreadUtils.newDaemonSingleThreadScheduledExecutor("ThreadUtilsTest", false)
+    @volatile var threadName = ""
+    service.submit(new Runnable {
+      override def run(): Unit = {
+        threadName = Thread.currentThread().getName
+      }
+    })
+    service.shutdown()
+    service.awaitTermination(10, TimeUnit.SECONDS)
+    assert(threadName startsWith "")
+  }
 }