[KYUUBI #2020] [Subtask] Hive Backend Engine - new APIs with hive-service-rpc 3.1.2 - TGetQueryId

### _Why are the changes needed?_

Hive Backend Engine - new APIs with hive-service-rpc 3.1.2 - TGetQueryId

### _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

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

Closes #2382 from yangrong688/KYUUBI-2020.

Closes #2020

97eccc0 [yangrong688] fix spotless check
efe0116 [yangrong688] add javadoc and add returns null if stmtHandle is null
a9d04e6 [yangrong688] fix
ac0dff2 [yangrong688] fix
3e93580 [yangrong688] skip test when java version beyond JAVA_1_8
2a777e2 [yangrong688] update getQueryId test case
ec12d2a [yangrong688] [feat] complete TGetQueryId op, should refined test case later
2996579 [yangrong] [feat] init operation getQueryId, just structure, need to improve quickly

Lead-authored-by: yangrong688 <yangrong.jxufe@gmail.com>
Co-authored-by: yangrong <yangrong.jxufe@gmail.com>
Signed-off-by: Kent Yao <yao@apache.org>

Author: yangrong688

externals/kyuubi-flink-sql-engine/src/main/scala/org/apache/kyuubi/engine/flink/operation/FlinkSQLOperationManager.scala

@@ -144,4 +144,8 @@ class FlinkSQLOperationManager extends OperationManager("FlinkSQLOperationManage
       foreignTable: String): Operation = {
     throw KyuubiSQLException.featureNotSupported()
   }
+
+  override def getQueryId(operation: Operation): String = {
+    throw KyuubiSQLException.featureNotSupported()
+  }
 }

externals/kyuubi-hive-sql-engine/src/main/scala/org/apache/kyuubi/engine/hive/operation/HiveOperationManager.scala

@@ -22,6 +22,7 @@ import java.util.List
 
 import scala.collection.JavaConverters._
 
+import org.apache.hadoop.hive.conf.HiveConf.ConfVars
 import org.apache.hadoop.hive.metastore.api.{FieldSchema, Schema}
 import org.apache.hive.service.cli.{RowSetFactory, TableSchema}
 import org.apache.hive.service.rpc.thrift.TRowSet
@@ -161,4 +162,10 @@ class HiveOperationManager() extends OperationManager("HiveOperationManager") {
 
     rowSet.toTRowSet
   }
+
+  override def getQueryId(operation: Operation): String = {
+    val hiveOperation = operation.asInstanceOf[HiveOperation]
+    val internalHiveOperation = hiveOperation.internalHiveOperation
+    internalHiveOperation.getParentSession.getHiveConf.getVar(ConfVars.HIVEQUERYID)
+  }
 }

externals/kyuubi-hive-sql-engine/src/test/scala/org/apache/kyuubi/engine/hive/operation/HiveOperationSuite.scala

@@ -17,8 +17,11 @@
 
 package org.apache.kyuubi.engine.hive.operation
 
+import org.apache.commons.lang3.{JavaVersion, SystemUtils}
+
 import org.apache.kyuubi.{HiveEngineTests, Utils}
 import org.apache.kyuubi.engine.hive.HiveSQLEngine
+import org.apache.kyuubi.jdbc.hive.KyuubiStatement
 
 class HiveOperationSuite extends HiveEngineTests {
 
@@ -35,4 +38,15 @@ class HiveOperationSuite extends HiveEngineTests {
   override protected def jdbcUrl: String = {
     "jdbc:hive2://" + HiveSQLEngine.currentEngine.get.frontendServices.head.connectionUrl + "/;"
   }
+
+  test("test get query id") {
+    assume(SystemUtils.isJavaVersionAtMost(JavaVersion.JAVA_1_8))
+    withJdbcStatement("hive_engine_test") { statement =>
+      statement.execute("CREATE TABLE hive_engine_test(id int, value string) stored as orc")
+      statement.execute("INSERT INTO hive_engine_test SELECT 1, '2'")
+      statement.executeQuery("SELECT ID, VALUE FROM hive_engine_test")
+      val kyuubiStatement = statement.asInstanceOf[KyuubiStatement]
+      assert(kyuubiStatement.getQueryId != null)
+    }
+  }
 }

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

@@ -151,4 +151,8 @@ class SparkSQLOperationManager private (name: String) extends OperationManager(n
       foreignTable: String): Operation = {
     throw KyuubiSQLException.featureNotSupported()
   }
+
+  override def getQueryId(operation: Operation): String = {
+    throw KyuubiSQLException.featureNotSupported()
+  }
 }

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

@@ -116,4 +116,8 @@ class TrinoOperationManager extends OperationManager("TrinoOperationManager") {
       foreignTable: String): Operation = {
     throw KyuubiSQLException.featureNotSupported()
   }
+
+  override def getQueryId(operation: Operation): String = {
+    throw KyuubiSQLException.featureNotSupported()
+  }
 }

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

@@ -85,6 +85,7 @@ abstract class OperationManager(name: String) extends AbstractService(name) {
       foreignCatalog: String,
       foreignSchema: String,
       foreignTable: String): Operation
+  def getQueryId(operation: Operation): String
 
   final def addOperation(operation: Operation): Operation = synchronized {
     handleToOperation.put(operation.getHandle, operation)

kyuubi-common/src/main/scala/org/apache/kyuubi/service/AbstractBackendService.scala

@@ -150,6 +150,12 @@ abstract class AbstractBackendService(name: String)
         foreignTable)
   }
 
+  override def getQueryId(operationHandle: OperationHandle): String = {
+    val operation = sessionManager.operationManager.getOperation(operationHandle)
+    val queryId = sessionManager.operationManager.getQueryId(operation)
+    queryId
+  }
+
   override def getOperationStatus(operationHandle: OperationHandle): OperationStatus = {
     val operation = sessionManager.operationManager.getOperation(operationHandle)
     if (operation.shouldRunAsync) {

kyuubi-common/src/main/scala/org/apache/kyuubi/service/BackendService.scala

@@ -89,6 +89,7 @@ trait BackendService {
       foreignCatalog: String,
       foreignSchema: String,
       foreignTable: String): OperationHandle
+  def getQueryId(operationHandle: OperationHandle): String
 
   def getOperationStatus(operationHandle: OperationHandle): OperationStatus
   def cancelOperation(operationHandle: OperationHandle): Unit

kyuubi-common/src/main/scala/org/apache/kyuubi/service/TFrontendService.scala

@@ -529,6 +529,8 @@ abstract class TFrontendService(name: String)
   override def GetQueryId(req: TGetQueryIdReq): TGetQueryIdResp = {
     debug(req.toString)
     val resp = new TGetQueryIdResp
+    val queryId = be.getQueryId(OperationHandle(req.getOperationHandle))
+    resp.setQueryId(queryId)
     resp
   }
 

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

@@ -200,6 +200,12 @@ abstract class AbstractSession(
     runOperation(operation)
   }
 
+  override def getQueryId(operationHandle: OperationHandle): String = {
+    val operation = sessionManager.operationManager.getOperation(operationHandle)
+    val queryId = sessionManager.operationManager.getQueryId(operation)
+    queryId
+  }
+
   override def cancelOperation(operationHandle: OperationHandle): Unit = withAcquireRelease() {
     sessionManager.operationManager.cancelOperation(operationHandle)
   }