[KYUUBI #3318] Transfer the TGetInfoReq to kyuubi engine side to check the connection valid and keep connection alive

### _Why are the changes needed?_

Now the connection path is:

`client`-> `kyuubiServer` -> `kyuubiEngine`

Maybe the connection between `client -> kyuubiServer` is valid, but the connection between `kyuubiServer -> kyuubiEngine` is not.

So we need check the whole path.

BTW, before, if customer invoke KyuubiConnection::isValid, it can make the connection between client and kyuubiServer keep alive.

Now, it can make the whole connection path keep alive.

### _How was this patch tested?_
- [x] 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 #3318 from turboFei/is_valid.

Closes #3318

af4d6b7 [Fei Wang] Transfer the TGetInfoReq to kyuubi engine side to keep the kyuubi connection alive

Authored-by: Fei Wang <fwang12@ebay.com>
Signed-off-by: Fei Wang <fwang12@ebay.com>

Author: turboFei

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

@@ -75,7 +75,7 @@ abstract class AbstractSession(
     }
   }
 
-  private def withAcquireRelease[T](userAccess: Boolean = true)(f: => T): T = {
+  protected def withAcquireRelease[T](userAccess: Boolean = true)(f: => T): T = {
     acquire(userAccess)
     try f
     finally release(userAccess)

kyuubi-server/src/main/scala/org/apache/kyuubi/client/KyuubiSyncThriftClient.scala

@@ -227,6 +227,13 @@ class KyuubiSyncThriftClient private (
     resp.getOperationHandle
   }
 
+  def getInfo(infoType: TGetInfoType): TGetInfoResp = {
+    val req = new TGetInfoReq(_remoteSessionHandle, infoType)
+    val resp = withLockAcquiredAsyncRequest(GetInfo(req))
+    ThriftUtils.verifyTStatus(resp.getStatus)
+    resp
+  }
+
   def getTypeInfo: TOperationHandle = {
     val req = new TGetTypeInfoReq(_remoteSessionHandle)
     val resp = withLockAcquiredAsyncRequest(GetTypeInfo(req))

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

@@ -197,4 +197,14 @@ class KyuubiSessionImpl(
       MetricsSystem.tracing(_.decCount(MetricRegistry.name(CONN_OPEN, user)))
     }
   }
+
+  override def getInfo(infoType: TGetInfoType): TGetInfoValue = {
+    if (client != null) {
+      withAcquireRelease() {
+        client.getInfo(infoType).getInfoValue
+      }
+    } else {
+      super.getInfo(infoType)
+    }
+  }
 }

kyuubi-server/src/test/scala/org/apache/kyuubi/operation/KyuubiOperationPerConnectionSuite.scala

@@ -29,9 +29,11 @@ import org.scalatest.time.SpanSugar.convertIntToGrainOfTime
 import org.apache.kyuubi.WithKyuubiServer
 import org.apache.kyuubi.config.KyuubiConf
 import org.apache.kyuubi.config.KyuubiConf.SESSION_CONF_ADVISOR
+import org.apache.kyuubi.engine.ApplicationState
 import org.apache.kyuubi.jdbc.KyuubiHiveDriver
 import org.apache.kyuubi.jdbc.hive.KyuubiConnection
 import org.apache.kyuubi.plugin.SessionConfAdvisor
+import org.apache.kyuubi.session.KyuubiSessionManager
 
 /**
  * UT with Connection level engine shared cost much time, only run basic jdbc tests.
@@ -210,6 +212,27 @@ class KyuubiOperationPerConnectionSuite extends WithKyuubiServer with HiveJDBCTe
       assert(kyuubiConnection.isClosed)
     }
   }
+
+  test("transfer the TGetInfoReq to kyuubi engine side to verify the connection valid") {
+    withSessionConf(Map.empty)(Map(KyuubiConf.SESSION_ENGINE_LAUNCH_ASYNC.key -> "false"))() {
+      withJdbcStatement() { statement =>
+        val conn = statement.getConnection.asInstanceOf[KyuubiConnection]
+        assert(conn.isValid(3000))
+        val sessionManager = server.backendService.sessionManager.asInstanceOf[KyuubiSessionManager]
+        eventually(timeout(10.seconds)) {
+          assert(sessionManager.allSessions().size === 1)
+        }
+        val engineId = sessionManager.allSessions().head.handle.identifier.toString
+        // kill the engine application and wait the engine terminate
+        sessionManager.applicationManager.killApplication(None, engineId)
+        eventually(timeout(30.seconds), interval(100.milliseconds)) {
+          assert(sessionManager.applicationManager.getApplicationInfo(None, engineId)
+            .exists(_.state == ApplicationState.NOT_FOUND))
+        }
+        assert(!conn.isValid(3000))
+      }
+    }
+  }
 }
 
 class TestSessionConfAdvisor extends SessionConfAdvisor {