code optimization

Author: kinsanities

linkis-computation-governance/linkis-computation-governance-common/src/main/scala/org/apache/linkis/governance/common/protocol/task/ResponseTaskExecute.scala

@@ -58,15 +58,6 @@ case class ResponseTaskStatus(execId: String, status: ExecutionNodeStatus)
     extends RetryableProtocol
     with RequestProtocol
 
-class ResponseTaskStatusWithExecuteCodeIndex(
-    execId: String,
-    status: ExecutionNodeStatus,
-    private var _errorIndex: Int = -1
-) extends ResponseTaskStatus(execId, status) {
-  def errorIndex: Int = _errorIndex
-  def errorIndex_=(value: Int): Unit = _errorIndex = value
-}
-
 case class ResponseTaskResultSet(execId: String, output: String, alias: String)
     extends RetryableProtocol
     with RequestProtocol {

linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/computation/service/ComputationTaskExecutionReceiver.scala

@@ -93,21 +93,10 @@ class ComputationTaskExecutionReceiver extends TaskExecutionReceiver with Loggin
         .getByEngineConnAndTaskId(serviceInstance, taskStatus.execId)
         .foreach { codeExecutor =>
           OrchestratorLoggerUtils.setJobIdMDC(codeExecutor.getExecTask)
-
-          taskStatus match {
-            case rts: ResponseTaskStatusWithExecuteCodeIndex =>
-              logger.info(s"execute error with index: ${rts.errorIndex}")
-              codeExecutor.getExecTask.updateParams(
-                "execute.error.code.index",
-                rts.errorIndex.toString
-              )
-            case _ =>
-          }
           val event = TaskStatusEvent(codeExecutor.getExecTask, taskStatus.status)
           logger.info(
             s"From engineConn receive status info:$taskStatus, now post to listenerBus event: $event"
           )
-
           codeExecutor.getExecTask.getPhysicalContext.broadcastSyncEvent(event)
           codeExecutor.getEngineConnExecutor.updateLastUpdateTime()
           isExist = true

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

@@ -70,6 +70,8 @@ class PhysicalContextImpl(private var rootTask: ExecTask, private var leafTasks:
       OrchestratorErrorCodeSummary.EXECUTION_ERROR_CODE,
       cause
     )
+    // 标识失败代码索引，以便重试的时候只执行未执行代码
+    failedResponse.errorIndex = this.rootTask.params.getOrElse("execute.error.code.index", "-1").toInt
     this.response = failedResponse
     syncListenerBus.postToAll(RootTaskResponseEvent(getRootTask, failedResponse))
   }

linkis-orchestrator/linkis-orchestrator-core/src/main/scala/org/apache/linkis/orchestrator/strategy/async/AsyncExecTaskRunnerImpl.scala

@@ -75,11 +75,6 @@ class AsyncExecTaskRunnerImpl(override val task: ExecTask)
         case succeed: SucceedTaskResponse =>
           logger.info(s"Succeed to execute ExecTask(${task.getIDInfo})")
           transientStatus(ExecutionNodeStatus.Succeed)
-        case failedTaskResponse: DefaultFailedTaskResponse =>
-          logger.info(s"DefaultFailed to execute ExecTask(${task.getIDInfo})")
-          failedTaskResponse.errorIndex =
-            task.params.getOrElse("execute.error.code.index", "-1").toInt
-          transientStatus(ExecutionNodeStatus.Failed)
         case failedTaskResponse: FailedTaskResponse =>
           logger.info(s"Failed to execute ExecTask(${task.getIDInfo})")
           transientStatus(ExecutionNodeStatus.Failed)