fix hive concurrent mode load udf error problems

Author: kinghao007

linkis-computation-governance/linkis-engineconn/linkis-engineconn-plugin-core/src/main/scala/org/apache/linkis/manager/engineplugin/common/conf/EnvConfiguration.scala

@@ -19,8 +19,6 @@ package org.apache.linkis.manager.engineplugin.common.conf
 
 import org.apache.linkis.common.conf.{ByteType, CommonVars, Configuration}
 
-import org.apache.commons.lang3.{JavaVersion, SystemUtils}
-
 object EnvConfiguration {
 
   val HIVE_CONF_DIR = CommonVars[String](

linkis-computation-governance/linkis-entrance/src/test/java/org/apache/linkis/entrance/interceptor/impl/SQLExplainTest.java

@@ -18,6 +18,7 @@
 package org.apache.linkis.entrance.interceptor.impl;
 
 import org.apache.linkis.governance.common.entity.job.JobRequest;
+
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
@@ -55,24 +56,21 @@ void isSelectOverLimit() {
   }
 
   /**
-   * 未修复前代码进行拼接sql时，输出的sql为
-   *      select
-   *      id,
-   *      name,
-   *      array_join(array_intersect(map_keys(info),array['abs','oda'],' limit 5000;
-   *      ') as infos
-   *      from ods.dim_ep22
+   * 未修复前代码进行拼接sql时，输出的sql为 select id, name,
+   * array_join(array_intersect(map_keys(info),array['abs','oda'],' limit 5000; ') as infos from
+   * ods.dim_ep22
    */
   @Test
   void splicingLimitSql() {
-    String code = "select\n" +
-            "id,\n" +
-            "name,\n" +
-            "array_join(array_intersect(map_keys(info),array['abs','oda'],';') as infos\n" +
-            "from ods.dim_ep22";
+    String code =
+        "select\n"
+            + "id,\n"
+            + "name,\n"
+            + "array_join(array_intersect(map_keys(info),array['abs','oda'],';') as infos\n"
+            + "from ods.dim_ep22";
     StringBuilder logAppender = new StringBuilder();
     JobRequest jobRequest = new JobRequest();
     SQLExplain.dealSQLLimit(code, jobRequest, logAppender);
-    Assertions.assertEquals(code+" limit 5000", jobRequest.getExecutionCode());
+    Assertions.assertEquals(code + " limit 5000", jobRequest.getExecutionCode());
   }
 }

linkis-engineconn-plugins/hive/src/main/scala/org/apache/linkis/engineplugin/hive/executor/HiveEngineConcurrentConnExecutor.scala

@@ -135,7 +135,7 @@ class HiveEngineConcurrentConnExecutor(
       code: String
   ): ExecuteResponse = {
     LOG.info(s"HiveEngineConcurrentConnExecutor Ready to executeLine: $code")
-    val taskId: String = engineExecutorContext.getJobId.get
+    val taskId: String = engineExecutorContext.getJobId.getOrElse("udf_init")
     CSHiveHelper.setContextIDInfoToHiveConf(engineExecutorContext, hiveConf)
 
     val realCode = code.trim()
@@ -166,12 +166,7 @@ class HiveEngineConcurrentConnExecutor(
 
                 val driver = new HiveDriverProxy(any)
                 driverCache.put(taskId, driver)
-                executeHQL(
-                  engineExecutorContext.getJobId.get,
-                  engineExecutorContext,
-                  realCode,
-                  driver
-                )
+                executeHQL(taskId, engineExecutorContext, realCode, driver)
               case _ =>
                 val resp = proc.run(realCode.substring(tokens(0).length).trim)
                 val result = new String(baos.toByteArray)

linkis-engineconn-plugins/hive/src/main/scala/org/apache/linkis/engineplugin/hive/hook/HiveAddJarsEngineHook.scala

@@ -24,7 +24,10 @@ import org.apache.linkis.engineconn.common.engineconn.EngineConn
 import org.apache.linkis.engineconn.common.hook.EngineConnHook
 import org.apache.linkis.engineconn.computation.executor.execute.EngineExecutionContext
 import org.apache.linkis.engineconn.core.executor.ExecutorManager
-import org.apache.linkis.engineplugin.hive.executor.HiveEngineConnExecutor
+import org.apache.linkis.engineplugin.hive.executor.{
+  HiveEngineConcurrentConnExecutor,
+  HiveEngineConnExecutor
+}
 import org.apache.linkis.manager.engineplugin.common.launch.process.Environment
 import org.apache.linkis.manager.label.entity.Label
 import org.apache.linkis.manager.label.entity.engine.{CodeLanguageLabel, RunType}
@@ -78,7 +81,10 @@ class HiveAddJarsEngineHook extends EngineConnHook with Logging {
           ExecutorManager.getInstance.getExecutorByLabels(labels) match {
             case executor: HiveEngineConnExecutor =>
               executor.executeLine(new EngineExecutionContext(executor), sql)
+            case executor: HiveEngineConcurrentConnExecutor =>
+              executor.executeLine(new EngineExecutionContext(executor), sql)
             case _ =>
+              logger.warn(s"Executor is not a ComputationExecutor, skip adding jar: $jar")
           }
         } catch {
           case t: Throwable => logger.error(s"run hive sql ${addSql + jar} failed", t)