diff --git a/linkis-commons/linkis-storage/src/main/scala/org/apache/linkis/storage/utils/StorageUtils.scala b/linkis-commons/linkis-storage/src/main/scala/org/apache/linkis/storage/utils/StorageUtils.scala
index 7a5b134749..7ee53a45cf 100644
--- a/linkis-commons/linkis-storage/src/main/scala/org/apache/linkis/storage/utils/StorageUtils.scala
+++ b/linkis-commons/linkis-storage/src/main/scala/org/apache/linkis/storage/utils/StorageUtils.scala
@@ -219,4 +219,8 @@ object StorageUtils extends Logging {
     StorageConfiguration.ENABLE_IO_PROXY.getValue
   }
 
+  def isHDFSPath(fsPath: FsPath): Boolean = {
+    HDFS.equals(fsPath.getFsType)
+  }
+
 }
diff --git a/linkis-computation-governance/linkis-entrance/src/main/scala/org/apache/linkis/entrance/conf/EntranceConfiguration.scala b/linkis-computation-governance/linkis-entrance/src/main/scala/org/apache/linkis/entrance/conf/EntranceConfiguration.scala
index 4d9e895de6..ca00508584 100644
--- a/linkis-computation-governance/linkis-entrance/src/main/scala/org/apache/linkis/entrance/conf/EntranceConfiguration.scala
+++ b/linkis-computation-governance/linkis-entrance/src/main/scala/org/apache/linkis/entrance/conf/EntranceConfiguration.scala
@@ -242,4 +242,7 @@ object EntranceConfiguration {
   val ENTRANCE_TASK_TIMEOUT_SCAN =
     CommonVars("wds.linkis.entrance.task.timeout.scan", new TimeType("12h"))
 
+  val ENABLE_HDFS_JVM_USER =
+    CommonVars[Boolean]("linkis.entrance.enable.hdfs.jvm.user", true).getValue
+
 }
diff --git a/linkis-computation-governance/linkis-entrance/src/main/scala/org/apache/linkis/entrance/log/CacheLogReader.scala b/linkis-computation-governance/linkis-entrance/src/main/scala/org/apache/linkis/entrance/log/CacheLogReader.scala
index 483cf9ab43..748f82df4b 100644
--- a/linkis-computation-governance/linkis-entrance/src/main/scala/org/apache/linkis/entrance/log/CacheLogReader.scala
+++ b/linkis-computation-governance/linkis-entrance/src/main/scala/org/apache/linkis/entrance/log/CacheLogReader.scala
@@ -19,7 +19,11 @@ package org.apache.linkis.entrance.log
 
 import org.apache.linkis.common.io.{Fs, FsPath}
 import org.apache.linkis.common.utils.Utils
+import org.apache.linkis.entrance.conf.EntranceConfiguration
+import org.apache.linkis.entrance.exception.LogReadFailedException
 import org.apache.linkis.storage.FSFactory
+import org.apache.linkis.storage.fs.FileSystem
+import org.apache.linkis.storage.utils.StorageUtils
 
 import java.io.{InputStream, IOException}
 import java.util
