#AI commit# 开发阶段： 优化spark引擎队列获取方式

Author: v-kkhuang

linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/scala/org/apache/linkis/manager/am/service/engine/DefaultEngineCreateService.scala

@@ -209,7 +209,8 @@ class DefaultEngineCreateService
         engineCreateRequest.getProperties,
         engineCreateRequest.getUser,
         labelFilter.choseEngineLabel(labelList),
-        timeout
+        timeout,
+        isCreateEngine = true
       )
     // 4. request resource
     val resourceTicketId = resourceManager.requestResource(
@@ -390,13 +391,15 @@ class DefaultEngineCreateService
    * @param engineCreateRequest
    * @param labelList
    * @param timeout
+   * @param isCreateEngine
    * @return
    */
   def generateResource(
       props: util.Map[String, String],
       user: String,
       labelList: util.List[Label[_]],
-      timeout: Long
+      timeout: Long,
+      isCreateEngine: Boolean = false
   ): NodeResource = {
     val configProp = engineConnConfigurationService.getConsoleConfiguration(labelList)
     if (null != configProp && configProp.asScala.nonEmpty) {
@@ -408,7 +411,7 @@ class DefaultEngineCreateService
     }
 
     // Smart queue selection (executed before creating YarnResource)
-    performSmartQueueSelection(props, labelList)
+    performSmartQueueSelection(props, labelList, isCreateEngine)
 
     val crossQueue = props.get(AMConfiguration.CROSS_QUEUE)
     if (StringUtils.isNotBlank(crossQueue)) {
@@ -435,139 +438,146 @@ class DefaultEngineCreateService
    */
   private def performSmartQueueSelection(
       properties: util.Map[String, String],
-      labelList: util.List[Label[_]]
+      labelList: util.List[Label[_]],
+      isCreateEngine: Boolean = false
   ): Unit = {
     try {
-      // 1. Get queue configuration
-      val primaryQueue = properties.getOrDefault(AMConfiguration.YARN_QUEUE_NAME_CONFIG_KEY, "").trim
-      val secondaryQueue =
-        properties.getOrDefault(AMConfiguration.SECONDARY_YARN_QUEUE_NAME_CONFIG_KEY, "").trim
-
-      // 2. Get system configuration
-      val enabled = RMConfiguration.SECONDARY_QUEUE_ENABLED.getValue
-      val threshold = RMConfiguration.SECONDARY_QUEUE_THRESHOLD.getValue
-      val supportedEngines = RMConfiguration.SECONDARY_QUEUE_ENGINES.getValue
-        .split(",")
-        .map(_.trim)
-        .map(_.toLowerCase())
-        .toSet
-      val supportedCreators = RMConfiguration.SECONDARY_QUEUE_CREATORS.getValue
-        .split(",")
-        .map(_.trim)
-        .map(_.toUpperCase())
-        .toSet
-
-      logger.info(
-        s"Smart queue config - primary queue: $primaryQueue, secondary queue: $secondaryQueue"
-      )
+      if (isCreateEngine) {
+        // 1. Get queue configuration
+        val primaryQueue =
+          properties.getOrDefault(AMConfiguration.YARN_QUEUE_NAME_CONFIG_KEY, "").trim
+        val secondaryQueue =
+          properties.getOrDefault(AMConfiguration.SECONDARY_YARN_QUEUE_NAME_CONFIG_KEY, "").trim
+
+        // 2. Get system configuration
+        val enabled = RMConfiguration.SECONDARY_QUEUE_ENABLED.getValue
+        val threshold = RMConfiguration.SECONDARY_QUEUE_THRESHOLD.getValue
+        val supportedEngines = RMConfiguration.SECONDARY_QUEUE_ENGINES.getValue
+          .split(",")
+          .map(_.trim)
+          .map(_.toLowerCase())
+          .toSet
+        val supportedCreators = RMConfiguration.SECONDARY_QUEUE_CREATORS.getValue
+          .split(",")
+          .map(_.trim)
+          .map(_.toUpperCase())
+          .toSet
 
-      // 3. Check if secondary queue feature is enabled
-      if (!enabled || StringUtils.isBlank(secondaryQueue) || StringUtils.isBlank(primaryQueue)) {
         logger.info(
-          "Smart queue selection is not enabled or secondary queue is empty, using primary queue"
+          s"Smart queue config - primary queue: $primaryQueue, secondary queue: $secondaryQueue"
         )
-        return
-      }
 
-      // 4. Get engine type and Creator
-      var engineType: String = null
-      var creator: String = null
-
-      try {
-        if (labelList != null && !labelList.isEmpty) {
-          engineType = LabelUtil.getEngineType(labelList)
-          val userCreatorLabel = labelList.asScala
-            .find(_.isInstanceOf[UserCreatorLabel])
-            .map(_.asInstanceOf[UserCreatorLabel])
-            .orNull
-          if (userCreatorLabel != null) {
-            creator = userCreatorLabel.getCreator
-          }
+        // 3. Check if secondary queue feature is enabled
+        if (!enabled || StringUtils.isBlank(secondaryQueue) || StringUtils.isBlank(primaryQueue)) {
+          logger.info(
+            "Smart queue selection is not enabled or secondary queue is empty, using primary queue"
+          )
+          return
         }
-      } catch {
-        case e: Exception =>
-          logger.error("Failed to parse labels for queue selection", e)
-      }
 
-      // 5. Check if engine type and Creator are in supported list
-      val engineMatched = engineType == null || supportedEngines.contains(engineType.toLowerCase())
-      val creatorMatched = creator == null || supportedCreators.contains(creator.toUpperCase())
+        // 4. Get engine type and Creator
+        var engineType: String = null
+        var creator: String = null
+
+        try {
+          if (labelList != null && !labelList.isEmpty) {
+            engineType = LabelUtil.getEngineType(labelList)
+            val userCreatorLabel = labelList.asScala
+              .find(_.isInstanceOf[UserCreatorLabel])
+              .map(_.asInstanceOf[UserCreatorLabel])
+              .orNull
+            if (userCreatorLabel != null) {
+              creator = userCreatorLabel.getCreator
+            }
+          }
+        } catch {
+          case e: Exception =>
+            logger.error("Failed to parse labels for queue selection", e)
+        }
 
-      if (!engineMatched || !creatorMatched) {
-        logger.info(
-          s"Engine type or Creator not in supported list - engineType: $engineType (matched: $engineMatched), creator: $creator (matched: $creatorMatched)"
-        )
-        return
-      }
+        // 5. Check if engine type and Creator are in supported list
+        val engineMatched =
+          engineType == null || supportedEngines.contains(engineType.toLowerCase())
+        val creatorMatched = creator == null || supportedCreators.contains(creator.toUpperCase())
 
-      // 6. Query secondary queue resource usage
-      try {
-        val labelContainer = labelResourceService.enrichLabels(labelList)
+        if (!engineMatched || !creatorMatched) {
+          logger.info(
+            s"Engine type or Creator not in supported list - engineType: $engineType (matched: $engineMatched), creator: $creator (matched: $creatorMatched)"
+          )
+          return
+        }
 
-        val yarnResourceIdentifier = new YarnResourceIdentifier(secondaryQueue)
-        val queueInfo = externalResourceService.getResource(
-          ResourceType.Yarn,
-          labelContainer,
-          yarnResourceIdentifier
-        )
+        // 6. Query secondary queue resource usage
+        try {
+          val labelContainer = labelResourceService.enrichLabels(labelList)
 
-        if (queueInfo != null) {
-          val usedResource = queueInfo.getUsedResource.asInstanceOf[YarnResource]
-          val maxResource = queueInfo.getMaxResource.asInstanceOf[YarnResource]
+          val yarnResourceIdentifier = new YarnResourceIdentifier(secondaryQueue)
+          val queueInfo = externalResourceService.getResource(
+            ResourceType.Yarn,
+            labelContainer,
+            yarnResourceIdentifier
+          )
 
-          // 7. Three-dimensional independent judgment
-          val useSecondaryQueue = if (maxResource != null && maxResource.getQueueMemory > 0) {
-            val memoryUsage =
-              usedResource.getQueueMemory.toDouble / maxResource.getQueueMemory.toDouble
-            val cpuUsage = if (maxResource.getQueueCores > 0) {
-              usedResource.getQueueCores.toDouble / maxResource.getQueueCores.toDouble
-            } else {
-              0.0
-            }
-            val instanceUsage = if (maxResource.getQueueInstances > 0) {
-              usedResource.getQueueInstances.toDouble / maxResource.getQueueInstances.toDouble
+          if (queueInfo != null) {
+            val usedResource = queueInfo.getUsedResource.asInstanceOf[YarnResource]
+            val maxResource = queueInfo.getMaxResource.asInstanceOf[YarnResource]
+
+            // 7. Three-dimensional independent judgment
+            val useSecondaryQueue = if (maxResource != null && maxResource.getQueueMemory > 0) {
+              val memoryUsage =
+                usedResource.getQueueMemory.toDouble / maxResource.getQueueMemory.toDouble
+              val cpuUsage = if (maxResource.getQueueCores > 0) {
+                usedResource.getQueueCores.toDouble / maxResource.getQueueCores.toDouble
+              } else {
+                0.0
+              }
+              val instanceUsage = if (maxResource.getQueueInstances > 0) {
+                usedResource.getQueueInstances.toDouble / maxResource.getQueueInstances.toDouble
+              } else {
+                0.0
+              }
+              // Do not use secondary queue if any dimension exceeds threshold
+              val memoryOverThreshold = memoryUsage > threshold
+              val cpuOverThreshold = cpuUsage > threshold
+              val instanceOverThreshold = instanceUsage > threshold
+
+              if (memoryOverThreshold || cpuOverThreshold || instanceOverThreshold) {
+                logger.info(
+                  s"Secondary queue resource usage- memory: $memoryOverThreshold, cpu: $cpuOverThreshold, instance: $instanceOverThreshold, using primary queue"
+                )
+                false
+              } else {
+                logger.info("Secondary queue has sufficient resources, using secondary queue")
+                true
+              }
             } else {
-              0.0
-            }
-            // Do not use secondary queue if any dimension exceeds threshold
-            val memoryOverThreshold = memoryUsage > threshold
-            val cpuOverThreshold = cpuUsage > threshold
-            val instanceOverThreshold = instanceUsage > threshold
-
-            if (memoryOverThreshold || cpuOverThreshold || instanceOverThreshold) {
-              logger.info(
-                s"Secondary queue resource usage- memory: $memoryOverThreshold, cpu: $cpuOverThreshold, instance: $instanceOverThreshold, using primary queue"
-              )
+              logger.warn("Secondary queue max resource is empty, using primary queue")
               false
-            } else {
-              logger.info("Secondary queue has sufficient resources, using secondary queue")
-              true
             }
-          } else {
-            logger.warn("Secondary queue max resource is empty, using primary queue")
-            false
-          }
 
-          // 8. Determine which queue to use and update wds.linkis.rm.yarnqueue
-          val selectedQueue = if (useSecondaryQueue) secondaryQueue else primaryQueue
-          val oldQueue = properties.get(AMConfiguration.YARN_QUEUE_NAME_CONFIG_KEY)
-          properties.put(AMConfiguration.YARN_QUEUE_NAME_CONFIG_KEY, selectedQueue)
+            // 8. Determine which queue to use and update wds.linkis.rm.yarnqueue
+            val selectedQueue = if (useSecondaryQueue) secondaryQueue else primaryQueue
+            val oldQueue = properties.get(AMConfiguration.YARN_QUEUE_NAME_CONFIG_KEY)
+            properties.put(AMConfiguration.YARN_QUEUE_NAME_CONFIG_KEY, selectedQueue)
 
-          logger.info(
-            s"Smart queue selection completed - original queue: $oldQueue, selected queue: $selectedQueue"
-          )
+            logger.info(
+              s"Smart queue selection completed - original queue: $oldQueue, selected queue: $selectedQueue"
+            )
 
-        } else {
-          logger.warn(
-            s"Unable to get secondary queue $secondaryQueue information, using primary queue: $primaryQueue"
-          )
-        }
+          } else {
+            logger.warn(
+              s"Unable to get secondary queue $secondaryQueue information, using primary queue: $primaryQueue"
+            )
+          }
 
-      } catch {
-        case e: Exception =>
-          logger.error(s"Exception in smart queue selection, using primary queue: $primaryQueue", e)
+        } catch {
+          case e: Exception =>
+            logger.error(
+              s"Exception in smart queue selection, using primary queue: $primaryQueue",
+              e
+            )
+        }
       }
-
     } catch {
       case e: Exception =>
         logger.error(