add RetryCountLabel and retrycount param

linkis-computation-governance/linkis-manager/label-common/src/main/java/org/apache/linkis/manager/label/constant/LabelKeyConstant.java

@@ -53,6 +53,7 @@ public class LabelKeyConstant {
 
     public static final String RETRY_TIMEOUT_KEY = "jobRetryTimeout";
 
+    public static final String RETRY_COUNT_KEY = "jobRetryCount";
     public static final String EXECUTE_ONCE_KEY = "executeOnce";
 
     public static final String LOAD_BALANCE_KEY = "loadBalance";

linkis-computation-governance/linkis-manager/label-common/src/main/java/org/apache/linkis/manager/label/entity/entrance/RetryCountLabel.java

@@ -0,0 +1,46 @@
+/*
+ * 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.manager.label.entity.entrance;
+
+import org.apache.linkis.manager.label.constant.LabelKeyConstant;
+import org.apache.linkis.manager.label.entity.GenericLabel;
+import org.apache.linkis.manager.label.entity.annon.ValueSerialNum;
+
+import java.util.HashMap;
+
+public class RetryCountLabel extends GenericLabel implements JobStrategyLabel {
+
+    public RetryCountLabel() {
+        setLabelKey(LabelKeyConstant.RETRY_COUNT_KEY);
+    }
+
+    public Integer getJobRetryCount() {
+        if (null == getValue()) {
+            return -1;
+        }
+        return Integer.parseInt(getValue().getOrDefault(LabelKeyConstant.RETRY_COUNT_KEY, "-1"));
+    }
+
+    @ValueSerialNum(0)
+    public void setJobRetryCount(String count) {
+        if (null == getValue()) {
+            setValue(new HashMap<>());
+        }
+        getValue().put(LabelKeyConstant.RETRY_COUNT_KEY, count);
+    }
+}

linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/computation/catalyst/reheater/PruneTaskReheaterTransform.scala

@@ -30,6 +30,9 @@ import org.apache.linkis.orchestrator.listener.task.TaskLogEvent
 import org.apache.linkis.orchestrator.plans.physical.{ExecTask, PhysicalContext, PhysicalOrchestration, ReheatableExecTask, RetryExecTask}
 import org.apache.linkis.orchestrator.strategy.ExecTaskStatusInfo
 
+import java.util
+import scala.collection.JavaConverters.mapAsScalaMapConverter
+
 /**
  * Transform physical tree by pruning it's nodes
  *
@@ -52,7 +55,7 @@ class PruneTaskRetryTransform extends ReheaterTransform with Logging{
                   Utils.tryCatch{
                     task match {
                       case retryExecTask: RetryExecTask => {
-                        if (retryExecTask.getAge() < ComputationOrchestratorConf.RETRYTASK_MAXIMUM_AGE.getValue) {
+                        if (retryExecTask.getAge() < retryExecTask.getMaxRetryCount()) {
                           val newTask = new RetryExecTask(retryExecTask.getOriginTask, retryExecTask.getAge() + 1)
                           newTask.initialize(retryExecTask.getPhysicalContext)
                           TreeNodeUtil.replaceNode(retryExecTask, newTask)

linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/computation/conf/ComputationOrchestratorConf.scala

@@ -47,8 +47,6 @@ object ComputationOrchestratorConf {
 
   val LOG_LEN = CommonVars("wds.linkis.computation.orchestrator.log.len", 100)
 
-  val RETRYTASK_MAXIMUM_AGE = CommonVars("wds.linkis.computation.orchestrator.retry.max.age", 10)
-
 
   val ENGINECONN_LASTUPDATE_TIMEOUT = CommonVars("wds.linkis.orchestrator.engine.lastupdate.timeout", new TimeType("5s"))
   val ENGINECONN_ACTIVITY_TIMEOUT = CommonVars("wds.linkis.orchestrator.engine.timeout", new TimeType("10s"))

linkis-orchestrator/linkis-orchestrator-core/src/main/scala/org/apache/linkis/orchestrator/conf/OrchestratorConfiguration.scala

@@ -54,6 +54,8 @@ object OrchestratorConfiguration {
 
   val RETRY_TASK_WAIT_TIME = CommonVars("wds.linkis.orchestrator.task.retry.wait.time", 30000)
 
+  val RETRYTASK_MAXIMUM_AGE = CommonVars("wds.linkis.computation.orchestrator.retry.max.age", 10)
+
   val SCHEDULER_RETRY_TASK_WAIT_TIME = CommonVars("wds.linkis.orchestrator.task.scheduler.retry.wait.time", 100000)
 
   val TASK_SCHEDULER_THREAD_POOL = CommonVars("wds.linkis.orchestrator.task.scheduler.thread.pool", 200)

linkis-orchestrator/linkis-orchestrator-core/src/main/scala/org/apache/linkis/orchestrator/plans/physical/RetryExecTask.scala

@@ -17,8 +17,9 @@
 
 package org.apache.linkis.orchestrator.plans.physical
 
+import org.apache.linkis.common.utils.Utils
 import org.apache.linkis.manager.label.entity.Label
-import org.apache.linkis.manager.label.entity.entrance.RetryWaitTimeOutLabel
+import org.apache.linkis.manager.label.entity.entrance.{RetryCountLabel, RetryWaitTimeOutLabel}
 import org.apache.linkis.manager.label.utils.LabelUtil
 import org.apache.linkis.orchestrator.conf.OrchestratorConfiguration
 import org.apache.linkis.orchestrator.exception.{OrchestratorErrorCodeSummary, OrchestratorErrorException}
@@ -29,6 +30,9 @@ import org.apache.linkis.orchestrator.strategy.{ResultSetExecTask, StatusInfoExe
 import org.apache.linkis.orchestrator.strategy.async.AsyncExecTask
 import org.apache.linkis.orchestrator.utils.OrchestratorIDCreator
 
+import java.util
+import scala.collection.JavaConverters.mapAsScalaMapConverter
+
 
 class RetryExecTask(private val originTask: ExecTask, private val age: Int = 1) extends AbstractExecTask
   with StatusInfoExecTask with ResultSetExecTask with AsyncExecTask{
@@ -55,6 +59,25 @@ class RetryExecTask(private val originTask: ExecTask, private val age: Int = 1)
     waitTime
   }
 
+  def getMaxRetryCount(): Integer = {
+    var count = -1
+    val retryCountLabel = LabelUtil.getLabelFromList[RetryCountLabel](getLabels)
+    if (null != retryCountLabel) {
+      count = retryCountLabel.getJobRetryCount
+    } else {
+      val runtimeMap = new util.HashMap[String, String]()
+      Utils.tryAndWarn {
+        getTaskDesc.getOrigin.getASTOrchestration.getASTContext.getParams.getRuntimeParams.toMap.asScala.foreach(kv => {
+          if (kv._2.isInstanceOf[String]) {
+            runtimeMap.put(kv._1, kv._2.asInstanceOf[String])
+          }
+        })
+        count = OrchestratorConfiguration.RETRYTASK_MAXIMUM_AGE.getValue(runtimeMap)
+      }
+    }
+    count
+  }
+
   def getOriginTask: ExecTask = {
     originTask
   }