@@ -36,13 +40,26 @@ class CacheLogReader(logPath: String, charset: String, sharedCache: Cache, user:
   var closed = false
 
   private def createInputStream: InputStream = {
+    if (!logPath.contains(user)) {
+      throw new LogReadFailedException(
+        s"${user} does not have permission to read the path $logPath"
+      )
+    }
+    val fsPath = new FsPath(logPath)
     if (fileSystem == null) lock synchronized {
       if (fileSystem == null) {
-        fileSystem = FSFactory.getFsByProxyUser(new FsPath(logPath), user)
+
+        fileSystem =
+          if (StorageUtils.isHDFSPath(fsPath) && EntranceConfiguration.ENABLE_HDFS_JVM_USER) {
+            FSFactory.getFs(new FsPath(logPath)).asInstanceOf[FileSystem]
+          } else {
+            FSFactory.getFsByProxyUser(new FsPath(logPath), user).asInstanceOf[FileSystem]
+          }
+
         fileSystem.init(new util.HashMap[String, String]())
       }
     }
-    val inputStream: InputStream = fileSystem.read(new FsPath(logPath))
+    val inputStream: InputStream = fileSystem.read(fsPath)
     inputStream
   }
 
diff --git a/linkis-computation-governance/linkis-entrance/src/main/scala/org/apache/linkis/entrance/log/HDFSCacheLogWriter.scala b/linkis-computation-governance/linkis-entrance/src/main/scala/org/apache/linkis/entrance/log/HDFSCacheLogWriter.scala
index 5ac90add66..a272f1e8a1 100644
--- a/linkis-computation-governance/linkis-entrance/src/main/scala/org/apache/linkis/entrance/log/HDFSCacheLogWriter.scala
+++ b/linkis-computation-governance/linkis-entrance/src/main/scala/org/apache/linkis/entrance/log/HDFSCacheLogWriter.scala
@@ -37,11 +37,15 @@ import java.util
 class HDFSCacheLogWriter(logPath: String, charset: String, sharedCache: Cache, user: String)
     extends LogWriter(charset) {
 
-  if (StringUtils.isBlank(logPath))
+  if (StringUtils.isBlank(logPath)) {
     throw new EntranceErrorException(LOGPATH_NOT_NULL.getErrorCode, LOGPATH_NOT_NULL.getErrorDesc)
+  }
 
-  protected var fileSystem =
+  protected var fileSystem = if (EntranceConfiguration.ENABLE_HDFS_JVM_USER) {
+    FSFactory.getFs(new FsPath(logPath)).asInstanceOf[FileSystem]
+  } else {
     FSFactory.getFsByProxyUser(new FsPath(logPath), user).asInstanceOf[FileSystem]
+  }
 
   override protected var outputStream: OutputStream = null
 
diff --git a/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/java/org/apache/linkis/manager/rm/exception/RMErrorCode.java b/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/java/org/apache/linkis/manager/rm/exception/RMErrorCode.java
index b513931414..eb69062d89 100644
--- a/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/java/org/apache/linkis/manager/rm/exception/RMErrorCode.java
+++ b/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/java/org/apache/linkis/manager/rm/exception/RMErrorCode.java
@@ -61,6 +61,8 @@ public enum RMErrorCode implements LinkisErrorCode {
 
   CLUSTER_QUEUE_INSTANCES_INSUFFICIENT(12012, "Insufficient cluster queue instance(集群队列实例不足)"),
 
+  ACROSS_CLUSTER_RULE_FAILED(12012, "across cluster rule failed(跨集群规则失败)"),
+
   ECM_RESOURCE_INSUFFICIENT(11000, "ECM resources are insufficient(ECM 资源不足)"),
 
   ECM_MEMORY_INSUFFICIENT(11001, "ECM memory resources are insufficient(ECM 内存资源不足)"),
diff --git a/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/scala/org/apache/linkis/manager/am/conf/AMConfiguration.scala b/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/scala/org/apache/linkis/manager/am/conf/AMConfiguration.scala
index c730edf0fd..edd3c001b5 100644
--- a/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/scala/org/apache/linkis/manager/am/conf/AMConfiguration.scala
+++ b/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/scala/org/apache/linkis/manager/am/conf/AMConfiguration.scala
@@ -23,6 +23,20 @@ import org.apache.linkis.manager.common.entity.enumeration.MaintainType
 
 object AMConfiguration {
 
+  val YARN_QUEUE_NAME_CONFIG_KEY = "wds.linkis.rm.yarnqueue"
+
+  val ACROSS_CLUSTER_QUEUE_SUFFIX = "queueRuleSuffix"
+
+  val ACROSS_CLUSTER_TASK = "acrossClusterTask"
+
+  val ACROSS_CLUSTER_CPU_THRESHOLD = "CPUThreshold"
+
+  val ACROSS_CLUSTER_MEMORY_THRESHOLD = "MemoryThreshold"
+
+  val ACROSS_CLUSTER_CPU_PERCENTAGE_THRESHOLD = "CPUPercentageThreshold"
+
+  val ACROSS_CLUSTER_MEMORY_PERCENTAGE_THRESHOLD = "MemoryPercentageThreshold"
+
   val ECM_ADMIN_OPERATIONS = CommonVars("wds.linkis.governance.admin.operations", "")
 
   val ENGINE_START_MAX_TIME =
diff --git a/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/scala/org/apache/linkis/manager/am/service/engine/DefaultEngineCreateService.scala b/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/scala/org/apache/linkis/manager/am/service/engine/DefaultEngineCreateService.scala
index 11f622e9fe..78a8adefd2 100644
--- a/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/scala/org/apache/linkis/manager/am/service/engine/DefaultEngineCreateService.scala
+++ b/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/scala/org/apache/linkis/manager/am/service/engine/DefaultEngineCreateService.scala
@@ -293,6 +293,16 @@ class DefaultEngineCreateService
       })
     }
 
+    val queueRuleSuffix = props.get(AMConfiguration.ACROSS_CLUSTER_QUEUE_SUFFIX)
+    if (StringUtils.isNotBlank(queueRuleSuffix)) {
+      val queueName = props.getOrDefault(AMConfiguration.YARN_QUEUE_NAME_CONFIG_KEY, "default")
+      val newQueueName = queueName + "_" + queueRuleSuffix
+      props.put(AMConfiguration.YARN_QUEUE_NAME_CONFIG_KEY, newQueueName)
+      logger.info(
+        s"Switch queues according to queueRule with queue name : $queueName to $newQueueName"
+      )
+    }
+
     val timeoutEngineResourceRequest = TimeoutEngineResourceRequest(
       timeout,
       engineCreateRequest.getUser,
diff --git a/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/scala/org/apache/linkis/manager/label/conf/LabelManagerConf.scala b/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/scala/org/apache/linkis/manager/label/conf/LabelManagerConf.scala
index 1e34f84d2d..12f1a26782 100644
--- a/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/scala/org/apache/linkis/manager/label/conf/LabelManagerConf.scala
+++ b/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/scala/org/apache/linkis/manager/label/conf/LabelManagerConf.scala
@@ -22,6 +22,6 @@ import org.apache.linkis.common.conf.CommonVars
 object LabelManagerConf {
 
   val LONG_LIVED_LABEL =
-    CommonVars("wds.linkis.label.node.long.lived.label.keys", "tenant").getValue
+    CommonVars("wds.linkis.label.node.long.lived.label.keys", "tenant|yarnCluster").getValue
 
 }
diff --git a/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/scala/org/apache/linkis/manager/rm/external/yarn/YarnResourceRequester.scala b/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/scala/org/apache/linkis/manager/rm/external/yarn/YarnResourceRequester.scala
index 8891c98935..a6ed7c7512 100644
--- a/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/scala/org/apache/linkis/manager/rm/external/yarn/YarnResourceRequester.scala
+++ b/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/scala/org/apache/linkis/manager/rm/external/yarn/YarnResourceRequester.scala
@@ -57,12 +57,11 @@ class YarnResourceRequester extends ExternalResourceRequester with Logging {
 
   private val HASTATE_ACTIVE = "active"
 
-  private var provider: ExternalResourceProvider = _
   private val rmAddressMap: util.Map[String, String] = new ConcurrentHashMap[String, String]()
 
-  private def getAuthorizationStr = {
-    val user = this.provider.getConfigMap.getOrDefault("user", "").asInstanceOf[String]
-    val pwd = this.provider.getConfigMap.getOrDefault("pwd", "").asInstanceOf[String]
+  private def getAuthorizationStr(provider: ExternalResourceProvider) = {
+    val user = provider.getConfigMap.getOrDefault("user", "").asInstanceOf[String]
+    val pwd = provider.getConfigMap.getOrDefault("pwd", "").asInstanceOf[String]
     val authKey = user + ":" + pwd
     Base64.getMimeEncoder.encodeToString(authKey.getBytes)
   }
@@ -71,9 +70,7 @@ class YarnResourceRequester extends ExternalResourceRequester with Logging {
       identifier: ExternalResourceIdentifier,
       provider: ExternalResourceProvider
   ): NodeResource = {
-    val rmWebHaAddress = provider.getConfigMap.get("rmWebAddress").asInstanceOf[String]
-    this.provider = provider
-    val rmWebAddress = getAndUpdateActiveRmWebAddress(rmWebHaAddress)
+    val rmWebAddress = getAndUpdateActiveRmWebAddress(provider)
     logger.info(s"rmWebAddress: $rmWebAddress")
     val queueName = identifier.asInstanceOf[YarnResourceIdentifier].getQueueName
 
@@ -87,7 +84,7 @@ class YarnResourceRequester extends ExternalResourceRequester with Logging {
     )
 
     def maxEffectiveHandle(queueValue: Option[JValue]): Option[YarnResource] = {
-      val metrics = getResponseByUrl("metrics", rmWebAddress)
+      val metrics = getResponseByUrl("metrics", rmWebAddress, provider)
       val totalResouceInfoResponse = (
         (metrics \ "clusterMetrics" \ "totalMB").asInstanceOf[JInt].values.toLong,
         (metrics \ "clusterMetrics" \ "totalVirtualCores").asInstanceOf[JInt].values.toLong
@@ -180,7 +177,7 @@ class YarnResourceRequester extends ExternalResourceRequester with Logging {
     def getChildQueuesOfCapacity(resp: JValue) = resp \ "queues" \ "queue"
 
     def getResources() = {
-      val resp = getResponseByUrl("scheduler", rmWebAddress)
+      val resp = getResponseByUrl("scheduler", rmWebAddress, provider)
       val schedulerType =
         (resp \ "scheduler" \ "schedulerInfo" \ "type").asInstanceOf[JString].values
       if ("capacityScheduler".equals(schedulerType)) {
@@ -255,9 +252,8 @@ class YarnResourceRequester extends ExternalResourceRequester with Logging {
       identifier: ExternalResourceIdentifier,
       provider: ExternalResourceProvider
   ): java.util.List[ExternalAppInfo] = {
-    val rmWebHaAddress = provider.getConfigMap.get("rmWebAddress").asInstanceOf[String]
 
-    val rmWebAddress = getAndUpdateActiveRmWebAddress(rmWebHaAddress)
+    val rmWebAddress = getAndUpdateActiveRmWebAddress(provider)
 
     val queueName = identifier.asInstanceOf[YarnResourceIdentifier].getQueueName
 
@@ -273,7 +269,7 @@ class YarnResourceRequester extends ExternalResourceRequester with Logging {
     val realQueueName = "root." + queueName
 
     def getAppInfos(): Array[ExternalAppInfo] = {
-      val resp = getResponseByUrl("apps", rmWebAddress)
+      val resp = getResponseByUrl("apps", rmWebAddress, provider)
       resp \ "apps" \ "app" match {
         case JArray(apps) =>
           val appInfoBuffer = new ArrayBuffer[YarnAppInfo]()
@@ -307,25 +303,29 @@ class YarnResourceRequester extends ExternalResourceRequester with Logging {
 
   override def getResourceType: ResourceType = ResourceType.Yarn
 
-  private def getResponseByUrl(url: String, rmWebAddress: String) = {
+  private def getResponseByUrl(
+      url: String,
+      rmWebAddress: String,
+      provider: ExternalResourceProvider
+  ) = {
     val httpGet = new HttpGet(rmWebAddress + "/ws/v1/cluster/" + url)
     httpGet.addHeader("Accept", "application/json")
-    val authorEnable: Any = this.provider.getConfigMap.get("authorEnable");
+    val authorEnable: Any = provider.getConfigMap.get("authorEnable");
     var httpResponse: HttpResponse = null
     authorEnable match {
       case flag: Boolean =>
         if (flag) {
-          httpGet.addHeader(HttpHeaders.AUTHORIZATION, "Basic " + getAuthorizationStr)
+          httpGet.addHeader(HttpHeaders.AUTHORIZATION, "Basic " + getAuthorizationStr(provider))
         }
       case _ =>
     }
-    val kerberosEnable: Any = this.provider.getConfigMap.get("kerberosEnable");
+    val kerberosEnable: Any = provider.getConfigMap.get("kerberosEnable");
     kerberosEnable match {
       case flag: Boolean =>
         if (flag) {
-          val principalName = this.provider.getConfigMap.get("principalName").asInstanceOf[String]
-          val keytabPath = this.provider.getConfigMap.get("keytabPath").asInstanceOf[String]
-          val krb5Path = this.provider.getConfigMap.get("krb5Path").asInstanceOf[String]
+          val principalName = provider.getConfigMap.get("principalName").asInstanceOf[String]
+          val keytabPath = provider.getConfigMap.get("keytabPath").asInstanceOf[String]
+          val krb5Path = provider.getConfigMap.get("krb5Path").asInstanceOf[String]
           val requestKuu = new RequestKerberosUrlUtils(principalName, keytabPath, krb5Path, false)
           val response =
             requestKuu.callRestUrl(rmWebAddress + "/ws/v1/cluster/" + url, principalName)
@@ -341,8 +341,9 @@ class YarnResourceRequester extends ExternalResourceRequester with Logging {
     parse(EntityUtils.toString(httpResponse.getEntity()))
   }
 
-  def getAndUpdateActiveRmWebAddress(haAddress: String): String = {
+  def getAndUpdateActiveRmWebAddress(provider: ExternalResourceProvider): String = {
     // todo check if it will stuck for many requests
+    val haAddress = provider.getConfigMap.get("rmWebAddress").asInstanceOf[String]
     var activeAddress = rmAddressMap.get(haAddress)
     if (StringUtils.isBlank(activeAddress)) haAddress.intern().synchronized {
       if (StringUtils.isBlank(activeAddress)) {
@@ -356,7 +357,7 @@ class YarnResourceRequester extends ExternalResourceRequester with Logging {
             .split(RMConfiguration.DEFAULT_YARN_RM_WEB_ADDRESS_DELIMITER.getValue)
             .foreach(address => {
               Utils.tryCatch {
-                val response = getResponseByUrl("info", address)
+                val response = getResponseByUrl("info", address, provider)
                 response \ "clusterInfo" \ "haState" match {
                   case state: JString =>
                     if (HASTATE_ACTIVE.equalsIgnoreCase(state.s)) {
@@ -393,12 +394,10 @@ class YarnResourceRequester extends ExternalResourceRequester with Logging {
   override def reloadExternalResourceAddress(
       provider: ExternalResourceProvider
   ): java.lang.Boolean = {
-    if (null == provider) {
-      rmAddressMap.clear()
-    } else {
+    if (null != provider) {
       val rmWebHaAddress = provider.getConfigMap.get("rmWebAddress").asInstanceOf[String]
       rmAddressMap.remove(rmWebHaAddress)
-      getAndUpdateActiveRmWebAddress(rmWebHaAddress)
+      getAndUpdateActiveRmWebAddress(provider)
     }
     true
   }
diff --git a/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/scala/org/apache/linkis/manager/rm/service/impl/DriverAndYarnReqResourceService.scala b/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/scala/org/apache/linkis/manager/rm/service/impl/DriverAndYarnReqResourceService.scala
index bebb34e00c..9a3b214318 100644
--- a/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/scala/org/apache/linkis/manager/rm/service/impl/DriverAndYarnReqResourceService.scala
+++ b/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/scala/org/apache/linkis/manager/rm/service/impl/DriverAndYarnReqResourceService.scala
@@ -17,12 +17,12 @@
 
 package org.apache.linkis.manager.rm.service.impl
 
+import org.apache.linkis.manager.am.conf.AMConfiguration
 import org.apache.linkis.manager.common.constant.RMConstant
 import org.apache.linkis.manager.common.entity.resource._
 import org.apache.linkis.manager.common.entity.resource.ResourceType.DriverAndYarn
 import org.apache.linkis.manager.common.exception.RMWarnException
-import org.apache.linkis.manager.common.protocol.engine.{EngineAskRequest, EngineCreateRequest}
-import org.apache.linkis.manager.label.entity.cluster.ClusterLabel
+import org.apache.linkis.manager.common.protocol.engine.EngineCreateRequest
 import org.apache.linkis.manager.rm.domain.RMLabelContainer
 import org.apache.linkis.manager.rm.exception.RMErrorCode
 import org.apache.linkis.manager.rm.external.service.ExternalResourceService
@@ -76,6 +76,55 @@ class DriverAndYarnReqResourceService(
       throw new RMWarnException(notEnoughMessage._1, notEnoughMessage._2)
     }
 
+    if (engineCreateRequest.getProperties != null) {
+      val user = labelContainer.getUserCreatorLabel.getUser
+      val creator = labelContainer.getUserCreatorLabel.getCreator
+      val properties = engineCreateRequest.getProperties
+      val acrossClusterTask = properties.getOrDefault(AMConfiguration.ACROSS_CLUSTER_TASK, "false")
+      val CPUThreshold = properties.get(AMConfiguration.ACROSS_CLUSTER_CPU_THRESHOLD)
+      val MemoryThreshold = properties.get(AMConfiguration.ACROSS_CLUSTER_MEMORY_THRESHOLD)
+      val CPUPercentageThreshold =
+        properties.get(AMConfiguration.ACROSS_CLUSTER_CPU_PERCENTAGE_THRESHOLD)
+      val MemoryPercentageThreshold =
+        properties.get(AMConfiguration.ACROSS_CLUSTER_MEMORY_PERCENTAGE_THRESHOLD)
+
+      if (
+          StringUtils.isNotBlank(acrossClusterTask) && acrossClusterTask.toBoolean && StringUtils
+            .isNotBlank(CPUThreshold) && StringUtils
+            .isNotBlank(MemoryThreshold)
+          && StringUtils
+            .isNotBlank(CPUPercentageThreshold) && StringUtils.isNotBlank(MemoryPercentageThreshold)
+      ) {
+
+        logger.info(
+          s"user: $user, creator: $creator task enter cross cluster resource judgment, " +
+            s"CPUThreshold: $CPUThreshold, MemoryThreshold: $MemoryThreshold," +
+            s"CPUPercentageThreshold: $CPUPercentageThreshold, MemoryPercentageThreshold: $MemoryPercentageThreshold"
+        )
+        try {
+          AcrossClusterRulesJudgeUtils.acrossClusterRuleCheck(
+            queueLeftResource.asInstanceOf[YarnResource],
+            usedCapacity.asInstanceOf[YarnResource],
+            maxCapacity.asInstanceOf[YarnResource],
+            CPUThreshold.toInt,
+            MemoryThreshold.toInt,
+            CPUPercentageThreshold.toDouble,
+            MemoryPercentageThreshold.toDouble
+          )
+        } catch {
+          case ex: Exception =>
+            throw new RMWarnException(
+            RMErrorCode.ACROSS_CLUSTER_RULE_FAILED.getErrorCode,
+            ex.getMessage
+          )
+        }
+
+        logger.info(s"user: $user, creator: $creator task meet the threshold rule")
+      } else {
+        logger.info(s"user: $user, creator: $creator task skip cross cluster resource judgment")
+      }
+    }
+
     true
   }
 
diff --git a/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/scala/org/apache/linkis/manager/rm/utils/AcrossClusterRulesJudgeUtils.scala b/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/scala/org/apache/linkis/manager/rm/utils/AcrossClusterRulesJudgeUtils.scala
index 51cf36bca1..76fc176395 100644
--- a/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/scala/org/apache/linkis/manager/rm/utils/AcrossClusterRulesJudgeUtils.scala
+++ b/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/scala/org/apache/linkis/manager/rm/utils/AcrossClusterRulesJudgeUtils.scala
@@ -19,47 +19,49 @@ package org.apache.linkis.manager.rm.utils
 
 import org.apache.linkis.common.utils.Logging
 import org.apache.linkis.manager.common.entity.resource.YarnResource
+import org.apache.linkis.manager.common.exception.RMWarnException
+import org.apache.linkis.manager.rm.exception.RMErrorCode
 
 object AcrossClusterRulesJudgeUtils extends Logging {
 
-  def acrossClusterRuleJudge(
+  def acrossClusterRuleCheck(
       leftResource: YarnResource,
       usedResource: YarnResource,
       maxResource: YarnResource,
       leftCPUThreshold: Int,
       leftMemoryThreshold: Int,
-      UsedCPUPercentageThreshold: Double,
-      UsedMemoryPercentageThreshold: Double
-  ): Boolean = {
+      CPUPercentageThreshold: Double,
+      MemoryPercentageThreshold: Double
+  ): Unit = {
     if (leftResource != null && usedResource != null && maxResource != null) {
       val leftQueueMemory = leftResource.queueMemory / Math.pow(1024, 3).toLong
-      logger.info(
-        s"leftResource.queueCores: ${leftResource.queueCores}, leftCPUThreshold: $leftCPUThreshold," +
-          s"leftQueueMemory: $leftQueueMemory, leftMemoryThreshold: $leftMemoryThreshold"
-      )
 
       if (leftResource.queueCores > leftCPUThreshold && leftQueueMemory > leftMemoryThreshold) {
-
         val usedCPUPercentage =
           usedResource.queueCores.asInstanceOf[Double] / maxResource.queueCores
             .asInstanceOf[Double]
         val usedMemoryPercentage = usedResource.queueMemory
           .asInstanceOf[Double] / maxResource.queueMemory.asInstanceOf[Double]
 
-        logger.info(
-          s"usedCPUPercentage: $usedCPUPercentage, UsedCPUPercentageThreshold: $UsedCPUPercentageThreshold" +
-            s"usedMemoryPercentage: $usedMemoryPercentage, UsedMemoryPercentageThreshold: $UsedMemoryPercentageThreshold"
-        )
-
         if (
-            usedCPUPercentage < UsedCPUPercentageThreshold && usedMemoryPercentage < UsedMemoryPercentageThreshold
+            usedCPUPercentage < CPUPercentageThreshold && usedMemoryPercentage < MemoryPercentageThreshold
         ) {
-          return true
+          return
+        } else {
+          throw new RMWarnException(
+            RMErrorCode.ACROSS_CLUSTER_RULE_FAILED.getErrorCode,
+              s"usedCPUPercentage: $usedCPUPercentage, CPUPercentageThreshold: $CPUPercentageThreshold" +
+                s"usedMemoryPercentage: $usedMemoryPercentage, MemoryPercentageThreshold: $MemoryPercentageThreshold"
+          )
         }
+      } else {
+        throw new RMWarnException(
+          RMErrorCode.ACROSS_CLUSTER_RULE_FAILED.getErrorCode,
+          s"leftResource.queueCores: ${leftResource.queueCores}, leftCPUThreshold: $leftCPUThreshold," +
+            s"leftQueueMemory: $leftQueueMemory, leftMemoryThreshold: $leftMemoryThreshold"
+        )
       }
     }
-
-    false
   }
 
 }
diff --git a/linkis-public-enhancements/linkis-configuration/src/main/java/org/apache/linkis/configuration/conf/AcrossClusterConfiguration.scala b/linkis-computation-governance/linkis-manager/linkis-application-manager/src/test/scala/org/apache/linkis/manager/label/conf/LabelManagerConfTest.scala
similarity index 66%
rename from linkis-public-enhancements/linkis-configuration/src/main/java/org/apache/linkis/configuration/conf/AcrossClusterConfiguration.scala
rename to linkis-computation-governance/linkis-manager/linkis-application-manager/src/test/scala/org/apache/linkis/manager/label/conf/LabelManagerConfTest.scala
index f6cd36f69c..836278336b 100644
--- a/linkis-public-enhancements/linkis-configuration/src/main/java/org/apache/linkis/configuration/conf/AcrossClusterConfiguration.scala
+++ b/linkis-computation-governance/linkis-manager/linkis-application-manager/src/test/scala/org/apache/linkis/manager/label/conf/LabelManagerConfTest.scala
@@ -15,13 +15,19 @@
  * limitations under the License.
  */
 
-package org.apache.linkis.configuration.conf
+package org.apache.linkis.manager.label.conf
 
-import org.apache.linkis.common.conf.CommonVars
+import org.junit.jupiter.api.Test
 
-object AcrossClusterConfiguration {
+class LabelManagerConfTest {
 
-  val ACROSS_CLUSTER_QUEUE_SUFFIX =
-    CommonVars.apply("linkis.configuration.across.cluster.queue.suffix", "_bdap2bdp").getValue
+  @Test def testRandomFiltering(): Unit = {
+    var label = "tenant"
+    assert(LabelManagerConf.LONG_LIVED_LABEL.contains(label))
+    label = "yarnCluster"
+    assert(LabelManagerConf.LONG_LIVED_LABEL.contains(label))
+    label = "test"
+    assert(!LabelManagerConf.LONG_LIVED_LABEL.contains(label))
+  }
 
 }
diff --git a/linkis-computation-governance/linkis-manager/linkis-label-common/src/main/java/org/apache/linkis/manager/label/entity/cluster/ClusterLabel.java b/linkis-computation-governance/linkis-manager/linkis-label-common/src/main/java/org/apache/linkis/manager/label/entity/cluster/ClusterLabel.java
index 499adb496b..5e1a3d5c23 100644
--- a/linkis-computation-governance/linkis-manager/linkis-label-common/src/main/java/org/apache/linkis/manager/label/entity/cluster/ClusterLabel.java
+++ b/linkis-computation-governance/linkis-manager/linkis-label-common/src/main/java/org/apache/linkis/manager/label/entity/cluster/ClusterLabel.java
@@ -26,7 +26,7 @@
 
 import java.util.HashMap;
 
-import static org.apache.linkis.manager.label.errorcode.LabelCommonErrorCodeSummary.LABEL_ERROR_CODE;
+import static org.apache.linkis.manager.label.errorcode.LabelCommonErrorCodeSummary.CLUSTER_LABEL_ERROR_CODE;
 
 public class ClusterLabel extends GenericLabel
     implements EMNodeLabel, EngineNodeLabel, UserModifiable {
@@ -75,7 +75,7 @@ public void valueCheck(String stringValue) throws LabelErrorException {
     if (!StringUtils.isEmpty(stringValue)) {
       if (stringValue.split(SerializableLabel.VALUE_SEPARATOR).length != 2) {
         throw new LabelErrorException(
-            LABEL_ERROR_CODE.getErrorCode(), LABEL_ERROR_CODE.getErrorDesc());
+            CLUSTER_LABEL_ERROR_CODE.getErrorCode(), CLUSTER_LABEL_ERROR_CODE.getErrorDesc());
       }
     }
   }
diff --git a/linkis-computation-governance/linkis-manager/linkis-label-common/src/main/java/org/apache/linkis/manager/label/errorcode/LabelCommonErrorCodeSummary.java b/linkis-computation-governance/linkis-manager/linkis-label-common/src/main/java/org/apache/linkis/manager/label/errorcode/LabelCommonErrorCodeSummary.java
index a0ebdffe8a..7ce6c6546d 100644
--- a/linkis-computation-governance/linkis-manager/linkis-label-common/src/main/java/org/apache/linkis/manager/label/errorcode/LabelCommonErrorCodeSummary.java
+++ b/linkis-computation-governance/linkis-manager/linkis-label-common/src/main/java/org/apache/linkis/manager/label/errorcode/LabelCommonErrorCodeSummary.java
@@ -21,6 +21,9 @@
 
 public enum LabelCommonErrorCodeSummary implements LinkisErrorCode {
   UPDATE_LABEL_FAILED(25001, "Update label realtion failed(更新标签属性失败)"),
+  CLUSTER_LABEL_ERROR_CODE(
+      25002,
+      "The value of the label is set incorrectly, the setting value is: ClusterType-ClusterName "),
   LABEL_ERROR_CODE(
       25002,
       "The value of the label is set incorrectly, only one value can be set, and the separator symbol '-' cannot be used(标签的值设置错误,只能设置一个值，不能使用分割符符号 '-') "),
diff --git a/linkis-computation-governance/linkis-manager/linkis-label-common/src/test/java/org/apache/linkis/manager/label/builder/factory/StdLabelBuilderFactoryTest.java b/linkis-computation-governance/linkis-manager/linkis-label-common/src/test/java/org/apache/linkis/manager/label/builder/factory/StdLabelBuilderFactoryTest.java
index 8d0b1e4de9..50f4cb44cb 100644
--- a/linkis-computation-governance/linkis-manager/linkis-label-common/src/test/java/org/apache/linkis/manager/label/builder/factory/StdLabelBuilderFactoryTest.java
+++ b/linkis-computation-governance/linkis-manager/linkis-label-common/src/test/java/org/apache/linkis/manager/label/builder/factory/StdLabelBuilderFactoryTest.java
@@ -19,11 +19,16 @@
 
 import org.apache.linkis.manager.label.builder.CombinedLabelBuilder;
 import org.apache.linkis.manager.label.builder.LabelBuilder;
+import org.apache.linkis.manager.label.entity.Label;
 import org.apache.linkis.manager.label.entity.em.EMInstanceLabel;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.test.context.SpringBootTest;
 
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
 import org.junit.jupiter.api.*;
 
 /** StdLabelBuilderFactory Tester */
@@ -73,4 +78,14 @@ public void testCreateLabelForInLabelKeyInValueStreamOutLabelClassOutValueTypes(
         stdLabelBuilderFactory.createLabel("testLabelKey", null, EMInstanceLabel.class, null);
     Assertions.assertTrue(emInstanceLabel1.getLabelKey().equals("emInstance"));
   }
+
+  @Test
+  public void test() {
+    Map input = new HashMap<String, String>();
+    input.put("userCreator", "username-IDE");
+    input.put("yarnCluster", "bdp-test");
+    input.put("executeOnce", "true");
+    List<Label> res = stdLabelBuilderFactory.getLabels(input);
+    System.out.println(res);
+  }
 }
diff --git a/linkis-dist/package/db/linkis_dml.sql b/linkis-dist/package/db/linkis_dml.sql
index fe7ccfc1ce..1383066cf2 100644
--- a/linkis-dist/package/db/linkis_dml.sql
+++ b/linkis-dist/package/db/linkis_dml.sql
@@ -89,7 +89,7 @@ INSERT INTO `linkis_ps_configuration_config_key` (`key`, `description`, `name`,
 INSERT INTO `linkis_ps_configuration_config_key` (`key`, `description`, `name`, `default_value`, `validate_type`, `validate_range`, `is_hidden`, `is_advanced`, `level`, `treeName`, `engine_conn_type`) VALUES ('spark.tispark.tidb.port', NULL, NULL, '4000', 'None', NULL, '0', '0', '1', 'tidb设置', 'spark');
 INSERT INTO `linkis_ps_configuration_config_key` (`key`, `description`, `name`, `default_value`, `validate_type`, `validate_range`, `is_hidden`, `is_advanced`, `level`, `treeName`, `engine_conn_type`) VALUES ('spark.tispark.tidb.user', NULL, NULL, 'root', 'None', NULL, '0', '0', '1', 'tidb设置', 'spark');
 INSERT INTO `linkis_ps_configuration_config_key` (`key`, `description`, `name`, `default_value`, `validate_type`, `validate_range`, `is_hidden`, `is_advanced`, `level`, `treeName`, `engine_conn_type`) VALUES ('spark.python.version', '取值范围：python2,python3', 'python版本','python2', 'OFT', '[\"python3\",\"python2\"]', '0', '0', '1', 'spark引擎设置', 'spark');
-INSERT INTO `linkis_ps_configuration_config_key` (`key`, `description`, `name`, `default_value`, `validate_type`, `validate_range`, `engine_conn_type`, `is_hidden`, `is_advanced`, `level`, `treeName`, `boundary_type`, `en_treeName`, `en_description`, `en_name`) VALUES ('spark.conf', '多个参数使用分号[;]分隔 例如spark.sql.shuffle.partitions=10;', 'spark自定义配置参数',null, 'None', NULL, 'spark',0, 1, 1,'spark资源设置', 0, 'Spark Resource Settings','Multiple parameters are separated by semicolons [;] For example, spark.sql.shuffle.partitions=10;', 'Spark Custom Configuration Parameters');
+INSERT INTO `linkis_ps_configuration_config_key` (`key`, `description`, `name`, `default_value`, `validate_type`, `validate_range`, `engine_conn_type`, `is_hidden`, `is_advanced`, `level`, `treeName`, `boundary_type`, `en_treeName`, `en_description`, `en_name`) VALUES ('spark.conf', '多个参数使用分号[;]分隔 例如spark.shuffle.spill=true;', 'spark自定义配置参数',null, 'None', NULL, 'spark',0, 1, 1,'spark资源设置', 0, 'Spark Resource Settings','Multiple parameters are separated by semicolons [;] For example, spark.sql.shuffle.partitions=10;', 'Spark Custom Configuration Parameters');
 INSERT INTO `linkis_ps_configuration_config_key` (`key`, `description`, `name`, `default_value`, `validate_type`, `validate_range`, `engine_conn_type`, `is_hidden`, `is_advanced`, `level`, `treeName`, `boundary_type`, `en_treeName`, `en_description`, `en_name`) VALUES ('spark.locality.wait', '范围：0-3000，单位：毫秒', '任务调度本地等待时间', '3000', 'OFT', '[\"0\",\"1000\",\"2000\",\"3000\"]', 'spark', 0, 1, 1, 'spark资源设置', 0, 'Spark Resource Settings', 'Range: 0-3000, Unit: millisecond', 'Task Scheduling Local Waiting Time');
 INSERT INTO `linkis_ps_configuration_config_key` (`key`, `description`, `name`, `default_value`, `validate_type`, `validate_range`, `engine_conn_type`, `is_hidden`, `is_advanced`, `level`, `treeName`, `boundary_type`, `en_treeName`, `en_description`, `en_name`) VALUES ('spark.memory.fraction', '范围：0.4,0.5,0.6，单位：百分比', '执行内存和存储内存的百分比', '0.6', 'OFT', '[\"0.4\",\"0.5\",\"0.6\"]', 'spark', 0, 1, 1, 'spark资源设置', 0, 'Spark Resource Settings', 'Range: 0.4, 0.5, 0.6, in percentage', 'Percentage Of Execution Memory And Storage Memory');
 
@@ -410,6 +410,7 @@ INSERT INTO linkis_ps_error_code (error_code,error_desc,error_regex,error_type)
 INSERT INTO linkis_ps_error_code (error_code,error_desc,error_regex,error_type) VALUES ('13006','Spark context退出，可能是复杂任务导致','Spark application sc has already stopped',0);
 INSERT INTO linkis_ps_error_code (error_code,error_desc,error_regex,error_type) VALUES ('13007','Pyspark子进程意外退出，可能是复杂任务导致','Pyspark process  has stopped',0);
 INSERT INTO linkis_ps_error_code (error_code,error_desc,error_regex,error_type) VALUES ('13008','任务产生的序列化结果总大小超过了配置的spark.driver.maxResultSize限制。请检查您的任务，看看是否有可能减小任务产生的结果大小，或则可以考虑压缩或合并结果，以减少传输的数据量','is bigger than spark.driver.maxResultSize',0);
+INSERT INTO linkis_ps_error_code (error_code,error_desc,error_regex,error_type) VALUES ('13009','您的任务因为引擎退出（退出可能是引擎进程OOM或者主动kill引擎）导致失败','ERROR EC exits unexpectedly and actively kills the task',0);
 
 -- 21 cluster Authority  22 db Authority
 INSERT INTO linkis_ps_error_code (error_code,error_desc,error_regex,error_type) VALUES ('21001','会话创建失败，用户%s不能提交应用到队列：%s，请联系提供队列给您的人员','User (\\S+) cannot submit applications to queue (\\S+)',0);
diff --git a/linkis-dist/package/db/upgrade/1.4.1_schema/mysql/linkis_ddl.sql b/linkis-dist/package/db/upgrade/1.4.1_schema/mysql/linkis_ddl.sql
index 93e5edb384..2974e37c5e 100644
--- a/linkis-dist/package/db/upgrade/1.4.1_schema/mysql/linkis_ddl.sql
+++ b/linkis-dist/package/db/upgrade/1.4.1_schema/mysql/linkis_ddl.sql
@@ -17,16 +17,16 @@
 
 
 
-ALTER TABLE `linkis_ps_udf_user_load` ADD CONSTRAINT  `uniq_uid_uname` UNIQUE (`udf_id`, `user_name`);
-ALTER TABLE `linkis_ps_bml_resources` ADD CONSTRAINT  `uniq_rid_eflag` UNIQUE (`resource_id`, `enable_flag`);
+ALTER TABLE linkis_ps_udf_user_load ADD CONSTRAINT  uniq_uid_uname UNIQUE (`udf_id`, `user_name`);
+ALTER TABLE linkis_ps_bml_resources ADD CONSTRAINT  uniq_rid_eflag UNIQUE (`resource_id`, `enable_flag`);
 
 
-ALTER TABLE `linkis_ps_configuration_config_key` ADD UNIQUE `uniq_key_ectype` (`key`,`engine_conn_type`);
+ALTER TABLE linkis_ps_configuration_config_key ADD UNIQUE uniq_key_ectype (`key`,`engine_conn_type`);
 
-ALTER TABLE `linkis_ps_configuration_config_key` modify column `engine_conn_type` varchar(50) DEFAULT '' COMMENT 'engine type,such as spark,hive etc';
+ALTER TABLE linkis_ps_configuration_config_key modify column engine_conn_type varchar(50) DEFAULT '' COMMENT 'engine type,such as spark,hive etc';
 
 ALTER TABLE linkis_ps_common_lock ADD COLUMN locker VARCHAR(255) NOT NULL COMMENT 'locker';
 
-ALTER TABLE `linkis_ps_configuration_config_key` ADD column `template_required` tinyint(1) DEFAULT 0 COMMENT 'template required 0 none / 1 must'
+ALTER TABLE linkis_ps_configuration_config_key ADD column template_required tinyint(1) DEFAULT 0 COMMENT 'template required 0 none / 1 must'
 
-ALTER TABLE  linkis_ps_configuration_config_value modify COLUMN  config_value varchar(500);
+ALTER TABLE linkis_ps_configuration_config_value modify COLUMN  config_value varchar(500);
diff --git a/linkis-dist/package/db/upgrade/1.4.1_schema/mysql/linkis_dml.sql b/linkis-dist/package/db/upgrade/1.4.1_schema/mysql/linkis_dml.sql
index 8393773757..36d3895b34 100644
--- a/linkis-dist/package/db/upgrade/1.4.1_schema/mysql/linkis_dml.sql
+++ b/linkis-dist/package/db/upgrade/1.4.1_schema/mysql/linkis_dml.sql
@@ -18,6 +18,7 @@
 update  linkis_ps_configuration_config_key set engine_conn_type = "" where engine_conn_type is NULL;
 
 INSERT INTO linkis_ps_error_code (error_code,error_desc,error_regex,error_type) VALUES ('13008','任务产生的序列化结果总大小超过了配置的spark.driver.maxResultSize限制。请检查您的任务，看看是否有可能减小任务产生的结果大小，或则可以考虑压缩或合并结果，以减少传输的数据量','is bigger than spark.driver.maxResultSize',0);
+INSERT INTO linkis_ps_error_code (error_code,error_desc,error_regex,error_type) VALUES ('13009','您的任务因为引擎退出（退出可能是引擎进程OOM或者主动kill引擎）导致失败','ERROR EC exits unexpectedly and actively kills the task',0);
 
 update  linkis_ps_configuration_config_key set template_required = 1 where `key` in (
 "spark.executor.instances",
@@ -36,11 +37,11 @@ is_hidden, is_advanced, `level`,
 treeName, boundary_type, en_treeName,
 en_description, en_name)
 VALUES(
-'spark.conf', '多个参数使用分号[;]分隔 例如spark.sql.shuffle.partitions=10;', 'spark自定义配置参数',
+'spark.conf', '多个参数使用分号[;]分隔 例如spark.shuffle.spill=true;', 'spark自定义配置参数',
 null, 'None', NULL, 'spark',
 0, 1, 1,
 'spark资源设置', 0, 'Spark Resource Settings',
-'Multiple parameters are separated by semicolons [;] For example, spark.sql.shuffle.partitions=10;', 'Spark Custom Configuration Parameters');
+'Multiple parameters are separated by semicolons [;] For example, spark.shuffle.compress=ture;', 'Spark Custom Configuration Parameters');
 
 INSERT INTO `linkis_ps_configuration_key_engine_relation` (`config_key_id`, `engine_type_label_id`)
 (
diff --git a/linkis-dist/package/sbin/kill-process-by-pid.sh b/linkis-dist/package/sbin/kill-process-by-pid.sh
index eea93d03bb..275fe88702 100644
--- a/linkis-dist/package/sbin/kill-process-by-pid.sh
+++ b/linkis-dist/package/sbin/kill-process-by-pid.sh
@@ -1,4 +1,4 @@
-#!/usr/bin/env bash
+#!/bin/bash
 #
 # Licensed to the Apache Software Foundation (ASF) under one or more
 # contributor license agreements.  See the NOTICE file distributed with
@@ -14,68 +14,60 @@
 # limitations under the License.
 #
 
-function list_descendants ()
-{
-  ppid=$1
-  local children=$(pgrep -P ${ppid})
+cd `dirname $0`
+cd ..
+INSTALL_HOME=`pwd`
+# set LINKIS_HOME
+if [ "$LINKIS_HOME" = "" ]; then
+  export LINKIS_HOME=$INSTALL_HOME
+fi
+if [ "$LINKIS_LOG_DIR" = "" ]; then
+  export LINKIS_LOG_DIR="/data/bdp/logs/linkis"
+fi
+ecmPid=`cat $LINKIS_HOME/pid/linkis_cg-engineconnmanager.pid`
+month=`date '+%Y-%m'`
+export killLogFile=$LINKIS_LOG_DIR/linkis-cg-engineconnmanager-kill-$month.log
 
-  for spid in $children
-  do
-    list_descendants "$spid"
-  done
+if [ -f '$killLogFile' ];then
+        echo "logfile exsts."
+else
+        echo "not exists"
+        echo '' >> $killLogFile
+        sudo chown hadoop:hadoop $killLogFile
+        sudo chmod 777 $killLogFile
 
-  sub_pid_list=(${sub_pid_list[@]} ${children[@]})
-}
-function kill_proc_by_pid() {
-    pid=$1
-    sub_pid_list=()
+fi
 
-    echo "`date '+%Y-%m-%d %H:%M:%S'` Starting to kill parent process ${pid}"
+function kill_proc_by_pid() {
+    killProcessPID=$1
+    echo "`date '+%Y-%m-%d %H:%M:%S'` Starting to kill parent process ${killProcessPID}"
+    echo "`date '+%Y-%m-%d %H:%M:%S'` Starting to kill parent process ${killProcessPID}" >> $killLogFile
     #KILL PARENT FIRST
     cnt=0
     MAX_RETRY=6
-    while kill -0 ${pid} > /dev/null 2>&1 && [ $cnt -lt $MAX_RETRY ]
-    do
-        list_descendants ${pid}
-        kill -15 ${pid}
-        let cnt++
-        sleep 5s
-    done
-    if kill -0 ${pid} > /dev/null 2>&1; then
-        list_descendants ${pid}
-        kill -9 ${pid}
-        echo "`date '+%Y-%m-%d %H:%M:%S'` Killed parent process ${pid} with \'-9\' signal. May not be able to kill gracefully(yarn application may still be alive)" 1>&2
-    else
-        echo "`date '+%Y-%m-%d %H:%M:%S'` Finished to kill parent process ${pid}"
-    fi
-
-    #REMOVE DUPLICATE
-    sub_pid_list=($(echo ${sub_pid_list[*]} | sed 's/ /\n/g' | sort | uniq))
-    echo "sub_pid_list:${sub_pid_list[*]}"
-    #THEN KILL CHILDREN
-    for sub_pid in ${sub_pid_list[@]}; do
-        cnt=0
-        MAX_RETRY=6
-        while kill -0 ${sub_pid} > /dev/null 2>&1 && [ $cnt -lt $MAX_RETRY ]
+    if [ "${killProcessPID}" != "${ecmPid}" ]; then
+        #pkill -TERM -P ${killProcessPID}
+        kill -- -$(ps -o pgid "$killProcessPID" | grep [0-9] | tr -d ' ')
+        while ps -p ${killProcessPID} > /dev/null && [ $cnt -lt $MAX_RETRY ]
         do
-            kill -15 ${sub_pid}
-            echo "to Killed sub-process ${sub_pid}"
+            kill -15 ${killProcessPID}
             let cnt++
             sleep 5s
-            if kill -0 ${sub_pid} > /dev/null 2>&1; then
-              let 1+1
-            else
-              echo "`date '+%Y-%m-%d %H:%M:%S'` Killed sub-process ${sub_pid}"
-            fi
         done
-        if kill -0 ${sub_pid} > /dev/null 2>&1; then
-            kill -9 ${sub_pid}
-            echo "`date '+%Y-%m-%d %H:%M:%S'` Killed sub-process ${sub_pid} with \'-9\' signal. May not be able to kill gracefully(yarn application may still be alive)" 1>&2
+        if ps -p ${killProcessPID} > /dev/null; then
+            kill -9 ${killProcessPID}
+            echo "`date '+%Y-%m-%d %H:%M:%S'` Killed parent process ${killProcessPID} with \'-9\' signal. May not be able to kill gracefully(yarn application may still be alive)" 1>&2
+            echo "`date '+%Y-%m-%d %H:%M:%S'` Killed parent process ${killProcessPID} with \'-9\' signal. May not be able to kill gracefully(yarn application may still be alive)"  >> $killLogFile 2>&1
+        else
+            echo "`date '+%Y-%m-%d %H:%M:%S'` Finished to kill parent process ${killProcessPID}"
+            echo "`date '+%Y-%m-%d %H:%M:%S'` Finished to kill parent process ${killProcessPID}" >> $killLogFile 2>&1
         fi
-    done
-
-    echo "`date '+%Y-%m-%d %H:%M:%S'` Finished killing process and its sub-process ${pid}"
+    else
+      echo "ERROR: ECM killProcessPID $ecmPid cannot be kill"
+      echo "ERROR: ECM killProcessPID $ecmPid cannot be kill" >>$killLogFile 2>&1
+    fi
+    echo "`date '+%Y-%m-%d %H:%M:%S'` Finished killing process and its ecmPid ${ecmPid} sub-process ${killProcessPID}"
+    echo "`date '+%Y-%m-%d %H:%M:%S'` Finished killing process and its ecmPid ${ecmPid} sub-process ${killProcessPID}"  >> $killLogFile  2>&1
 }
 
-
 kill_proc_by_pid $@
\ No newline at end of file
diff --git a/linkis-engineconn-plugins/python/src/main/scala/org/apache/linkis/manager/engineplugin/python/executor/PythonSession.scala b/linkis-engineconn-plugins/python/src/main/scala/org/apache/linkis/manager/engineplugin/python/executor/PythonSession.scala
index 79f8f7dc67..dad6a0feb5 100644
--- a/linkis-engineconn-plugins/python/src/main/scala/org/apache/linkis/manager/engineplugin/python/executor/PythonSession.scala
+++ b/linkis-engineconn-plugins/python/src/main/scala/org/apache/linkis/manager/engineplugin/python/executor/PythonSession.scala
@@ -21,6 +21,7 @@ import org.apache.linkis.common.utils.{ByteTimeUtils, Logging, Utils}
 import org.apache.linkis.engineconn.computation.executor.execute.EngineExecutionContext
 import org.apache.linkis.engineconn.computation.executor.rs.RsOutputStream
 import org.apache.linkis.engineconn.launch.EngineConnServer
+import org.apache.linkis.governance.common.utils.GovernanceUtils
 import org.apache.linkis.manager.engineplugin.python.conf.PythonEngineConfiguration
 import org.apache.linkis.manager.engineplugin.python.errorcode.LinkisPythonErrorCodeSummary._
 import org.apache.linkis.manager.engineplugin.python.exception.{
@@ -271,9 +272,15 @@ class PythonSession extends Logging {
       }
       IOUtils.closeQuietly(outputStream)
       Utils.tryAndErrorMsg {
-        pid.foreach(p => Utils.exec(Array("kill", "-9", p), 3000L))
-        process.destroy()
-        process = null
+        // invoke kill process method  to kill all tree process
+        pid.foreach(p => {
+          logger.info(s"Try to kill pyspark process with: [kill -15 ${p}]")
+          GovernanceUtils.killProcess(String.valueOf(p), s"kill pyspark process,pid: $pid", false)
+        })
+        if (pid.isEmpty) {
+          process.destroy()
+          process = null
+        }
         this.pythonScriptInitialized = false
       }("process close failed")
     }
diff --git a/linkis-engineconn-plugins/shell/src/main/scala/org/apache/linkis/manager/engineplugin/shell/executor/ShellEngineConnExecutor.scala b/linkis-engineconn-plugins/shell/src/main/scala/org/apache/linkis/manager/engineplugin/shell/executor/ShellEngineConnExecutor.scala
index a7d321e3aa..bc684ac79a 100644
--- a/linkis-engineconn-plugins/shell/src/main/scala/org/apache/linkis/manager/engineplugin/shell/executor/ShellEngineConnExecutor.scala
+++ b/linkis-engineconn-plugins/shell/src/main/scala/org/apache/linkis/manager/engineplugin/shell/executor/ShellEngineConnExecutor.scala
@@ -309,6 +309,7 @@ class ShellEngineConnExecutor(id: Int) extends ComputationExecutor with Logging
 
   override def close(): Unit = {
     try {
+      super.close()
       process.destroy()
     } catch {
       case e: Exception =>
@@ -316,7 +317,7 @@ class ShellEngineConnExecutor(id: Int) extends ComputationExecutor with Logging
       case t: Throwable =>
         logger.error(s"kill process ${process.toString} failed ", t)
     }
-    super.close()
+
   }
 
 }
diff --git a/linkis-engineconn-plugins/spark/src/main/scala/org/apache/linkis/engineplugin/spark/config/SparkConfiguration.scala b/linkis-engineconn-plugins/spark/src/main/scala/org/apache/linkis/engineplugin/spark/config/SparkConfiguration.scala
index 14d35e23d2..7b224671a9 100644
--- a/linkis-engineconn-plugins/spark/src/main/scala/org/apache/linkis/engineplugin/spark/config/SparkConfiguration.scala
+++ b/linkis-engineconn-plugins/spark/src/main/scala/org/apache/linkis/engineplugin/spark/config/SparkConfiguration.scala
@@ -134,7 +134,7 @@ object SparkConfiguration extends Logging {
 
   val REPLACE_PACKAGE_TO_HEADER = "org.apache.linkis"
 
-  val SPARK_CONF = CommonVars[String]("spark.conf", "")
+  val LINKIS_SPARK_CONF = CommonVars[String]("spark.conf", "")
   val SPARK_APPLICATION_ARGS = CommonVars("spark.app.args", "")
   val SPARK_APPLICATION_MAIN_CLASS = CommonVars("spark.app.main.class", "")
 
diff --git a/linkis-engineconn-plugins/spark/src/main/scala/org/apache/linkis/engineplugin/spark/executor/SparkPythonExecutor.scala b/linkis-engineconn-plugins/spark/src/main/scala/org/apache/linkis/engineplugin/spark/executor/SparkPythonExecutor.scala
index 11964120b3..000c746ae4 100644
--- a/linkis-engineconn-plugins/spark/src/main/scala/org/apache/linkis/engineplugin/spark/executor/SparkPythonExecutor.scala
+++ b/linkis-engineconn-plugins/spark/src/main/scala/org/apache/linkis/engineplugin/spark/executor/SparkPythonExecutor.scala
@@ -32,6 +32,7 @@ import org.apache.linkis.engineplugin.spark.exception.ExecuteError
 import org.apache.linkis.engineplugin.spark.imexport.CsvRelation
 import org.apache.linkis.engineplugin.spark.utils.EngineUtils
 import org.apache.linkis.governance.common.paser.PythonCodeParser
+import org.apache.linkis.governance.common.utils.GovernanceUtils
 import org.apache.linkis.scheduler.executer.{ExecuteResponse, SuccessExecuteResponse}
 import org.apache.linkis.storage.resultset.ResultSetWriter
 
@@ -120,16 +121,15 @@ class SparkPythonExecutor(val sparkEngineSession: SparkEngineSession, val id: In
       }
       IOUtils.closeQuietly(lineOutputStream)
       Utils.tryAndErrorMsg {
-        process.destroy()
-        process = null
-        Thread.sleep(1000 * 2L)
-        // process.destroy will kills the subprocess,not need to force kill with -9,
-        // kill -9 may cause resources not to be released
+        // invoke kill process method  to kill all tree process
         pid.foreach(p => {
           logger.info(s"Try to kill pyspark process with: [kill -15 ${p}]")
-          Utils.exec(Array("kill", "-15", p), 3000L)
+          GovernanceUtils.killProcess(String.valueOf(p), s"kill pyspark process,pid: $pid", false)
         })
-
+        if (pid.isEmpty) {
+          process.destroy()
+          process = null
+        }
       }("process close failed")
     }
     logger.info(s"To delete python executor")
@@ -253,9 +253,7 @@ class SparkPythonExecutor(val sparkEngineSession: SparkEngineSession, val id: In
       //      close
       Utils.tryFinally({
         if (promise != null && !promise.isCompleted) {
-          promise.failure(
-            new ExecuteError(PYSPARK_STOPPED.getErrorCode, PYSPARK_STOPPED.getErrorDesc)
-          )
+          promise.failure(ExecuteError(PYSPARK_STOPPED.getErrorCode, PYSPARK_STOPPED.getErrorDesc))
         }
       }) {
         close
diff --git a/linkis-engineconn-plugins/spark/src/main/scala/org/apache/linkis/engineplugin/spark/launch/SparkSubmitProcessEngineConnLaunchBuilder.scala b/linkis-engineconn-plugins/spark/src/main/scala/org/apache/linkis/engineplugin/spark/launch/SparkSubmitProcessEngineConnLaunchBuilder.scala
index e51b80fdfa..37a0c0e99c 100644
--- a/linkis-engineconn-plugins/spark/src/main/scala/org/apache/linkis/engineplugin/spark/launch/SparkSubmitProcessEngineConnLaunchBuilder.scala
+++ b/linkis-engineconn-plugins/spark/src/main/scala/org/apache/linkis/engineplugin/spark/launch/SparkSubmitProcessEngineConnLaunchBuilder.scala
@@ -21,8 +21,8 @@ import org.apache.linkis.common.conf.CommonVars
 import org.apache.linkis.common.utils.Logging
 import org.apache.linkis.engineplugin.spark.config.SparkConfiguration.{
   ENGINE_JAR,
+  LINKIS_SPARK_CONF,
   SPARK_APP_NAME,
-  SPARK_CONF,
   SPARK_DEFAULT_EXTERNAL_JARS_PATH,
   SPARK_DRIVER_CLASSPATH,
   SPARK_DRIVER_EXTRA_JAVA_OPTIONS,
@@ -63,13 +63,13 @@ class SparkSubmitProcessEngineConnLaunchBuilder(builder: JavaProcessEngineConnLa
     val executorCores = getValueAndRemove(properties, LINKIS_SPARK_EXECUTOR_CORES)
     val executorMemory = getValueAndRemove(properties, LINKIS_SPARK_EXECUTOR_MEMORY)
     val numExecutors = getValueAndRemove(properties, LINKIS_SPARK_EXECUTOR_INSTANCES)
-    val sparkcsonf = getValueAndRemove(properties, SPARK_CONF)
+    val sparkcsonf = getValueAndRemove(properties, LINKIS_SPARK_CONF)
     // sparkcsonf DEMO:spark.sql.shuffle.partitions=10;spark.memory.fraction=0.6
     if (StringUtils.isNotBlank(sparkcsonf)) {
       val strArrary = sparkcsonf.split(";").toList
       strArrary.foreach { keyAndValue =>
-        val key = keyAndValue.split("=")(0)
-        val value = keyAndValue.split("=")(1)
+        val key = keyAndValue.split("=")(0).trim
+        val value = keyAndValue.split("=")(1).trim
         if (StringUtils.isNotBlank(key) && StringUtils.isNotBlank(value)) {
           engineConnBuildRequest.engineConnCreationDesc.properties.put(key, value)
         } else {
diff --git a/linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/ecm/ComputationEngineConnManager.scala b/linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/ecm/ComputationEngineConnManager.scala
index ec18cfb228..58aecefa2f 100644
--- a/linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/ecm/ComputationEngineConnManager.scala
+++ b/linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/ecm/ComputationEngineConnManager.scala
@@ -115,7 +115,9 @@ class ComputationEngineConnManager extends AbstractEngineConnManager with Loggin
             s"${mark.getMarkId()} Failed to askEngineAskRequest time taken ($taken), ${t.getMessage}"
           )
           retryException = t
-        // add isCrossClusterRetryException flag
+          // add isCrossClusterRetryException flag
+          engineAskRequest.getProperties.put("isCrossClusterRetryException", "true")
+
         case t: Throwable =>
           val taken = ByteTimeUtils.msDurationToString(System.currentTimeMillis - start)
           logger.warn(s"${mark.getMarkId()} Failed to askEngineAskRequest time taken ($taken)")
diff --git a/linkis-public-enhancements/linkis-configuration/src/main/java/org/apache/linkis/configuration/conf/AcrossClusterRuleKeys.java b/linkis-public-enhancements/linkis-configuration/src/main/java/org/apache/linkis/configuration/conf/AcrossClusterRuleKeys.java
new file mode 100644
index 0000000000..f2fee2ff1f
--- /dev/null
+++ b/linkis-public-enhancements/linkis-configuration/src/main/java/org/apache/linkis/configuration/conf/AcrossClusterRuleKeys.java
@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.linkis.configuration.conf;
+
+public class AcrossClusterRuleKeys {
+
+  public static final String KEY_QUEUE_SUFFIX = "suffix";
+
+  public static final String KEY_ACROSS_CLUSTER_QUEUE_SUFFIX = "bdap2bdp";
+
+  public static final String KEY_START_TIME = "startTime";
+
+  public static final String KEY_END_TIME = "endTime";
+
+  public static final String KEY_CPU_THRESHOLD = "CPUThreshold";
+
+  public static final String KEY_MEMORY_THRESHOLD = "MemoryThreshold";
+
+  public static final String KEY_CPU_PERCENTAGE_THRESHOLD = "CPUPercentageThreshold";
+
+  public static final String KEY_MEMORY_PERCENTAGE_THRESHOLD = "MemoryPercentageThreshold";
+
+  public static final String KEY_QUEUE_RULE = "queueRule";
+
+  public static final String KEY_TIME_RULE = "timeRule";
+
+  public static final String KEY_THRESHOLD_RULE = "thresholdRule";
+}
diff --git a/linkis-public-enhancements/linkis-configuration/src/main/java/org/apache/linkis/configuration/dao/AcrossClusterRuleMapper.java b/linkis-public-enhancements/linkis-configuration/src/main/java/org/apache/linkis/configuration/dao/AcrossClusterRuleMapper.java
index dc10d5b02d..9dadcf918c 100644
--- a/linkis-public-enhancements/linkis-configuration/src/main/java/org/apache/linkis/configuration/dao/AcrossClusterRuleMapper.java
+++ b/linkis-public-enhancements/linkis-configuration/src/main/java/org/apache/linkis/configuration/dao/AcrossClusterRuleMapper.java
@@ -27,16 +27,17 @@ public interface AcrossClusterRuleMapper {
 
   AcrossClusterRule getAcrossClusterRule(@Param("id") Long id);
 
-  void deleteAcrossClusterRule(@Param("creator") String creator, @Param("user") String user);
+  void deleteAcrossClusterRule(
+      @Param("creator") String creator, @Param("username") String username);
 
   void updateAcrossClusterRule(@Param("acrossClusterRule") AcrossClusterRule acrossClusterRule);
 
   void insertAcrossClusterRule(@Param("acrossClusterRule") AcrossClusterRule acrossClusterRule);
 
   List<AcrossClusterRule> queryAcrossClusterRuleList(
-      @Param("user") String user,
+      @Param("username") String username,
       @Param("creator") String creator,
       @Param("clusterName") String clusterName);
 
-  void validAcrossClusterRule(@Param("acrossClusterRule") AcrossClusterRule acrossClusterRule);
+  void validAcrossClusterRule(@Param("isValid") String isValid, @Param("id") Long id);
 }
diff --git a/linkis-public-enhancements/linkis-configuration/src/main/java/org/apache/linkis/configuration/dao/TemplateConfigKeyMapper.java b/linkis-public-enhancements/linkis-configuration/src/main/java/org/apache/linkis/configuration/dao/TemplateConfigKeyMapper.java
index 6acdd6b825..7dbaef5f02 100644
--- a/linkis-public-enhancements/linkis-configuration/src/main/java/org/apache/linkis/configuration/dao/TemplateConfigKeyMapper.java
+++ b/linkis-public-enhancements/linkis-configuration/src/main/java/org/apache/linkis/configuration/dao/TemplateConfigKeyMapper.java
@@ -41,4 +41,6 @@ List<TemplateConfigKey> selectListByTemplateUuidList(
 
   List<TemplateConfigKeyVO> selectInfoListByTemplateUuid(
       @Param("templateUuid") String templateUuid);
+
+  List<String> selectEngineTypeByTemplateUuid(@Param("templateUuid") String templateUuid);
 }
diff --git a/linkis-public-enhancements/linkis-configuration/src/main/java/org/apache/linkis/configuration/entity/AcrossClusterRule.java b/linkis-public-enhancements/linkis-configuration/src/main/java/org/apache/linkis/configuration/entity/AcrossClusterRule.java
index 3ea0c22d47..c24cfd3d44 100644
--- a/linkis-public-enhancements/linkis-configuration/src/main/java/org/apache/linkis/configuration/entity/AcrossClusterRule.java
+++ b/linkis-public-enhancements/linkis-configuration/src/main/java/org/apache/linkis/configuration/entity/AcrossClusterRule.java
@@ -24,7 +24,7 @@ public class AcrossClusterRule {
   private Long id;
   private String clusterName;
   private String creator;
-  private String user;
+  private String username;
   private Date createTime;
   private String createBy;
   private Date updateTime;
@@ -58,12 +58,12 @@ public void setCreator(String creator) {
     this.creator = creator;
   }
 
-  public String getUser() {
-    return user;
+  public String getUsername() {
+    return username;
   }
 
-  public void setUser(String user) {
-    this.user = user;
+  public void setUsername(String username) {
+    this.username = username;
   }
 
   public Date getCreateTime() {
@@ -125,8 +125,8 @@ public String toString() {
         + ", creator='"
         + creator
         + '\''
-        + ", user='"
-        + user
+        + ", username='"
+        + username
         + '\''
         + ", createTime="
         + createTime
diff --git a/linkis-public-enhancements/linkis-configuration/src/main/java/org/apache/linkis/configuration/entity/ConfigKey.java b/linkis-public-enhancements/linkis-configuration/src/main/java/org/apache/linkis/configuration/entity/ConfigKey.java
index c78b7b83a4..4c471409ab 100644
--- a/linkis-public-enhancements/linkis-configuration/src/main/java/org/apache/linkis/configuration/entity/ConfigKey.java
+++ b/linkis-public-enhancements/linkis-configuration/src/main/java/org/apache/linkis/configuration/entity/ConfigKey.java
@@ -60,7 +60,7 @@ public class ConfigKey {
 
   private String enTreeName;
 
-  private String templateRequired;
+  private Boolean templateRequired;
 
   public String getEngineType() {
     return engineType;
@@ -190,11 +190,11 @@ public void setEnTreeName(String enTreeName) {
     this.enTreeName = enTreeName;
   }
 
-  public String getTemplateRequired() {
+  public Boolean getTemplateRequired() {
     return templateRequired;
   }
 
-  public void setTemplateRequired(String templateRequired) {
+  public void setTemplateRequired(Boolean templateRequired) {
     this.templateRequired = templateRequired;
   }
 
diff --git a/linkis-public-enhancements/linkis-configuration/src/main/java/org/apache/linkis/configuration/restful/api/AcrossClusterRuleRestfulApi.java b/linkis-public-enhancements/linkis-configuration/src/main/java/org/apache/linkis/configuration/restful/api/AcrossClusterRuleRestfulApi.java
index a0ae390576..3a01c86060 100644
--- a/linkis-public-enhancements/linkis-configuration/src/main/java/org/apache/linkis/configuration/restful/api/AcrossClusterRuleRestfulApi.java
+++ b/linkis-public-enhancements/linkis-configuration/src/main/java/org/apache/linkis/configuration/restful/api/AcrossClusterRuleRestfulApi.java
@@ -60,8 +60,8 @@ public class AcrossClusterRuleRestfulApi {
   })
   @RequestMapping(path = "/isValid", method = RequestMethod.PUT)
   public Message isValidRule(HttpServletRequest req, @RequestBody Map<String, Object> json) {
-    String username = ModuleUserUtils.getOperationUser(req, "execute valid acrossClusterRule");
-    if (!Configuration.isAdmin(username)) {
+    String operationUser = ModuleUserUtils.getOperationUser(req, "execute valid acrossClusterRule");
+    if (!Configuration.isAdmin(operationUser)) {
       return Message.error(
           "Failed to valid acrossClusterRule List,msg: only administrators can configure");
     }
@@ -73,10 +73,12 @@ public Message isValidRule(HttpServletRequest req, @RequestBody Map<String, Obje
     if (StringUtils.isBlank(isValid)) {
       return Message.error("Failed to valid acrossClusterRule: Illegal Input Param");
     }
+
     try {
-      acrossClusterRuleService.validAcrossClusterRule(id, isValid, username);
+      acrossClusterRuleService.validAcrossClusterRule(id, isValid);
     } catch (Exception e) {
-      return Message.error("valid acrossClusterRule failed：" + e.getMessage());
+      log.info("valid acrossClusterRule failed：" + e.getMessage());
+      return Message.error("valid acrossClusterRule failed");
     }
 
     return Message.ok();
@@ -89,24 +91,25 @@ public Message isValidRule(HttpServletRequest req, @RequestBody Map<String, Obje
   @ApiImplicitParams({
     @ApiImplicitParam(name = "req", dataType = "HttpServletRequest", value = "req"),
     @ApiImplicitParam(name = "creator", dataType = "String", value = "creator"),
-    @ApiImplicitParam(name = "user", dataType = "String", value = "user"),
+    @ApiImplicitParam(name = "username", dataType = "String", value = "username"),
     @ApiImplicitParam(name = "clusterName", dataType = "String", value = "clusterName"),
   })
   @RequestMapping(path = "/list", method = RequestMethod.GET)
   public Message queryAcrossClusterRuleList(
       HttpServletRequest req,
       @RequestParam(value = "creator", required = false) String creator,
-      @RequestParam(value = "user", required = false) String user,
+      @RequestParam(value = "username", required = false) String username,
       @RequestParam(value = "clusterName", required = false) String clusterName,
       @RequestParam(value = "pageNow", required = false) Integer pageNow,
       @RequestParam(value = "pageSize", required = false) Integer pageSize) {
-    String username = ModuleUserUtils.getOperationUser(req, "execute query acrossClusterRule List");
-    if (!Configuration.isAdmin(username)) {
+    String operationUser =
+        ModuleUserUtils.getOperationUser(req, "execute query acrossClusterRule List");
+    if (!Configuration.isAdmin(operationUser)) {
       return Message.error(
           "Failed to query acrossClusterRule List,msg: only administrators can configure");
     }
 
-    if (StringUtils.isBlank(user)) user = null;
+    if (StringUtils.isBlank(username)) username = null;
     if (StringUtils.isBlank(creator)) creator = null;
     if (StringUtils.isBlank(clusterName)) clusterName = null;
     if (null == pageNow) pageNow = 1;
@@ -116,7 +119,7 @@ public Message queryAcrossClusterRuleList(
     try {
       resultMap =
           acrossClusterRuleService.queryAcrossClusterRuleList(
-              creator, user, clusterName, pageNow, pageSize);
+              creator, username, clusterName, pageNow, pageSize);
     } catch (Exception e) {
       log.info("query acrossClusterRule List failed：" + e.getMessage());
       return Message.error("query acrossClusterRule List failed");
@@ -134,25 +137,26 @@ public Message queryAcrossClusterRuleList(
   @ApiImplicitParams({
     @ApiImplicitParam(name = "req", dataType = "HttpServletRequest", value = "req"),
     @ApiImplicitParam(name = "creator", dataType = "String", value = "creator"),
-    @ApiImplicitParam(name = "user", dataType = "String", value = "user"),
+    @ApiImplicitParam(name = "username", dataType = "String", value = "username"),
   })
   @RequestMapping(path = "/delete", method = RequestMethod.DELETE)
   public Message deleteAcrossClusterRule(
       HttpServletRequest req,
       @RequestParam(value = "creator", required = false) String creator,
-      @RequestParam(value = "user", required = false) String user) {
-    String username = ModuleUserUtils.getOperationUser(req, "execute delete acrossClusterRule");
-    if (!Configuration.isAdmin(username)) {
+      @RequestParam(value = "username", required = false) String username) {
+    String operationUser =
+        ModuleUserUtils.getOperationUser(req, "execute delete acrossClusterRule");
+    if (!Configuration.isAdmin(operationUser)) {
       return Message.error(
           "Failed to delete acrossClusterRule,msg: only administrators can configure");
     }
 
-    if (StringUtils.isBlank(creator) || StringUtils.isBlank(user)) {
+    if (StringUtils.isBlank(creator) || StringUtils.isBlank(username)) {
       return Message.error("Failed to delete acrossClusterRule: Illegal Input Param");
     }
 
     try {
-      acrossClusterRuleService.deleteAcrossClusterRule(creator, user);
+      acrossClusterRuleService.deleteAcrossClusterRule(creator, username);
     } catch (Exception e) {
       log.info("delete acrossClusterRule failed：" + e.getMessage());
       return Message.error("delete acrossClusterRule failed");
@@ -170,7 +174,7 @@ public Message deleteAcrossClusterRule(
     @ApiImplicitParam(name = "id", dataType = "Integer", value = "id"),
     @ApiImplicitParam(name = "clusterName", dataType = "String", value = "clusterName"),
     @ApiImplicitParam(name = "creator", dataType = "String", value = "creator"),
-    @ApiImplicitParam(name = "user", dataType = "String", value = "user"),
+    @ApiImplicitParam(name = "username", dataType = "String", value = "username"),
     @ApiImplicitParam(name = "isValid", dataType = "String", value = "isValid"),
     @ApiImplicitParam(name = "startTime", dataType = "String", value = "startTime"),
     @ApiImplicitParam(name = "endTime", dataType = "String", value = "endTime"),
@@ -188,8 +192,9 @@ public Message deleteAcrossClusterRule(
   @RequestMapping(path = "/update", method = RequestMethod.PUT)
   public Message updateAcrossClusterRule(
       HttpServletRequest req, @RequestBody Map<String, Object> json) {
-    String username = ModuleUserUtils.getOperationUser(req, "execute update acrossClusterRule");
-    if (!Configuration.isAdmin(username)) {
+    String operationUser =
+        ModuleUserUtils.getOperationUser(req, "execute update acrossClusterRule");
+    if (!Configuration.isAdmin(operationUser)) {
       return Message.error(
           "Failed to update acrossClusterRule,msg: only administrators can configure");
     }
@@ -198,7 +203,7 @@ public Message updateAcrossClusterRule(
     Long id = idInt.longValue();
     String clusterName = (String) json.get("clusterName");
     String creator = (String) json.get("creator");
-    String user = (String) json.get("user");
+    String username = (String) json.get("username");
     String isValid = (String) json.get("isValid");
     String startTime = (String) json.get("startTime");
     String endTime = (String) json.get("endTime");
@@ -208,7 +213,7 @@ public Message updateAcrossClusterRule(
     String MemoryPercentageThreshold = (String) json.get("MemoryPercentageThreshold");
     if (StringUtils.isBlank(clusterName)
         || StringUtils.isBlank(creator)
-        || StringUtils.isBlank(user)
+        || StringUtils.isBlank(username)
         || StringUtils.isBlank(isValid)
         || StringUtils.isBlank(startTime)
         || StringUtils.isBlank(endTime)
@@ -232,8 +237,8 @@ public Message updateAcrossClusterRule(
       acrossClusterRule.setId(id);
       acrossClusterRule.setClusterName(clusterName.toLowerCase());
       acrossClusterRule.setCreator(creator);
-      acrossClusterRule.setUser(user);
-      acrossClusterRule.setUpdateBy(username);
+      acrossClusterRule.setUsername(username);
+      acrossClusterRule.setUpdateBy(operationUser);
       acrossClusterRule.setRules(rules);
       acrossClusterRule.setIsValid(isValid);
       acrossClusterRuleService.updateAcrossClusterRule(acrossClusterRule);
@@ -252,7 +257,7 @@ public Message updateAcrossClusterRule(
     @ApiImplicitParam(name = "req", dataType = "HttpServletRequest", value = "req"),
     @ApiImplicitParam(name = "clusterName", dataType = "String", value = "clusterName"),
     @ApiImplicitParam(name = "creator", dataType = "String", value = "creator"),
-    @ApiImplicitParam(name = "user", dataType = "String", value = "user"),
+    @ApiImplicitParam(name = "username", dataType = "String", value = "username"),
     @ApiImplicitParam(name = "isValid", dataType = "String", value = "isValid"),
     @ApiImplicitParam(name = "startTime", dataType = "String", value = "startTime"),
     @ApiImplicitParam(name = "endTime", dataType = "String", value = "endTime"),
@@ -270,15 +275,15 @@ public Message updateAcrossClusterRule(
   @RequestMapping(path = "/add", method = RequestMethod.POST)
   public Message insertAcrossClusterRule(
       HttpServletRequest req, @RequestBody Map<String, Object> json) {
-    String username = ModuleUserUtils.getOperationUser(req, "execute add acrossClusterRule");
-    if (!Configuration.isAdmin(username)) {
+    String operationUser = ModuleUserUtils.getOperationUser(req, "execute add acrossClusterRule");
+    if (!Configuration.isAdmin(operationUser)) {
       return Message.error(
           "Failed to add acrossClusterRule,msg: only administrators can configure");
     }
 
     String clusterName = (String) json.get("clusterName");
     String creator = (String) json.get("creator");
-    String user = (String) json.get("user");
+    String username = (String) json.get("username");
     String isValid = (String) json.get("isValid");
     String startTime = (String) json.get("startTime");
     String endTime = (String) json.get("endTime");
@@ -288,7 +293,7 @@ public Message insertAcrossClusterRule(
     String MemoryPercentageThreshold = (String) json.get("MemoryPercentageThreshold");
     if (StringUtils.isBlank(clusterName)
         || StringUtils.isBlank(creator)
-        || StringUtils.isBlank(user)
+        || StringUtils.isBlank(username)
         || StringUtils.isBlank(isValid)
         || StringUtils.isBlank(startTime)
         || StringUtils.isBlank(endTime)
@@ -311,9 +316,9 @@ public Message insertAcrossClusterRule(
       AcrossClusterRule acrossClusterRule = new AcrossClusterRule();
       acrossClusterRule.setClusterName(clusterName.toLowerCase());
       acrossClusterRule.setCreator(creator);
-      acrossClusterRule.setUser(user);
-      acrossClusterRule.setCreateBy(username);
-      acrossClusterRule.setUpdateBy(username);
+      acrossClusterRule.setUsername(username);
+      acrossClusterRule.setCreateBy(operationUser);
+      acrossClusterRule.setUpdateBy(operationUser);
       acrossClusterRule.setRules(rules);
       acrossClusterRule.setIsValid(isValid);
       acrossClusterRuleService.insertAcrossClusterRule(acrossClusterRule);
diff --git a/linkis-public-enhancements/linkis-configuration/src/main/java/org/apache/linkis/configuration/restful/api/ConfigurationRestfulApi.java b/linkis-public-enhancements/linkis-configuration/src/main/java/org/apache/linkis/configuration/restful/api/ConfigurationRestfulApi.java
index 17e32cebe4..37d24ad3a6 100644
--- a/linkis-public-enhancements/linkis-configuration/src/main/java/org/apache/linkis/configuration/restful/api/ConfigurationRestfulApi.java
+++ b/linkis-public-enhancements/linkis-configuration/src/main/java/org/apache/linkis/configuration/restful/api/ConfigurationRestfulApi.java
@@ -187,11 +187,7 @@ public Message getItemList(
       temp.put("validateRange", configKey.getValidateRange());
       temp.put("boundaryType", configKey.getBoundaryType());
       temp.put("defaultValue", configKey.getDefaultValue());
-      if (StringUtils.isNotBlank(configKey.getTemplateRequired())) {
-        temp.put("require", configKey.getTemplateRequired().equals("1"));
-      } else {
-        temp.put("require", "false");
-      }
+      temp.put("require", configKey.getTemplateRequired());
       filterResult.add(temp);
     }
 
@@ -371,6 +367,7 @@ private void sparkConfCheck(List<ConfigKeyValue> settings, String sparkConf)
       throws ConfigurationException {
     if (StringUtils.isNotBlank(sparkConf)) {
       // Check if there are any duplicates in spark. conf
+      // spark.conf : spark.shuffle.compress=ture;spark.executor.memory=4g
       String[] split = sparkConf.split(";");
       int setSize =
           Arrays.stream(split).map(s -> s.split("=")[0].trim()).collect(Collectors.toSet()).size();
@@ -722,9 +719,6 @@ public Message saveBaseKeyValue(HttpServletRequest req, @RequestBody ConfigKey c
     if (null == boundaryType) {
       return Message.error("boundaryType cannot be empty");
     }
-    if (null == configKey.getTemplateRequired()) {
-      configKey.setTemplateRequired("1");
-    }
     if (StringUtils.isNotEmpty(defaultValue)
         && !validatorManager
             .getOrCreateValidator(validateType)
diff --git a/linkis-public-enhancements/linkis-configuration/src/main/java/org/apache/linkis/configuration/service/AcrossClusterRuleService.java b/linkis-public-enhancements/linkis-configuration/src/main/java/org/apache/linkis/configuration/service/AcrossClusterRuleService.java
index 30588cb1ac..2fff11c871 100644
--- a/linkis-public-enhancements/linkis-configuration/src/main/java/org/apache/linkis/configuration/service/AcrossClusterRuleService.java
+++ b/linkis-public-enhancements/linkis-configuration/src/main/java/org/apache/linkis/configuration/service/AcrossClusterRuleService.java
@@ -23,15 +23,15 @@
 
 public interface AcrossClusterRuleService {
 
-  void deleteAcrossClusterRule(String creator, String user) throws Exception;
+  void deleteAcrossClusterRule(String creator, String username) throws Exception;
 
   void updateAcrossClusterRule(AcrossClusterRule acrossClusterRule) throws Exception;
 
   void insertAcrossClusterRule(AcrossClusterRule acrossClusterRule) throws Exception;
 
   Map<String, Object> queryAcrossClusterRuleList(
-      String creator, String user, String clusterName, Integer pageNow, Integer pageSize)
+      String creator, String username, String clusterName, Integer pageNow, Integer pageSize)
       throws Exception;
 
-  void validAcrossClusterRule(Long id, String isValid, String updateBy) throws Exception;
+  void validAcrossClusterRule(Long id, String isValid) throws Exception;
 }
diff --git a/linkis-public-enhancements/linkis-configuration/src/main/java/org/apache/linkis/configuration/service/impl/AcrossClusterRuleServiceImpl.java b/linkis-public-enhancements/linkis-configuration/src/main/java/org/apache/linkis/configuration/service/impl/AcrossClusterRuleServiceImpl.java
index 377c27fd72..a906ca2d1a 100644
--- a/linkis-public-enhancements/linkis-configuration/src/main/java/org/apache/linkis/configuration/service/impl/AcrossClusterRuleServiceImpl.java
+++ b/linkis-public-enhancements/linkis-configuration/src/main/java/org/apache/linkis/configuration/service/impl/AcrossClusterRuleServiceImpl.java
@@ -39,17 +39,14 @@ public class AcrossClusterRuleServiceImpl implements AcrossClusterRuleService {
   @Autowired private AcrossClusterRuleMapper ruleMapper;
 
   @Override
-  public void deleteAcrossClusterRule(String creator, String user) throws Exception {
-    ruleMapper.deleteAcrossClusterRule(creator, user);
-    logger.info("delete acrossClusterRule success");
-    return;
+  public void deleteAcrossClusterRule(String creator, String username) throws Exception {
+    ruleMapper.deleteAcrossClusterRule(creator, username);
   }
 
   @Override
   public void updateAcrossClusterRule(AcrossClusterRule newRule) throws Exception {
     AcrossClusterRule beforeRule = ruleMapper.getAcrossClusterRule(newRule.getId());
     if (beforeRule == null) {
-      logger.info("acrossClusterRule not exit");
       throw new Exception("acrossClusterRule not exit");
     }
 
@@ -59,8 +56,6 @@ public void updateAcrossClusterRule(AcrossClusterRule newRule) throws Exception
     newRule.setUpdateTime(time);
 
     ruleMapper.updateAcrossClusterRule(newRule);
-    logger.info("update acrossClusterRule success");
-    return;
   }
 
   @Override
@@ -69,13 +64,11 @@ public void insertAcrossClusterRule(AcrossClusterRule acrossClusterRule) throws
     acrossClusterRule.setCreateTime(time);
     acrossClusterRule.setUpdateTime(time);
     ruleMapper.insertAcrossClusterRule(acrossClusterRule);
-    logger.info("insert acrossClusterRule success");
-    return;
   }
 
   @Override
   public Map<String, Object> queryAcrossClusterRuleList(
-      String creator, String user, String clusterName, Integer pageNow, Integer pageSize) {
+      String creator, String username, String clusterName, Integer pageNow, Integer pageSize) {
     Map<String, Object> result = new HashMap<>(2);
     List<AcrossClusterRule> acrossClusterRules = null;
     if (Objects.isNull(pageNow)) {
@@ -87,7 +80,7 @@ public Map<String, Object> queryAcrossClusterRuleList(
     PageHelper.startPage(pageNow, pageSize);
 
     try {
-      acrossClusterRules = ruleMapper.queryAcrossClusterRuleList(user, creator, clusterName);
+      acrossClusterRules = ruleMapper.queryAcrossClusterRuleList(username, creator, clusterName);
     } finally {
       PageHelper.clearPage();
     }
@@ -98,18 +91,13 @@ public Map<String, Object> queryAcrossClusterRuleList(
   }
 
   @Override
-  public void validAcrossClusterRule(Long id, String isValid, String updateBy) throws Exception {
-    AcrossClusterRule acrossClusterRule = ruleMapper.getAcrossClusterRule(id);
-    if (acrossClusterRule == null) {
-      logger.info("acrossClusterRule not exit");
+  public void validAcrossClusterRule(Long id, String isValid) throws Exception {
+    AcrossClusterRule beforeRule = ruleMapper.getAcrossClusterRule(id);
+
+    if (beforeRule == null) {
       throw new Exception("acrossClusterRule not exit");
     }
-    acrossClusterRule.setIsValid(isValid);
-    acrossClusterRule.setUpdateBy(updateBy);
-    acrossClusterRule.setUpdateTime(new Date());
-    logger.info("delete acrossClusterRule success");
-    ruleMapper.validAcrossClusterRule(acrossClusterRule);
-    logger.info("valid acrossClusterRule success");
-    return;
+
+    ruleMapper.validAcrossClusterRule(isValid, id);
   }
 }
diff --git a/linkis-public-enhancements/linkis-configuration/src/main/java/org/apache/linkis/configuration/service/impl/TemplateConfigKeyServiceImpl.java b/linkis-public-enhancements/linkis-configuration/src/main/java/org/apache/linkis/configuration/service/impl/TemplateConfigKeyServiceImpl.java
index 5cbd22287e..b85bd9ba63 100644
--- a/linkis-public-enhancements/linkis-configuration/src/main/java/org/apache/linkis/configuration/service/impl/TemplateConfigKeyServiceImpl.java
+++ b/linkis-public-enhancements/linkis-configuration/src/main/java/org/apache/linkis/configuration/service/impl/TemplateConfigKeyServiceImpl.java
@@ -24,6 +24,7 @@
 import org.apache.linkis.configuration.entity.*;
 import org.apache.linkis.configuration.enumeration.BoundaryTypeEnum;
 import org.apache.linkis.configuration.exception.ConfigurationException;
+import org.apache.linkis.configuration.service.ConfigKeyService;
 import org.apache.linkis.configuration.service.ConfigurationService;
 import org.apache.linkis.configuration.service.TemplateConfigKeyService;
 import org.apache.linkis.configuration.util.LabelEntityParser;
@@ -67,6 +68,8 @@ public class TemplateConfigKeyServiceImpl implements TemplateConfigKeyService {
 
   @Autowired private ConfigurationService configurationService;
 
+  @Autowired private ConfigKeyService configKeyService;
+
   @Autowired private ValidatorManager validatorManager;
 
   @Autowired private ConfigKeyLimitForUserMapper configKeyLimitForUserMapper;
@@ -237,37 +240,67 @@ public List<Object> queryKeyInfoList(List<String> uuidList) throws Configuration
       List<Object> keys = new ArrayList<>();
       item.put("templateUid", uuid);
 
-      List<TemplateConfigKey> group = templateConfigKeyListGroupByUuid.get(uuid);
-      for (TemplateConfigKey templateConfigKey : group) {
-        Map temp = new HashMap();
-
-        temp.put("configValue", templateConfigKey.getConfigValue());
-        temp.put("maxValue", templateConfigKey.getMaxValue());
-        temp.put("createBy", templateConfigKey.getCreateBy());
-        temp.put("createTime", templateConfigKey.getCreateTime());
-        temp.put("updateBy", templateConfigKey.getUpdateBy());
-        temp.put("updateTime", templateConfigKey.getUpdateTime());
-        temp.put("keyId", templateConfigKey.getKeyId());
-
-        ConfigKey info = configKeyMap.get(templateConfigKey.getKeyId());
-        if (info != null) {
-          temp.put("key", info.getKey());
-          temp.put("name", info.getName());
-          temp.put("description", info.getDescription());
-          temp.put("engineType", info.getEngineType());
-          temp.put("validateType", info.getValidateType());
-          temp.put("validateRange", info.getValidateRange());
-          temp.put("boundaryType", info.getBoundaryType());
-          temp.put("defaultValue", info.getDefaultValue());
-          if (StringUtils.isNotBlank(info.getTemplateRequired())) {
-            temp.put("require", info.getTemplateRequired().equals("1"));
-          } else {
-            temp.put("require", "false");
-          }
+      String engineType = "";
+      List<String> engineTypeList = templateConfigKeyMapper.selectEngineTypeByTemplateUuid(uuid);
+
+      if (engineTypeList.size() > 1) {
+        String msg =
+            MessageFormat.format(
+                "template uuid:{0} associated with the engine type:{1} more than one! Please check if the keys are correct",
+                uuid, StringUtils.join(engineTypeList.toArray(), ","));
+        throw new ConfigurationException(msg);
+      }
+
+      if (engineTypeList.size() == 0) {
+        String msg =
+            MessageFormat.format(
+                "template uuid:{0} can not associated with any engine type! Please check if the keys are correct",
+                uuid);
+        throw new ConfigurationException(msg);
+      }
+
+      engineType = engineTypeList.get(0);
+
+      Map<Long, TemplateConfigKey> templateConfigKeyMap =
+          templateConfigKeyListGroupByUuid.get(uuid).stream()
+              .collect(Collectors.toMap(TemplateConfigKey::getKeyId, elemt -> elemt));
+
+      List<ConfigKey> ecKeyList = configKeyService.getConfigKeyList(engineType);
+      for (ConfigKey configKey : ecKeyList) {
+        Map<String, Object> temp = new HashMap<>();
+        temp.put("key", configKey.getKey());
+        temp.put("name", configKey.getName());
+        temp.put("description", configKey.getDescription());
+        temp.put("engineType", configKey.getEngineType());
+        temp.put("validateType", configKey.getValidateType());
+        temp.put("validateRange", configKey.getValidateRange());
+        temp.put("boundaryType", configKey.getBoundaryType());
+        temp.put("defaultValue", configKey.getDefaultValue());
+        temp.put("require", configKey.getTemplateRequired());
+        temp.put("keyId", configKey.getId());
+
+        Long keyId = configKey.getId();
+        TemplateConfigKey templateConfigKey = templateConfigKeyMap.get(keyId);
+
+        if (templateConfigKey == null) {
+          temp.put("configValue", null);
+          temp.put("maxValue", null);
+          temp.put("createBy", null);
+          temp.put("createTime", null);
+          temp.put("updateBy", null);
+          temp.put("updateTime", null);
+        } else {
+          temp.put("configValue", templateConfigKey.getConfigValue());
+          temp.put("maxValue", templateConfigKey.getMaxValue());
+          temp.put("createBy", templateConfigKey.getCreateBy());
+          temp.put("createTime", templateConfigKey.getCreateTime());
+          temp.put("updateBy", templateConfigKey.getUpdateBy());
+          temp.put("updateTime", templateConfigKey.getUpdateTime());
         }
 
         keys.add(temp);
       }
+
       item.put("itemList", keys);
       result.add(item);
     }
diff --git a/linkis-public-enhancements/linkis-configuration/src/main/java/org/apache/linkis/configuration/util/CommonUtils.java b/linkis-public-enhancements/linkis-configuration/src/main/java/org/apache/linkis/configuration/util/CommonUtils.java
index 12e049595a..2d3f9b2008 100644
--- a/linkis-public-enhancements/linkis-configuration/src/main/java/org/apache/linkis/configuration/util/CommonUtils.java
+++ b/linkis-public-enhancements/linkis-configuration/src/main/java/org/apache/linkis/configuration/util/CommonUtils.java
@@ -17,7 +17,7 @@
 
 package org.apache.linkis.configuration.util;
 
-import org.apache.linkis.configuration.conf.AcrossClusterConfiguration;
+import org.apache.linkis.server.BDPJettyServerHelper;
 
 import java.util.HashMap;
 import java.util.Map;
@@ -25,6 +25,8 @@
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 
+import static org.apache.linkis.configuration.conf.AcrossClusterRuleKeys.*;
+
 public class CommonUtils {
   public static boolean ipCheck(String str) {
     if (str != null && !str.isEmpty()) {
@@ -49,17 +51,17 @@ public static String ruleMap2String(
     Map<String, String> timeRuleMap = new HashMap<>();
     Map<String, String> thresholdRuleMap = new HashMap<>();
     Map<String, Object> ruleMap = new HashMap<>();
-    queueRuleMap.put("suffix", AcrossClusterConfiguration.ACROSS_CLUSTER_QUEUE_SUFFIX());
-    timeRuleMap.put("startTime", startTime);
-    timeRuleMap.put("endTime", endTime);
-    thresholdRuleMap.put("CPUThreshold", CPUThreshold);
-    thresholdRuleMap.put("MemoryThreshold", MemoryThreshold);
-    thresholdRuleMap.put("CPUPercentageThreshold", CPUPercentageThreshold);
-    thresholdRuleMap.put("MemoryPercentageThreshold", MemoryPercentageThreshold);
-    ruleMap.put("queueRule", queueRuleMap);
-    ruleMap.put("timeRule", timeRuleMap);
-    ruleMap.put("thresholdRule", thresholdRuleMap);
-    ObjectMapper map2Json = new ObjectMapper();
+    queueRuleMap.put(KEY_QUEUE_SUFFIX, KEY_ACROSS_CLUSTER_QUEUE_SUFFIX);
+    timeRuleMap.put(KEY_START_TIME, startTime);
+    timeRuleMap.put(KEY_END_TIME, endTime);
+    thresholdRuleMap.put(KEY_CPU_THRESHOLD, CPUThreshold);
+    thresholdRuleMap.put(KEY_MEMORY_THRESHOLD, MemoryThreshold);
+    thresholdRuleMap.put(KEY_CPU_PERCENTAGE_THRESHOLD, CPUPercentageThreshold);
+    thresholdRuleMap.put(KEY_MEMORY_PERCENTAGE_THRESHOLD, MemoryPercentageThreshold);
+    ruleMap.put(KEY_QUEUE_RULE, queueRuleMap);
+    ruleMap.put(KEY_TIME_RULE, timeRuleMap);
+    ruleMap.put(KEY_THRESHOLD_RULE, thresholdRuleMap);
+    ObjectMapper map2Json = BDPJettyServerHelper.jacksonJson();
     String rules = map2Json.writeValueAsString(ruleMap);
 
     return rules;
diff --git a/linkis-public-enhancements/linkis-configuration/src/main/resources/mapper/common/AcrossClusterRuleMapper.xml b/linkis-public-enhancements/linkis-configuration/src/main/resources/mapper/common/AcrossClusterRuleMapper.xml
index 6997963313..29ae208a71 100644
--- a/linkis-public-enhancements/linkis-configuration/src/main/resources/mapper/common/AcrossClusterRuleMapper.xml
+++ b/linkis-public-enhancements/linkis-configuration/src/main/resources/mapper/common/AcrossClusterRuleMapper.xml
@@ -24,7 +24,7 @@
         <id property="id" column="id" jdbcType="BIGINT"/>
         <result property="clusterName" column="cluster_name" jdbcType="VARCHAR"/>
         <result property="creator" column="creator" jdbcType="VARCHAR"/>
-        <result property="user" column="user" jdbcType="VARCHAR"/>
+        <result property="username" column="username" jdbcType="VARCHAR"/>
         <result property="createTime" column="create_time" jdbcType="TIMESTAMP"/>
         <result property="createBy" column="create_by" jdbcType="VARCHAR"/>
         <result property="updateTime" column="update_time" jdbcType="TIMESTAMP"/>
@@ -34,17 +34,17 @@
     </resultMap>
 
     <sql id="AcrossClusterRule_Column_List">
-        id,cluster_name,creator,user,create_time,create_by,update_time,update_by,rules,is_valid
+        id,cluster_name,creator,username,create_time,create_by,update_time,update_by,rules,is_valid
     </sql>
 
     <sql id="AcrossClusterRule_insert_Column_List">
-        cluster_name,creator,user,create_time,create_by,update_time,update_by,rules,is_valid
+        cluster_name,creator,username,create_time,create_by,update_time,update_by,rules,is_valid
     </sql>
 
     <select id="getAcrossClusterRule" resultMap="acrossClusterRuleMap">
         SELECT
             acr.id, acr.cluster_name, acr.creator,
-            acr.user, acr.create_time, acr.create_by,
+            acr.username, acr.create_time, acr.create_by,
             acr.update_time, acr.update_by, acr.rules
         FROM
             `linkis_cg_lm_across_cluster_rule` acr
@@ -56,7 +56,7 @@
         SELECT <include refid="AcrossClusterRule_Column_List"/>
         FROM `linkis_cg_lm_across_cluster_rule`
         <where>
-            <if test="user != null"> user = #{user}</if>
+            <if test="username != null"> username = #{username}</if>
             <if test="creator != null">and creator = #{creator}</if>
             <if test="clusterName != null">and cluster_name = #{clusterName}</if>
         </where>
@@ -67,7 +67,7 @@
         INSERT INTO
             linkis_cg_lm_across_cluster_rule (<include refid="AcrossClusterRule_insert_Column_List"/>)
         VALUES
-            (#{acrossClusterRule.clusterName},#{acrossClusterRule.creator}, #{acrossClusterRule.user},
+            (#{acrossClusterRule.clusterName},#{acrossClusterRule.creator}, #{acrossClusterRule.username},
             <choose>
                 <when test="acrossClusterRule.createTime != null">
                     #{acrossClusterRule.createTime}
@@ -93,7 +93,7 @@
         FROM
             `linkis_cg_lm_across_cluster_rule`
         WHERE
-            creator = #{creator} AND user = #{user}
+            creator = #{creator} AND username = #{username}
     </delete>
 
     <update id="updateAcrossClusterRule" parameterType="org.apache.linkis.configuration.entity.AcrossClusterRule">
@@ -101,7 +101,7 @@
             `linkis_cg_lm_across_cluster_rule`
         SET
             cluster_name = #{acrossClusterRule.clusterName}, creator = #{acrossClusterRule.creator},
-            user=#{acrossClusterRule.user}, create_time=#{acrossClusterRule.createTime},
+            username=#{acrossClusterRule.username}, create_time=#{acrossClusterRule.createTime},
             create_By=#{acrossClusterRule.createBy},
             <choose>
                 <when test="acrossClusterRule.updateTime != null">
@@ -117,22 +117,13 @@
             id = #{acrossClusterRule.id}
     </update>
 
-    <update id="validAcrossClusterRule" parameterType="org.apache.linkis.configuration.entity.AcrossClusterRule">
+    <update id="validAcrossClusterRule">
         UPDATE
             `linkis_cg_lm_across_cluster_rule`
         SET
-            <choose>
-                <when test="acrossClusterRule.updateTime != null">
-                    update_time=#{acrossClusterRule.updateTime}
-                </when>
-                <otherwise>
-                    update_time = now()
-                </otherwise>
-            </choose>,
-            update_By=#{acrossClusterRule.updateBy},
-            is_valid=#{acrossClusterRule.isValid}
+            is_valid = #{isValid}
         WHERE
-            id = #{acrossClusterRule.id}
+            id = #{id}
     </update>
 
 
diff --git a/linkis-public-enhancements/linkis-configuration/src/main/resources/mapper/common/ConfigMapper.xml b/linkis-public-enhancements/linkis-configuration/src/main/resources/mapper/common/ConfigMapper.xml
index a7ea542e20..e57b7eed65 100644
--- a/linkis-public-enhancements/linkis-configuration/src/main/resources/mapper/common/ConfigMapper.xml
+++ b/linkis-public-enhancements/linkis-configuration/src/main/resources/mapper/common/ConfigMapper.xml
@@ -51,7 +51,7 @@
         <result column="level" property="level" jdbcType="TINYINT"/>
         <result column="treeName" property="treeName" jdbcType="VARCHAR"/>
         <result column="boundary_type" property="boundaryType" jdbcType="TINYINT"/>
-        <result column="template_required" property="templateRequired" jdbcType="VARCHAR"/>
+        <result column="template_required" property="templateRequired" jdbcType="TINYINT"/>
     </resultMap>
 
     <resultMap id="ConfigValueMap" type="org.apache.linkis.configuration.entity.ConfigValue">
@@ -192,26 +192,27 @@
 
     <select id="selectKeyByEngineType" resultMap="ConfigKeyMap">
 
-        SELECT   `key`, `description`, `name`, `engine_conn_type`,
+        SELECT   `id`,  `key`, `description`, `name`, `engine_conn_type`,
                 `default_value`, `validate_type`, `validate_range`,boundary_type,template_required
         FROM linkis_ps_configuration_config_key
-<!--        filter some unused and duplicate key-->
-        <where>
-            <if test="engineType != null">
-                `engine_conn_type` = #{engineType}
-                AND id  IN
-                (
-                    SELECT temp.config_key_id FROM
-                    ( SELECT * from linkis_ps_configuration_key_engine_relation) AS temp
-                    WHERE  temp.config_key_id IN
-                    (
-                    SELECT B.id FROM
-                    (SELECT * from linkis_ps_configuration_config_key ) AS B
-                    WHERE B. `engine_conn_type` = #{engineType}
-                    )
-                );
-            </if>
-        </where>
+        WHERE `engine_conn_type` = #{engineType}
+<!--&lt;!&ndash;        filter some unused and duplicate key&ndash;&gt;-->
+<!--        <where>-->
+<!--            <if test="engineType != null">-->
+<!--                `engine_conn_type` = #{engineType}-->
+<!--                AND id  IN-->
+<!--                (-->
+<!--                    SELECT temp.config_key_id FROM-->
+<!--                    ( SELECT * from linkis_ps_configuration_key_engine_relation) AS temp-->
+<!--                    WHERE  temp.config_key_id IN-->
+<!--                    (-->
+<!--                    SELECT B.id FROM-->
+<!--                    (SELECT * from linkis_ps_configuration_config_key ) AS B-->
+<!--                    WHERE B. `engine_conn_type` = #{engineType}-->
+<!--                    )-->
+<!--                );-->
+<!--            </if>-->
+<!--        </where>-->
     </select>
 
     <select id="selectKeyByEngineTypeAndKeyList" resultMap="ConfigKeyMap">
diff --git a/linkis-public-enhancements/linkis-configuration/src/main/resources/mapper/common/TemplateConfigKeyMapper.xml b/linkis-public-enhancements/linkis-configuration/src/main/resources/mapper/common/TemplateConfigKeyMapper.xml
index fbc02a3f4f..cc1f4ced39 100644
--- a/linkis-public-enhancements/linkis-configuration/src/main/resources/mapper/common/TemplateConfigKeyMapper.xml
+++ b/linkis-public-enhancements/linkis-configuration/src/main/resources/mapper/common/TemplateConfigKeyMapper.xml
@@ -138,4 +138,14 @@
     where t.template_uuid = #{templateUuid,jdbcType=VARCHAR}
   </select>
 
+
+  <select id="selectEngineTypeByTemplateUuid" resultType="java.lang.String">
+    select distinct(k.`engine_conn_type`) from linkis_ps_configuration_config_key  k
+    where k.id in
+      (
+        select t.key_id from linkis_ps_configuration_template_config_key t
+        where t.template_uuid = #{templateUuid,jdbcType=VARCHAR}
+      )
+  </select>
+
 </mapper>
diff --git a/linkis-public-enhancements/linkis-jobhistory/src/main/java/org/apache/linkis/jobhistory/restful/api/QueryRestfulApi.java b/linkis-public-enhancements/linkis-jobhistory/src/main/java/org/apache/linkis/jobhistory/restful/api/QueryRestfulApi.java
index 870401a560..ff5a912702 100644
--- a/linkis-public-enhancements/linkis-jobhistory/src/main/java/org/apache/linkis/jobhistory/restful/api/QueryRestfulApi.java
+++ b/linkis-public-enhancements/linkis-jobhistory/src/main/java/org/apache/linkis/jobhistory/restful/api/QueryRestfulApi.java
@@ -25,6 +25,7 @@
 import org.apache.linkis.jobhistory.conversions.TaskConversions;
 import org.apache.linkis.jobhistory.entity.*;
 import org.apache.linkis.jobhistory.service.JobHistoryQueryService;
+import org.apache.linkis.jobhistory.transitional.TaskStatus;
 import org.apache.linkis.jobhistory.util.QueryUtils;
 import org.apache.linkis.protocol.constants.TaskConstant;
 import org.apache.linkis.server.Message;
@@ -101,7 +102,7 @@ public Message getTaskByID(HttpServletRequest req, @PathVariable("id") Long jobI
       return Message.error(
           "The corresponding job was not found, or there may be no permission to view the job"
               + "(没有找到对应的job，也可能是没有查看该job的权限)");
-    } else if (taskVO.getStatus().equals("Running")) {
+    } else if (taskVO.getStatus().equals(TaskStatus.Running.toString())) {
       //  任务运行时不显示异常信息(Do not display exception information during task runtime)
       taskVO.setErrCode(null);
       taskVO.setErrDesc(null);
diff --git a/linkis-public-enhancements/linkis-script-dev/linkis-storage-script-dev-server/src/main/java/org/apache/linkis/filesystem/restful/api/FsRestfulApi.java b/linkis-public-enhancements/linkis-script-dev/linkis-storage-script-dev-server/src/main/java/org/apache/linkis/filesystem/restful/api/FsRestfulApi.java
index 1e1aacda49..fa0d1d86b6 100644
--- a/linkis-public-enhancements/linkis-script-dev/linkis-storage-script-dev-server/src/main/java/org/apache/linkis/filesystem/restful/api/FsRestfulApi.java
+++ b/linkis-public-enhancements/linkis-script-dev/linkis-storage-script-dev-server/src/main/java/org/apache/linkis/filesystem/restful/api/FsRestfulApi.java
@@ -91,9 +91,11 @@ public class FsRestfulApi {
    * @param userName
    * @return
    */
-  private boolean checkIsUsersDirectory(String requestPath, String userName) {
+  private boolean checkIsUsersDirectory(String requestPath, String userName, Boolean withAdmin) {
+    // 配置文件默认关闭检查，withadmin默认true，特殊情况传false 开启权限检查（
+    // The configuration file defaults to disable checking, with admin defaulting to true, and in special cases, false is passed to enable permission checking）
     boolean ownerCheck = WorkSpaceConfiguration.FILESYSTEM_PATH_CHECK_OWNER.getValue();
-    if (!ownerCheck) {
+    if (!ownerCheck && withAdmin) {
       LOGGER.debug("not check filesystem owner.");
       return true;
     }
@@ -102,11 +104,14 @@ private boolean checkIsUsersDirectory(String requestPath, String userName) {
         WorkspaceUtil.suffixTuning(HDFS_USER_ROOT_PATH_PREFIX.getValue());
     String hdfsUserRootPathSuffix = HDFS_USER_ROOT_PATH_SUFFIX.getValue();
     String localUserRootPath = WorkspaceUtil.suffixTuning(LOCAL_USER_ROOT_PATH.getValue());
-    String path;
 
     String workspacePath = hdfsUserRootPathPrefix + userName + hdfsUserRootPathSuffix;
     String enginconnPath = localUserRootPath + userName;
-    if (Configuration.isJobHistoryAdmin(userName)) {
+    // 管理员修改其他用户文件目录时，会导致用户无法使用文件，故此优化管理员不能修改(When administrators modify the file directory of other
+    // users,
+    // it will cause users to be unable to use the file, so the optimization administrator cannot
+    // modify it)
+    if (withAdmin && Configuration.isJobHistoryAdmin(userName)) {
       workspacePath = hdfsUserRootPathPrefix;
       enginconnPath = localUserRootPath;
     }
@@ -117,6 +122,10 @@ private boolean checkIsUsersDirectory(String requestPath, String userName) {
     return (requestPath.contains(workspacePath)) || (requestPath.contains(enginconnPath));
   }
 
+  private boolean checkIsUsersDirectory(String requestPath, String userName) {
+    return checkIsUsersDirectory(requestPath, userName, true);
+  }
+
   @ApiOperation(value = "getUserRootPath", notes = "get user root path", response = Message.class)
   @ApiImplicitParams({
     @ApiImplicitParam(name = "pathType", required = false, dataType = "String", value = "path type")
@@ -233,7 +242,7 @@ public Message rename(HttpServletRequest req, @RequestBody JsonNode json)
       PathValidator$.MODULE$.validate(oldDest, userName);
       PathValidator$.MODULE$.validate(newDest, userName);
     }
-    if (!checkIsUsersDirectory(newDest, userName)) {
+    if (!checkIsUsersDirectory(newDest, userName, false)) {
       throw WorkspaceExceptionManager.createException(80010, userName, newDest);
     }
     if (StringUtils.isEmpty(oldDest)) {
diff --git a/tool/dependencies/known-dependencies.txt b/tool/dependencies/known-dependencies.txt
index 06ea1b3362..ab85adce4c 100644
--- a/tool/dependencies/known-dependencies.txt
+++ b/tool/dependencies/known-dependencies.txt
@@ -30,9 +30,7 @@ aspectjweaver-1.9.6.jar
 audience-annotations-0.5.0.jar
 avatica-1.8.0.jar
 avatica-metrics-1.8.0.jar
-avro-1.7.4.jar
 avro-1.7.7.jar
-avro-1.11.0.jar
 bcpkix-jdk15on-1.64.jar
 bcprov-jdk15on-1.64.jar
 bonecp-0.8.0.RELEASE.jar
@@ -52,6 +50,7 @@ chill-java-0.7.6.jar
 chill_2.11-0.7.6.jar
 classgraph-4.1.7.jar
 classmate-1.5.1.jar
+clickhouse-jdbc-0.3.2-patch11.jar
 commons-beanutils-1.9.4.jar
 commons-cli-1.2.jar
 commons-cli-1.3.1.jar
@@ -77,6 +76,7 @@ commons-math3-3.6.1.jar
 commons-net-3.1.jar
 commons-net-3.9.0.jar
 commons-pool-1.6.jar
+commons-pool2-2.8.1.jar
 commons-text-1.10.0.jar
 concurrent-0.191.jar
 config-1.3.3.jar
@@ -89,8 +89,8 @@ datanucleus-api-jdo-4.2.4.jar
 datanucleus-core-4.1.17.jar
 datanucleus-rdbms-4.1.19.jar
 derby-10.14.2.0.jar
-disruptor-3.4.0.jar
 disruptor-3.3.0.jar
+disruptor-3.4.0.jar
 dropwizard-metrics-hadoop-metrics2-reporter-0.1.2.jar
 druid-1.1.22.jar
 eigenbase-properties-1.1.5.jar
@@ -172,6 +172,10 @@ hadoop-mapreduce-client-shuffle-2.7.2.jar
 hadoop-yarn-api-2.7.2.jar
 hadoop-yarn-client-2.7.2.jar
 hadoop-yarn-common-2.7.2.jar
+
+
+decimal( 10 , 2)
+
 hadoop-yarn-registry-2.7.1.jar
 hadoop-yarn-server-common-2.7.2.jar
 hbase-annotations-1.1.1.jar
@@ -269,6 +273,7 @@ jcommander-1.30.jar
 jdbi3-core-3.4.0.jar
 jdbi3-sqlobject-3.4.0.jar
 jdo-api-3.0.1.jar
+jedis-2.9.2.jar
 jersey-apache-client4-1.19.4.jar
 jersey-client-1.19.4.jar
 jersey-client-2.23.1.jar
@@ -303,10 +308,10 @@ jline-2.14.6.jar
 jmxutils-1.19.jar
 jna-5.12.1.jar
 jna-platform-5.12.1.jar
+joda-time-2.10.5.jar
 joda-time-2.3.jar
 joda-time-2.8.1.jar
 joda-time-2.9.3.jar
-joda-time-2.10.5.jar
 jol-core-0.2.jar
 joni-2.1.2.jar
 jpam-1.1.jar
@@ -318,7 +323,6 @@ json4s-core_2.11-3.5.3.jar
 json4s-jackson_2.11-3.5.3.jar
 json4s-scalap_2.11-3.5.3.jar
 jsp-api-2.1.jar
-jsqlparser-1.0.jar
 jsqlparser-4.2.jar
 jsr305-1.3.9.jar
 jsr305-3.0.1.jar
@@ -326,8 +330,8 @@ jsr305-3.0.2.jar
 jsr311-api-1.1.1.jar
 jta-1.1.jar
 jul-to-slf4j-1.7.30.jar
-kafka-clients-2.5.1.jar
 kafka-clients-2.7.0.jar
+kafka-clients-3.4.0.jar
 knife4j-annotations-2.0.9.jar
 knife4j-core-2.0.9.jar
 knife4j-spring-2.0.9.jar
@@ -361,6 +365,7 @@ metrics-jvm-4.1.22.jar
 micrometer-core-1.5.14.jar
 micrometer-registry-prometheus-1.5.14.jar
 minlog-1.3.0.jar
+mongo-java-driver-3.12.8.jar
 mybatis-3.5.6.jar
 mybatis-plus-3.4.1.jar
 mybatis-plus-annotation-3.4.1.jar
@@ -419,13 +424,13 @@ oro-2.0.8.jar
 osgi-resource-locator-1.0.1.jar
 oshi-core-6.2.1.jar
 pagehelper-5.3.1.jar
-paranamer-2.3.jar
 paranamer-2.8.jar
 parquet-hadoop-bundle-1.8.1.jar
 poi-5.2.3.jar
 poi-ooxml-5.2.3.jar
 poi-ooxml-lite-5.2.3.jar
 poi-shared-strings-2.5.6.jar
+postgresql-42.3.8.jar
 presto-client-0.234.jar
 presto-client-1.5.0.jar
 presto-resource-group-managers-0.234.jar
@@ -461,6 +466,9 @@ scala-reflect-2.11.12.jar
 scala-xml_2.11-1.0.5.jar
 scalap-2.11.12.jar
 scopt_2.11-3.5.0.jar
+seatunnel-core-flink-2.1.2.jar
+seatunnel-core-flink-sql-2.1.2.jar
+seatunnel-core-spark-2.1.2.jar
 security-0.191.jar
 security-0.193.jar
 servo-core-0.12.21.jar
@@ -469,11 +477,10 @@ simpleclient_common-0.8.1.jar
 slf4j-api-1.7.30.jar
 slice-0.38.jar
 slider-core-0.90.2-incubating.jar
-snakeyaml-1.33.jar
-snappy-java-1.0.4.1.jar
+snakeyaml-1.33.jar>hadoop-mapreduce-client-core
 snappy-java-1.0.5.jar
-snappy-java-1.1.4.jar
 snappy-java-1.1.7.7.jar
+snappy-java-1.1.8.4.jar
 spring-aop-5.2.22.RELEASE.jar
 spring-beans-5.2.22.RELEASE.jar
 spring-boot-2.3.12.RELEASE.jar
@@ -485,7 +492,6 @@ spring-boot-starter-actuator-2.3.12.RELEASE.jar
 spring-boot-starter-aop-2.3.12.RELEASE.jar
 spring-boot-starter-cache-2.3.12.RELEASE.jar
 spring-boot-starter-freemarker-2.3.12.RELEASE.jar
-spring-boot-starter-jdbc-2.3.12.RELEASE.jar
 spring-boot-starter-jetty-2.3.12.RELEASE.jar
 spring-boot-starter-json-2.3.12.RELEASE.jar
 spring-boot-starter-log4j2-2.3.12.RELEASE.jar
@@ -577,15 +583,5 @@ xmlenc-0.52.jar
 xstream-1.4.20.jar
 zookeeper-3.5.9.jar
 zookeeper-jute-3.5.9.jar
-zstd-jni-1.4.4-7.jar
 zstd-jni-1.4.5-6.jar
-commons-pool2-2.8.1.jar
-jedis-2.9.2.jar
-dss-gateway-support-1.1.1.jar
-seatunnel-core-flink-2.1.2.jar
-seatunnel-core-flink-sql-2.1.2.jar
-seatunnel-core-spark-2.1.2.jar
-mongo-java-driver-3.12.8.jar
-clickhouse-jdbc-0.3.2-patch11.jar
-postgresql-42.3.8.jar
-
+zstd-jni-1.5.2-1.jar
\ No newline at end of file