[KYUUBI #5784] Implement HiveTBinaryFrontendService#RenewDelegationToken

# 🔍 Description
## Issue References 🔗

We had a KyuubiServer with `kyuubi.kinit.principal=hive/xxxxxxx.xxx` and connected to it using beeline:
 ```
./bin/beeline -u "jdbc:hive2://xxxxxxx:10009/;principal=hive/_HOSTxxx.xxx;hive.server2.proxy.user=zhouyifan03#kyuubi.engine.type=hive_sql;hive.server2.enable.doAs=false;"
```
When we execute SQL `select 1`, it failed with error:
```
0: jdbc:hive2://xxxxxxx:10009/> select 1;
Error: org.apache.kyuubi.KyuubiSQLException: Failed to get metastore connection (state=,code=0)
```
HiveSQLEngine log:
```
2023-11-27 15:19:09.217 ERROR HiveTBinaryFrontendHandler-Pool: Thread-27 org.apache.thrift.transport.TSaslTransport: SASL negotiation failure
javax.security.sasl.SaslException: GSS initiate failed [Caused by GSSException: No valid credentials provided (Mechanism level: Failed to find any Kerberos tgt)]
        at com.sun.security.sasl.gsskerb.GssKrb5Client.evaluateChallenge(GssKrb5Client.java:211)
        at org.apache.thrift.transport.TSaslClientTransport.handleSaslStartMessage(TSaslClientTransport.java:94)
        at org.apache.thrift.transport.TSaslTransport.open(TSaslTransport.java:271)
        at org.apache.thrift.transport.TSaslClientTransport.open(TSaslClientTransport.java:37)
        at org.apache.hadoop.hive.thrift.client.TUGIAssumingTransport$1.run(TUGIAssumingTransport.java:52)
        at org.apache.hadoop.hive.thrift.client.TUGIAssumingTransport$1.run(TUGIAssumingTransport.java:49)
        at java.security.AccessController.doPrivileged(Native Method)
        at javax.security.auth.Subject.doAs(Subject.java:422)
        at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1875)
        at org.apache.hadoop.hive.thrift.client.TUGIAssumingTransport.open(TUGIAssumingTransport.java:49)
        at org.apache.hadoop.hive.metastore.HiveMetaStoreClient.open(HiveMetaStoreClient.java:545)
        at org.apache.hadoop.hive.metastore.HiveMetaStoreClient.<init>(HiveMetaStoreClient.java:303)
        at org.apache.hadoop.hive.ql.metadata.SessionHiveMetaStoreClient.<init>(SessionHiveMetaStoreClient.java:70)
        at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
        at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
        at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
        at java.lang.reflect.Constructor.newInstance(Constructor.java:423)
        at org.apache.hadoop.hive.metastore.MetaStoreUtils.newInstance(MetaStoreUtils.java:1773)
        at org.apache.hadoop.hive.metastore.RetryingMetaStoreClient.<init>(RetryingMetaStoreClient.java:80)
        at org.apache.hadoop.hive.metastore.RetryingMetaStoreClient.getProxy(RetryingMetaStoreClient.java:130)
        at org.apache.hadoop.hive.metastore.RetryingMetaStoreClient.getProxy(RetryingMetaStoreClient.java:101)
        at org.apache.hadoop.hive.ql.metadata.Hive.createMetaStoreClient(Hive.java:3819)
        at org.apache.hadoop.hive.ql.metadata.Hive.getMSC(Hive.java:3871)
        at org.apache.hadoop.hive.ql.metadata.Hive.getMSC(Hive.java:3851)
        at org.apache.hadoop.hive.ql.metadata.Hive.getAllFunctions(Hive.java:4105)
        at org.apache.hadoop.hive.ql.metadata.Hive.reloadFunctions(Hive.java:254)
        at org.apache.hadoop.hive.ql.metadata.Hive.registerAllFunctionsOnce(Hive.java:237)
        at org.apache.hadoop.hive.ql.metadata.Hive.<init>(Hive.java:394)
        at org.apache.hadoop.hive.ql.metadata.Hive.create(Hive.java:338)
        at org.apache.hadoop.hive.ql.metadata.Hive.getInternal(Hive.java:318)
        at org.apache.hadoop.hive.ql.metadata.Hive.get(Hive.java:294)
        at org.apache.hive.service.cli.session.HiveSessionImpl.open(HiveSessionImpl.java:181)
        at org.apache.kyuubi.engine.hive.session.HiveSessionImpl.open(HiveSessionImpl.scala:51)
        at org.apache.kyuubi.session.SessionManager.openSession(SessionManager.scala:109)
        at org.apache.kyuubi.service.AbstractBackendService.openSession(AbstractBackendService.scala:46)
        at org.apache.kyuubi.service.TFrontendService.getSessionHandle(TFrontendService.scala:182)
```

## Describe Your Solution 🔧

In order to pass HiveMetaStoreClient authentication, we need to add a HIVE_DELEGATION_TOKEN  into  proxy user `zhouyifan03`. It can be achieved by implementing HiveTBinaryFrontendService#RenewDelegationToken.

## Types of changes 🔖

- [x] Bugfix (non-breaking change which fixes an issue)
- [ ] New feature (non-breaking change which adds functionality)
- [ ] Breaking change (fix or feature that would cause existing functionality to change)

## Test Plan 🧪

#### Behavior Without This Pull Request ⚰️

<img width="1009" alt="image" src="https://github.com/apache/kyuubi/assets/88070094/9a22232f-dc1b-4557-acd7-1c16463d651f">

#### Behavior With This Pull Request 🎉

<img width="1542" alt="image" src="https://github.com/apache/kyuubi/assets/88070094/9e0658e1-af3a-4970-8c47-9629c183ea9e">

#### Related Unit Tests

---

# Checklists
## 📝 Author Self Checklist

- [x] My code follows the [style guidelines](https://kyuubi.readthedocs.io/en/master/contributing/code/style.html) of this project
- [x] I have performed a self-review
- [x] I have commented my code, particularly in hard-to-understand areas
- [ ] I have made corresponding changes to the documentation
- [x] My changes generate no new warnings
- [ ] I have added tests that prove my fix is effective or that my feature works
- [x] New and existing unit tests pass locally with my changes
- [x] This patch was not authored or co-authored using [Generative Tooling](https://www.apache.org/legal/generative-tooling.html)

## 📝 Committer Pre-Merge Checklist

- [ ] Pull request title is okay.
- [ ] No license issues.
- [ ] Milestone correctly set?
- [ ] Test coverage is ok
- [ ] Assignees are selected.
- [ ] Minimum number of approvals
- [ ] No changes are requested

**Be nice. Be informative.**

Closes #5784 from zhouyifan279/hive-engine-renew-token.

Closes #5784

d0e7917 [Cheng Pan] Update externals/kyuubi-hive-sql-engine/src/main/scala/org/apache/kyuubi/engine/hive/HiveTBinaryFrontendService.scala
9f6c452 [zhouyifan279] Implement HiveTBinaryFrontendService#RenewDelegationToken

Lead-authored-by: zhouyifan279 <[email protected]>
Co-authored-by: Cheng Pan <[email protected]>
Signed-off-by: Cheng Pan <[email protected]>

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

@@ -130,7 +130,15 @@ object HiveSQLEngine extends Logging {
       } else {
         val effectiveUser = UserGroupInformation.createProxyUser(sessionUser.get, realUser)
         effectiveUser.doAs(new PrivilegedExceptionAction[Unit] {
-          override def run(): Unit = startEngine()
+          override def run(): Unit = {
+            val engineCredentials =
+              kyuubiConf.getOption(KyuubiReservedKeys.KYUUBI_ENGINE_CREDENTIALS_KEY)
+            kyuubiConf.unset(KyuubiReservedKeys.KYUUBI_ENGINE_CREDENTIALS_KEY)
+            engineCredentials.filter(_.nonEmpty).foreach { credentials =>
+              HiveTBinaryFrontendService.renewDelegationToken(credentials)
+            }
+            startEngine()
+          }
         })
       }
 

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

@@ -17,11 +17,19 @@
 
 package org.apache.kyuubi.engine.hive
 
+import org.apache.hadoop.io.Text
+import org.apache.hadoop.security.UserGroupInformation
+import org.apache.hive.service.rpc.thrift.{TRenewDelegationTokenReq, TRenewDelegationTokenResp}
+
+import org.apache.kyuubi.KyuubiSQLException
 import org.apache.kyuubi.ha.client.{EngineServiceDiscovery, ServiceDiscovery}
 import org.apache.kyuubi.service.{Serverable, Service, TBinaryFrontendService}
+import org.apache.kyuubi.service.TFrontendService.OK_STATUS
+import org.apache.kyuubi.util.KyuubiHadoopUtils
 
 class HiveTBinaryFrontendService(override val serverable: Serverable)
   extends TBinaryFrontendService("HiveTBinaryFrontend") {
+  import HiveTBinaryFrontendService._
 
   override lazy val discoveryService: Option[Service] = {
     if (ServiceDiscovery.supportServiceDiscovery(conf)) {
@@ -30,4 +38,39 @@ class HiveTBinaryFrontendService(override val serverable: Serverable)
       None
     }
   }
+
+  override def RenewDelegationToken(req: TRenewDelegationTokenReq): TRenewDelegationTokenResp = {
+    debug(req.toString)
+
+    // We hacked `TCLIService.Iface.RenewDelegationToken` to transfer Credentials from Kyuubi
+    // Server to Hive SQL engine
+    val resp = new TRenewDelegationTokenResp()
+    try {
+      renewDelegationToken(req.getDelegationToken)
+      resp.setStatus(OK_STATUS)
+    } catch {
+      case e: Exception =>
+        warn("Error renew delegation tokens: ", e)
+        resp.setStatus(KyuubiSQLException.toTStatus(e))
+    }
+    resp
+  }
+}
+
+object HiveTBinaryFrontendService {
+
+  def renewDelegationToken(tokenStr: String): Unit = {
+    val currentUser = UserGroupInformation.getCurrentUser
+    // `currentUser` is either `UserGroupInformation.getLoginUser` or a proxy user.
+    // If `currentUser` is a proxy user, it needs a HIVE_DELEGATION_TOKEN to pass
+    // HiveMetastoreClient authentication.
+    if (currentUser.getAuthenticationMethod == UserGroupInformation.AuthenticationMethod.PROXY) {
+      val newCreds = KyuubiHadoopUtils.decodeCredentials(tokenStr)
+      KyuubiHadoopUtils.getTokenMap(newCreds).values
+        .find(_.getKind == new Text("HIVE_DELEGATION_TOKEN"))
+        .foreach { token =>
+          UserGroupInformation.getCurrentUser.addToken(token)
+        }
+    }
+  }
 }