Remove unused test case

Author: ruanwenjun

externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/operation/PlanOnlyStatement.scala

@@ -18,7 +18,6 @@
 package org.apache.kyuubi.engine.spark.operation
 
 import java.lang.{Boolean => JBoolean}
-import java.util.concurrent.{TimeoutException, TimeUnit}
 
 import com.fasterxml.jackson.databind.ObjectMapper
 import com.fasterxml.jackson.module.scala.DefaultScalaModule
@@ -39,7 +38,6 @@ import org.apache.kyuubi.operation.PlanOnlyMode.{notSupportedModeError, unknownM
 import org.apache.kyuubi.operation.PlanOnlyStyle.{notSupportedStyleError, unknownStyleError}
 import org.apache.kyuubi.operation.log.OperationLog
 import org.apache.kyuubi.session.Session
-import org.apache.kyuubi.util.ThreadUtils
 import org.apache.kyuubi.util.reflect.DynMethods
 
 /**
@@ -80,20 +78,10 @@ class PlanOnlyStatement(
   override protected def runInternal(): Unit =
     try {
       if (queryTimeout > 0) {
-        val timeoutExecutor =
-          ThreadUtils.newDaemonSingleThreadScheduledExecutor("query-timeout-thread", false)
-        val future = timeoutExecutor.submit(new Runnable {
-          override def run(): Unit = doParsePlan()
-        })
-        try {
-          future.get(queryTimeout, TimeUnit.SECONDS)
-        } catch {
-          case _: TimeoutException =>
-            future.cancel(true)
-            cleanup(OperationState.TIMEOUT)
-        } finally {
-          timeoutExecutor.shutdownNow()
-        }
+        session.sessionManager.operationManager.runWithTimeoutFallback(
+          () => doParsePlan(),
+          () => cleanup(OperationState.TIMEOUT),
+          queryTimeout)
       } else {
         doParsePlan()
       }

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

@@ -17,7 +17,7 @@
 
 package org.apache.kyuubi.operation
 
-import java.util.concurrent.{ScheduledExecutorService, ScheduledFuture, TimeUnit}
+import java.util.concurrent.{ScheduledExecutorService, ScheduledFuture, TimeoutException, TimeUnit}
 
 import scala.collection.JavaConverters._
 
@@ -70,6 +70,22 @@ abstract class OperationManager(name: String) extends AbstractService(name) {
     super.stop()
   }
 
+  /**
+   * Submit a task to the internal timeout scheduler and run it with a timeout.
+   * If the task execution exceeds the specified timeout, the task will be cancelled
+   * and the provided onTimeout callback will be executed.
+   */
+  def runWithTimeoutFallback(task: Runnable, onTimeout: Runnable, timeoutSeconds: Long): Unit = {
+    val future = timeoutScheduler.submit(task)
+    try {
+      future.get(timeoutSeconds, TimeUnit.SECONDS)
+    } catch {
+      case _: TimeoutException =>
+        future.cancel(true)
+        onTimeout.run()
+    }
+  }
+
   /** Schedule a timeout task using the internal scheduler */
   def scheduleTimeout(action: Runnable, timeoutSeconds: Long): ScheduledFuture[_] = {
     timeoutScheduler.schedule(action, timeoutSeconds, TimeUnit.SECONDS)

kyuubi-server/src/test/scala/org/apache/kyuubi/engine/spark/SparkSqlEngineSuite.scala

@@ -17,8 +17,6 @@
 
 package org.apache.kyuubi.engine.spark
 
-import java.sql.SQLTimeoutException
-
 import org.apache.kyuubi.WithKyuubiServer
 import org.apache.kyuubi.config.KyuubiConf
 import org.apache.kyuubi.config.KyuubiConf._
@@ -151,38 +149,5 @@ class SparkSqlEngineSuite extends WithKyuubiServer with HiveJDBCTestHelper {
     }
   }
 
-  test("KYUUBI-7323: Support timeout at PlanOnlyMode") {
-    val tableName = "t_timeout"
-    val sessionConfMap = Map("kyuubi.operation.plan.only.mode=analyze" -> "analyze")
-    withSessionConf(sessionConfMap)(Map.empty)(Map.empty) {
-
-      withJdbcStatement(tableName) { statement =>
-        statement.setQueryTimeout(2)
-        val layers = 30
-
-        val cteBuilder = new StringBuilder()
-        cteBuilder.append("WITH p AS (SELECT dt, tid, count(1) as pv, DENSE_RANK() " +
-          "over(order by dt) as dt_rank FROM t GROUP BY 1, 2), ")
-        cteBuilder.append("test_d1 AS (SELECT dt, tid FROM (SELECT dt, tid, row_number() " +
-          "over(order by pv desc) as r FROM p WHERE dt_rank=1) WHERE r<=100)")
-
-        for (i <- 2 to layers) {
-          cteBuilder.append(s", test_d$i AS ( " +
-            s"SELECT dt, tid FROM (SELECT p.dt, p.tid, row_number() over(order by pv desc) as r " +
-            s"FROM p LEFT JOIN test_d${i - 1} prev ON p.tid = prev.tid " +
-            s"WHERE p.dt_rank=$i AND prev.tid IS NULL) WHERE r<=100 " +
-            s"UNION ALL SELECT dt, tid FROM test_d${i - 1})")
-        }
-
-        val complexSql = cteBuilder.toString() + s" SELECT * FROM test_d$layers"
-
-        val e = intercept[SQLTimeoutException] {
-          statement.executeQuery(complexSql)
-        }
-        assert(e.getMessage.equals("Query timed out after 2 seconds"))
-      }
-    }
-  }
-
   override protected def jdbcUrl: String = getJdbcUrl
 }