apache · zhang-arvin · Sep 8, 2026
diff --git a/...is-entrance/src/main/java/org/apache/linkis/entrance/restful/EntranceLabelRestfulApi.java b/...is-entrance/src/main/java/org/apache/linkis/entrance/restful/EntranceLabelRestfulApi.java
@@ -21,6 +21,7 @@
 import org.apache.linkis.common.ServiceInstance;
 import org.apache.linkis.common.conf.Configuration;
 import org.apache.linkis.entrance.EntranceServer;
+import org.apache.linkis.entrance.orchestrator.EntranceOrchestrationFactory;
 import org.apache.linkis.entrance.scheduler.EntranceSchedulerContext;
 import org.apache.linkis.instance.label.client.InstanceLabelClient;
 import org.apache.linkis.manager.label.constant.LabelKeyConstant;
@@ -106,7 +107,17 @@ public Message updateRouteLabel(HttpServletRequest req) {
     SchedulerContext schedulerContext =
         entranceServer.getEntranceContext().getOrCreateScheduler().getSchedulerContext();
     if (schedulerContext instanceof EntranceSchedulerContext) {
-      ((EntranceSchedulerContext) schedulerContext).setOfflineFlag(true);
+      EntranceSchedulerContext entranceSchedulerContext = (EntranceSchedulerContext) schedulerContext;
+      entranceSchedulerContext.setOfflineFlag(true);
+      if (entranceSchedulerContext.getOrCreateGroupFactory()
+          instanceof org.apache.linkis.entrance.scheduler.EntranceGroupFactory) {
+        ((org.apache.linkis.entrance.scheduler.EntranceGroupFactory)
+                entranceSchedulerContext.getOrCreateGroupFactory())
+            .refreshAllGroups();
+        invalidateUserParallelCache();
+        logger.info(
+            "Refreshed scheduler group cache and user parallel plugin cache after mark offline.");
+      }
     }
     entranceServer.updateAllNotExecutionTaskInstances(true);
     logger.info("Finished to update all not execution task instances to empty string");
@@ -128,10 +139,29 @@ public Message backOnline(HttpServletRequest req) {
     synchronized (offlineFlag) { // NOSONAR
       offlineFlag = false;
     }
+    if (entranceServer.getEntranceContext().getOrCreateScheduler().getSchedulerContext()
+        instanceof EntranceSchedulerContext) {
+      EntranceSchedulerContext entranceSchedulerContext =
+          (EntranceSchedulerContext)
+              entranceServer.getEntranceContext().getOrCreateScheduler().getSchedulerContext();
+      if (entranceSchedulerContext.getOrCreateGroupFactory()
+          instanceof org.apache.linkis.entrance.scheduler.EntranceGroupFactory) {
+        ((org.apache.linkis.entrance.scheduler.EntranceGroupFactory)
+                entranceSchedulerContext.getOrCreateGroupFactory())
+            .refreshAllGroups();
+        invalidateUserParallelCache();
+        logger.info(
+            "Refreshed scheduler group cache and user parallel plugin cache after back online.");
+      }
+    }
     logger.info("Finished to backonline");
     return Message.ok();
   }
 
+  private void invalidateUserParallelCache() {
+    new EntranceOrchestrationFactory().invalidateUserParallelPluginCache();
+  }
+
   @ApiOperation(value = "isOnline", notes = "entrance isOnline", response = Message.class)
   @RequestMapping(path = "/isOnline", method = RequestMethod.GET)
   public Message isOnline(HttpServletRequest req) {

diff --git a/...src/main/scala/org/apache/linkis/entrance/orchestrator/EntranceOrchestrationFactory.scala b/...src/main/scala/org/apache/linkis/entrance/orchestrator/EntranceOrchestrationFactory.scala
@@ -52,4 +52,27 @@ object EntranceOrchestrationFactory {
 
   def getOrchestrationSession(): OrchestratorSession = orchestratorSession
 
+  /**
+   * Invalidate the cached user max running jobs so that they are recalculated with the latest
+   * number of running entrance instances (e.g. after entrances go offline/back online).
+   */
+  def invalidateUserParallelPluginCache(): Unit =
+    orchestratorSession.orchestrator.getOrchestratorContext match {
+      case ctx: AbstractOrchestratorContext =>
+        ctx.getOrchestratorPlugins.foreach {
+          case plugin: EntranceUserParallelOrchestratorPlugin => plugin.invalidateCache()
+          case _                                             =>
+        }
+      case _ =>
+    }
+
+}
+
+/**
+ * Java-visible companion of the `EntranceOrchestrationFactory` object. Scala 2.11 does not emit
+ * static forwarders for objects, so Java callers route through an instance method here.
+ */
+class EntranceOrchestrationFactory {
+  def invalidateUserParallelPluginCache(): Unit =
+    EntranceOrchestrationFactory.invalidateUserParallelPluginCache()
 }
diff --git a/...g/apache/linkis/entrance/orchestrator/plugin/EntranceUserParallelOrchestratorPlugin.scala b/...g/apache/linkis/entrance/orchestrator/plugin/EntranceUserParallelOrchestratorPlugin.scala
@@ -97,6 +97,12 @@ class EntranceUserParallelOrchestratorPlugin extends UserParallelOrchestratorPlu
     configCache.get(EntranceUtils.getUserCreatorEcTypeKey(userCreatorLabel, engineTypeLabel))
   }
 
+  /**
+   * Invalidate the cached user max running jobs, so that they are recalculated with the latest
+   * number of running entrance instances (e.g. after entrances go offline/back online).
+   */
+  def invalidateCache(): Unit = this.configCache.invalidateAll()
+
   override def isReady: Boolean = true
 
   override def start(): Unit = {}

diff --git a/...s-entrance/src/main/scala/org/apache/linkis/entrance/scheduler/EntranceGroupFactory.scala b/...s-entrance/src/main/scala/org/apache/linkis/entrance/scheduler/EntranceGroupFactory.scala
@@ -28,6 +28,8 @@ import org.apache.linkis.governance.common.protocol.conf.{
   RequestQueryEngineConfigWithGlobalConfig,
   ResponseQueryConfig
 }
+import org.apache.linkis.manager.label.builder.factory.LabelBuilderFactoryContext
+import org.apache.linkis.manager.label.constant.LabelKeyConstant
 import org.apache.linkis.manager.label.entity.Label
 import org.apache.linkis.manager.label.entity.engine.{EngineTypeLabel, UserCreatorLabel}
 import org.apache.linkis.manager.label.utils.LabelUtil
@@ -44,14 +46,38 @@ import java.util.regex.Pattern
 
 import com.google.common.cache.{Cache, CacheBuilder}
 
+import scala.collection.mutable
+import scala.collection.JavaConverters._
+
 class EntranceGroupFactory extends GroupFactory with Logging {
 
+  case class GroupMaxRunningJobsKey(creator: String, user: String, engineType: String)
+
+  private val labelFactory = LabelBuilderFactoryContext.getLabelBuilderFactory
+
   private val groupNameToGroups: Cache[String, Group] = CacheBuilder
     .newBuilder()
     .expireAfterAccess(EntranceConfiguration.GROUP_CACHE_EXPIRE_TIME.getValue, TimeUnit.MINUTES)
     .maximumSize(EntranceConfiguration.GROUP_CACHE_MAX.getValue)
     .build()
 
+  /** creator_user_engineType -> GroupMaxRunningJobsKey, used to recalculate maxRunningJobs */
+  private val groupNameToKey: mutable.Map[String, GroupMaxRunningJobsKey] = mutable.Map()
+
+  private def getGroupMaxRunningJobsKey(labels: java.util.List[Label[_]])
+      : GroupMaxRunningJobsKey = {
+    val userCreatorLabel: UserCreatorLabel = LabelUtil.getUserCreatorLabel(labels)
+    val engineTypeLabel: EngineTypeLabel = LabelUtil.getEngineTypeLabel(labels)
+    if (null == userCreatorLabel || null == engineTypeLabel) {
+      throw new EntranceErrorException(LABEL_NOT_NULL.getErrorCode, LABEL_NOT_NULL.getErrorDesc)
+    }
+    GroupMaxRunningJobsKey(
+      userCreatorLabel.getCreator,
+      userCreatorLabel.getUser,
+      engineTypeLabel.getEngineType
+    )
+  }
+
   private val GROUP_MAX_CAPACITY = CommonVars("wds.linkis.entrance.max.capacity", 1000)
 
   private val SPECIFIED_USERNAME_REGEX =
@@ -120,13 +146,53 @@ class EntranceGroupFactory extends GroupFactory with Logging {
       group.setMaxRunningJobs(maxRunningJobs)
       group.setMaxAskExecutorTimes(maxAskExecutorTimes)
       groupNameToGroups.put(groupName, group)
+      groupNameToKey.put(groupName, getGroupMaxRunningJobsKey(labels))
       group
     }
     else {
       cacheGroup
     }
   }
 
+  /**
+   * Recalculate the maxRunningJobs of a cached group with the latest user configuration and the
+   * latest number of running entrance instances, and set it to the group.
+   * @param key
+   * @return
+   */
+  private def refreshMaxRunningJobs(key: GroupMaxRunningJobsKey): Unit = {
+    val groupName = s"${key.creator}_${key.user}_${key.engineType}"
+    val group = groupNameToGroups.getIfPresent(groupName)
+    if (null == group) return
+    val labels = labelFactory.createLabel[UserCreatorLabel](LabelKeyConstant.USER_CREATOR_TYPE_KEY)
+    labels.setCreator(key.creator)
+    labels.setUser(key.user)
+    val ecTypeLabel = labelFactory.createLabel[EngineTypeLabel](LabelKeyConstant.ENGINE_TYPE_KEY)
+    ecTypeLabel.setEngineType(key.engineType)
+    val sender: Sender =
+      Sender.getSender(Configuration.CLOUD_CONSOLE_CONFIGURATION_SPRING_APPLICATION_NAME.getValue)
+    val keyAndValue = Utils.tryAndWarnMsg {
+      sender
+        .ask(RequestQueryEngineConfigWithGlobalConfig(labels, ecTypeLabel))
+        .asInstanceOf[ResponseQueryConfig]
+        .getKeyAndValue
+    }(
+      "Get user configurations from configuration server failed! Next use the default value to continue."
+    )
+    val maxRunningJobs = EntranceGroupFactory.getUserMaxRunningJobs(keyAndValue)
+    group.asInstanceOf[ParallelGroup].setMaxRunningJobs(maxRunningJobs)
+    logger.info(s"Refreshed maxRunningJobs of group $groupName to $maxRunningJobs.")
+  }
+
+  /**
+   * Refresh maxRunningJobs for all cached groups, so that after entrances go offline/back online
+   * the concurrency is recalculated with the latest number of running entrance instances.
+   */
+  def refreshAllGroups(): Unit =
+    groupNameToGroups.asMap().keySet().asScala.foreach { groupName =>
+      groupNameToKey.get(groupName).foreach(refreshMaxRunningJobs)
+    }
+
   override def getGroup(groupName: String): Group = {
     val group = groupNameToGroups.getIfPresent(groupName)
     if (group == null) {