[KYUUBI #5778] Hive engine shall respect hive.server2.enable.doAs

# 🔍 Description
## Issue References 🔗

Currently, the Hive engine ignores `hive.server2.enable.doAs` and always constructs `HiveSessionImplwithUGI`

## Describe Your Solution 🔧

Hive engine shall respect `hive.server2.enable.doAs`, to align the behavior with HiveServer2

## 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 ⚰️

The Hive engine always constructs `HiveSessionImplwithUGI`

#### Behavior With This Pull Request 🎉

When `hive.server2.enable.doAs` is `true`, the Hive engine constructs `HiveSessionImplwithUGI`;

When `hive.server2.enable.doAs` is `false`, the Hive engine constructs `HiveSessionImpl`;

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

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

**Be nice. Be informative.**

Closes #5778 from pan3793/hive-doas.

Closes #5778

4ac97a7 [Cheng Pan] Hive engine shall respect `hive.server2.enable.doAs`

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/session/HiveSessionManager.scala

@@ -22,10 +22,12 @@ import java.util.{List => JList}
 import java.util.concurrent.Future
 
 import scala.collection.JavaConverters._
+import scala.language.reflectiveCalls
 
 import org.apache.hadoop.hive.conf.HiveConf
+import org.apache.hadoop.hive.conf.HiveConf.ConfVars
 import org.apache.hive.service.cli.{SessionHandle => ImportedSessionHandle}
-import org.apache.hive.service.cli.session.{HiveSessionImplwithUGI => ImportedHiveSessionImpl, HiveSessionProxy, SessionManager => ImportedHiveSessionManager}
+import org.apache.hive.service.cli.session.{HiveSessionImpl => ImportedHiveSessionImpl, HiveSessionImplwithUGI => ImportedHiveSessionImplwithUGI, HiveSessionProxy, SessionManager => ImportedHiveSessionManager}
 import org.apache.hive.service.rpc.thrift.TProtocolVersion
 
 import org.apache.kyuubi.config.KyuubiConf.ENGINE_SHARE_LEVEL
@@ -44,11 +46,14 @@ class HiveSessionManager(engine: HiveSQLEngine) extends SessionManager("HiveSess
 
   private val internalSessionManager = new ImportedHiveSessionManager(null) {
 
+    var doAsEnabled: Boolean = _
+
     /**
      * Avoid unnecessary hive initialization
      */
     override def init(hiveConf: HiveConf): Unit = {
       // this.hiveConf = hiveConf
+      this.doAsEnabled = hiveConf.getBoolVar(ConfVars.HIVE_SERVER2_ENABLE_DOAS)
     }
 
     /**
@@ -79,11 +84,10 @@ class HiveSessionManager(engine: HiveSQLEngine) extends SessionManager("HiveSess
       getSessionOption).getOrElse {
       val sessionHandle =
         conf.get(KYUUBI_SESSION_HANDLE_KEY).map(SessionHandle.fromUUID).getOrElse(SessionHandle())
-      val hive = {
-
+      val hive = if (internalSessionManager.doAsEnabled) {
         val sessionWithUGI = DynConstructors.builder()
           .impl( // for Hive 3.1
-            classOf[ImportedHiveSessionImpl],
+            classOf[ImportedHiveSessionImplwithUGI],
             classOf[ImportedSessionHandle],
             classOf[TProtocolVersion],
             classOf[String],
@@ -93,15 +97,15 @@ class HiveSessionManager(engine: HiveSQLEngine) extends SessionManager("HiveSess
             classOf[String],
             classOf[JList[String]])
           .impl( // for Hive 2.3
-            classOf[ImportedHiveSessionImpl],
+            classOf[ImportedHiveSessionImplwithUGI],
             classOf[ImportedSessionHandle],
             classOf[TProtocolVersion],
             classOf[String],
             classOf[String],
             classOf[HiveConf],
             classOf[String],
             classOf[String])
-          .build[ImportedHiveSessionImpl]()
+          .build[ImportedHiveSessionImplwithUGI]()
           .newInstance(
             new ImportedSessionHandle(sessionHandle.toTSessionHandle, protocol),
             protocol,
@@ -114,6 +118,34 @@ class HiveSessionManager(engine: HiveSQLEngine) extends SessionManager("HiveSess
         val proxy = HiveSessionProxy.getProxy(sessionWithUGI, sessionWithUGI.getSessionUgi)
         sessionWithUGI.setProxySession(proxy)
         proxy
+      } else {
+        DynConstructors.builder()
+          .impl( // for Hive 3.1
+            classOf[ImportedHiveSessionImpl],
+            classOf[ImportedSessionHandle],
+            classOf[TProtocolVersion],
+            classOf[String],
+            classOf[String],
+            classOf[HiveConf],
+            classOf[String],
+            classOf[JList[String]])
+          .impl( // for Hive 2.3
+            classOf[ImportedHiveSessionImpl],
+            classOf[ImportedSessionHandle],
+            classOf[TProtocolVersion],
+            classOf[String],
+            classOf[String],
+            classOf[HiveConf],
+            classOf[String])
+          .build[ImportedHiveSessionImpl]()
+          .newInstance(
+            new ImportedSessionHandle(sessionHandle.toTSessionHandle, protocol),
+            protocol,
+            user,
+            password,
+            HiveSQLEngine.hiveConf,
+            ipAddress,
+            Seq(ipAddress).asJava)
       }
       hive.setSessionManager(internalSessionManager)
       hive.setOperationManager(internalSessionManager.getOperationManager)