[KYUUBI #7290] Prevent engine session leak after Kyuubi session closed

### Why are the changes needed?
close #7290
When we close a kyuubi session, if the engine session(thrift client) has not initialized, then the kyuubi session will be closed, but the engine session might be alive, then will result in a resource leak.

### How was this patch tested?
Can be tested by when the engine pod is pending, and then kill the jdbc client, the driver pod should be killed after initialized.

### Was this patch authored or co-authored using generative AI tooling?
NO

Closes #7294 from ruanwenjun/dev_wenjun_fix7290.

Closes #7290

142a210 [ruanwenjun] make sure the client closed
9745206 [ruanwenjun] set shouldRetry to false
d3c8608 [ruanwenjun] remove unused comment
866126d [ruanwenjun] If the session already closed, then return
3b2ee39 [ruanwenjun] [KYUUBI #7290] Fix the engine session might still alive when kyuubi session has been closed

Authored-by: ruanwenjun <[email protected]>
Signed-off-by: Cheng Pan <[email protected]>
(cherry picked from commit 8bed9de)
Signed-off-by: Cheng Pan <[email protected]>

Author: ruanwenjun

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

@@ -52,6 +52,8 @@ abstract class AbstractSession(
   @volatile private var _lastIdleTime: Long = _createTime
   override def lastIdleTime: Long = _lastIdleTime
 
+  @volatile private var _closed: Boolean = false
+
   override def getNoOperationTime: Long = {
     if (lastIdleTime > 0) System.currentTimeMillis() - _lastIdleTime else 0
   }
@@ -87,6 +89,11 @@ abstract class AbstractSession(
   }
 
   override def close(): Unit = withAcquireRelease() {
+    if (_closed) {
+      return
+    }
+    _closed = true
+    info(s"Mark session $handle closed")
     opHandleSet.forEach { opHandle =>
       try {
         sessionManager.operationManager.closeOperation(opHandle)
@@ -95,8 +102,11 @@ abstract class AbstractSession(
           warn(s"Error closing operation $opHandle during closing $handle for", e)
       }
     }
+    info(s"Closed all operations for session $handle")
   }
 
+  override def isClosed: Boolean = _closed
+
   protected def runOperation(operation: Operation): OperationHandle = {
     try {
       val opHandle = operation.getHandle

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

@@ -44,6 +44,8 @@ trait Session {
   def open(): Unit
   def close(): Unit
 
+  def isClosed: Boolean
+
   def getInfo(infoType: TGetInfoType): TGetInfoValue
 
   def executeStatement(

kyuubi-server/src/main/scala/org/apache/kyuubi/session/KyuubiSessionImpl.scala

@@ -180,6 +180,10 @@ class KyuubiSessionImpl(
             _engineSessionHandle =
               engineClient.openSession(protocol, user, passwd, openEngineSessionConf)
             _client = engineClient
+            if (isClosed) {
+              shouldRetry = false
+              throw KyuubiSQLException(s"KyuubiSession $handle has been closed")
+            }
             logSessionInfo(s"Connected to engine [$host:$port]/[${client.engineId.getOrElse("")}]" +
               s" with ${_engineSessionHandle}]")
             shouldRetry = false
@@ -210,7 +214,7 @@ class KyuubiSessionImpl(
               throw e
           } finally {
             attempt += 1
-            if (shouldRetry && engineClient != null) {
+            if ((isClosed || shouldRetry) && engineClient != null) {
               try {
                 engineClient.closeSession()
               } catch {