[Feature][JsonSplit] Fix dependTask bug (#5360)

* update SnowFlake

* update processDefinite from processInstance

* update processDefinite from processInstance

* Fix task logger path

* Fix dependTask bug
Co-authored-by: NJinyLeeChina <297062848@qq.com>

dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/TaskInstance.java

@@ -241,7 +241,6 @@ public class TaskInstance implements Serializable {
     /**
      * task params
      */
-    @TableField(exist = false)
     private String taskParams;
 
     public void init(String host, Date startTime, String executePath) {

dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/TaskInstanceMapper.xml

@@ -22,13 +22,13 @@
         id, name, task_type, process_instance_id, task_code, task_definition_version, state, submit_time,
         start_time, end_time, host, execute_path, log_path, alert_flag, retry_times, pid, app_link,
         flag, retry_interval, max_retry_times, task_instance_priority, worker_group, executor_id,
-        first_submit_time, delay_time, var_pool
+        first_submit_time, delay_time, task_params, var_pool
     </sql>
     <sql id="baseSqlV2">
         ${alias}.id, ${alias}.name, ${alias}.task_type, ${alias}.task_code, ${alias}.task_definition_version, ${alias}.process_instance_id, ${alias}.state, ${alias}.submit_time,
         ${alias}.start_time, ${alias}.end_time, ${alias}.host, ${alias}.execute_path, ${alias}.log_path, ${alias}.alert_flag, ${alias}.retry_times, ${alias}.pid, ${alias}.app_link,
         ${alias}.flag, ${alias}.retry_interval, ${alias}.max_retry_times, ${alias}.task_instance_priority, ${alias}.worker_group, ${alias}.executor_id,
-        ${alias}.first_submit_time, ${alias}.delay_time, ${alias}.var_pool
+        ${alias}.first_submit_time, ${alias}.delay_time, ${alias}.task_params, ${alias}.var_pool
     </sql>
     <update id="setFailoverByHostAndStateArray">
         update t_ds_task_instance

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/DependentTaskExecThread.java

@@ -26,12 +26,11 @@ import org.apache.dolphinscheduler.common.model.DependentTaskModel;
 import org.apache.dolphinscheduler.common.task.dependent.DependentParameters;
 import org.apache.dolphinscheduler.common.thread.Stopper;
 import org.apache.dolphinscheduler.common.utils.DependentUtils;
-import org.apache.dolphinscheduler.common.utils.JSONUtils;
 import org.apache.dolphinscheduler.common.utils.LoggerUtils;
 import org.apache.dolphinscheduler.common.utils.NetUtils;
 import org.apache.dolphinscheduler.dao.entity.TaskInstance;
-import org.apache.dolphinscheduler.server.utils.LogUtils;
 import org.apache.dolphinscheduler.server.utils.DependentExecute;
+import org.apache.dolphinscheduler.server.utils.LogUtils;
 
 import java.util.ArrayList;
 import java.util.Date;
@@ -62,13 +61,13 @@ public class DependentTaskExecThread extends MasterBaseTaskExecThread {
     /**
      * dependent date
      */
-    @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss",timezone="GMT+8")
+    @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss", timezone = "GMT+8")
     private Date dependentDate;
 
     /**
      * constructor of MasterBaseTaskExecThread
      *
-     * @param taskInstance    task instance
+     * @param taskInstance task instance
      */
     public DependentTaskExecThread(TaskInstance taskInstance) {
         super(taskInstance);
@@ -78,7 +77,7 @@ public class DependentTaskExecThread extends MasterBaseTaskExecThread {
 
     @Override
     public Boolean submitWaitComplete() {
-        try{
+        try {
             logger.info("dependent task start");
             this.taskInstance = submit();
             logger = LoggerFactory.getLogger(LoggerUtils.buildTaskId(LoggerUtils.TASK_LOGGER_INFO_PREFIX,
@@ -92,8 +91,8 @@ public class DependentTaskExecThread extends MasterBaseTaskExecThread {
             initDependParameters();
             waitTaskQuit();
             updateTaskState();
-        }catch (Exception e){
-            logger.error("dependent task run exception" , e);
+        } catch (Exception e) {
+            logger.error("dependent task run exception", e);
         }
         return true;
     }
@@ -102,16 +101,13 @@ public class DependentTaskExecThread extends MasterBaseTaskExecThread {
      * init dependent parameters
      */
     private void initDependParameters() {
-
         this.dependentParameters = taskInstance.getDependency();
-
-        for(DependentTaskModel taskModel : dependentParameters.getDependTaskList()){
-            this.dependentTaskList.add(new DependentExecute(
-                    taskModel.getDependItemList(), taskModel.getRelation()));
+        for (DependentTaskModel taskModel : dependentParameters.getDependTaskList()) {
+            this.dependentTaskList.add(new DependentExecute(taskModel.getDependItemList(), taskModel.getRelation()));
         }
-        if(this.processInstance.getScheduleTime() != null){
+        if (this.processInstance.getScheduleTime() != null) {
             this.dependentDate = this.processInstance.getScheduleTime();
-        }else{
+        } else {
             this.dependentDate = new Date();
         }
     }
@@ -121,9 +117,9 @@ public class DependentTaskExecThread extends MasterBaseTaskExecThread {
      */
     private void updateTaskState() {
         ExecutionStatus status;
-        if(this.cancel){
+        if (this.cancel) {
             status = ExecutionStatus.KILL;
-        }else{
+        } else {
             DependResult result = getTaskDependResult();
             status = (result == DependResult.SUCCESS) ? ExecutionStatus.SUCCESS : ExecutionStatus.FAILURE;
         }
@@ -144,8 +140,8 @@ public class DependentTaskExecThread extends MasterBaseTaskExecThread {
             return true;
         }
         while (Stopper.isRunning()) {
-            try{
-                if(this.processInstance == null){
+            try {
+                if (this.processInstance == null) {
                     logger.error("process instance not exists , master task exec thread exit");
                     return true;
                 }
@@ -153,12 +149,12 @@ public class DependentTaskExecThread extends MasterBaseTaskExecThread {
                     this.checkTimeoutFlag = !alertTimeout();
                     handleTimeoutFailed();
                 }
-                if(this.cancel || this.processInstance.getState() == ExecutionStatus.READY_STOP){
+                if (this.cancel || this.processInstance.getState() == ExecutionStatus.READY_STOP) {
                     cancelTaskInstance();
                     break;
                 }
 
-                if ( allDependentTaskFinish() || taskInstance.getState().typeIsFinished()){
+                if (allDependentTaskFinish() || taskInstance.getState().typeIsFinished()) {
                     break;
                 }
                 // update process task
@@ -166,7 +162,7 @@ public class DependentTaskExecThread extends MasterBaseTaskExecThread {
                 processInstance = processService.findProcessInstanceById(processInstance.getId());
                 Thread.sleep(Constants.SLEEP_TIME_MILLIS);
             } catch (Exception e) {
-                logger.error("exception",e);
+                logger.error("exception", e);
                 if (processInstance != null) {
                     logger.error("wait task quit failed, instance id:{}, task id:{}",
                             processInstance.getId(), taskInstance.getId());
@@ -196,20 +192,20 @@ public class DependentTaskExecThread extends MasterBaseTaskExecThread {
 
     /**
      * judge all dependent tasks finish
+     *
      * @return whether all dependent tasks finish
      */
-    private boolean allDependentTaskFinish(){
+    private boolean allDependentTaskFinish() {
         boolean finish = true;
-        for(DependentExecute dependentExecute : dependentTaskList){
-            for(Map.Entry<String, DependResult> entry: dependentExecute.getDependResultMap().entrySet()) {
-                if(!dependResultMap.containsKey(entry.getKey())){
+        for (DependentExecute dependentExecute : dependentTaskList) {
+            for (Map.Entry<String, DependResult> entry : dependentExecute.getDependResultMap().entrySet()) {
+                if (!dependResultMap.containsKey(entry.getKey())) {
                     dependResultMap.put(entry.getKey(), entry.getValue());
                     //save depend result to log
-                    logger.info("dependent item complete {} {},{}",
-                            DEPENDENT_SPLIT, entry.getKey(), entry.getValue());
+                    logger.info("dependent item complete {} {},{}", DEPENDENT_SPLIT, entry.getKey(), entry.getValue());
                 }
             }
-            if(!dependentExecute.finish(dependentDate)){
+            if (!dependentExecute.finish(dependentDate)) {
                 finish = false;
             }
         }
@@ -218,17 +214,16 @@ public class DependentTaskExecThread extends MasterBaseTaskExecThread {
 
     /**
      * get dependent result
+     *
      * @return DependResult
      */
-    private DependResult getTaskDependResult(){
+    private DependResult getTaskDependResult() {
         List<DependResult> dependResultList = new ArrayList<>();
-        for(DependentExecute dependentExecute : dependentTaskList){
+        for (DependentExecute dependentExecute : dependentTaskList) {
             DependResult dependResult = dependentExecute.getModelDependResult(dependentDate);
             dependResultList.add(dependResult);
         }
-        DependResult result = DependentUtils.getDependResultForRelation(
-                this.dependentParameters.getRelation(), dependResultList
-        );
+        DependResult result = DependentUtils.getDependResultForRelation(this.dependentParameters.getRelation(), dependResultList);
         logger.info("dependent task completed, dependent result:{}", result);
         return result;
     }

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterExecThread.java

@@ -34,6 +34,7 @@ import org.apache.dolphinscheduler.common.enums.FailureStrategy;
 import org.apache.dolphinscheduler.common.enums.Flag;
 import org.apache.dolphinscheduler.common.enums.Priority;
 import org.apache.dolphinscheduler.common.enums.TaskDependType;
+import org.apache.dolphinscheduler.common.enums.TaskType;
 import org.apache.dolphinscheduler.common.graph.DAG;
 import org.apache.dolphinscheduler.common.model.TaskNode;
 import org.apache.dolphinscheduler.common.model.TaskNodeRelation;
@@ -448,13 +449,14 @@ public class MasterExecThread implements Runnable {
      * find task instance in db.
      * in case submit more than one same name task in the same time.
      *
-     * @param taskName task name
+     * @param taskCode task code
+     * @param taskVersion task version
      * @return TaskInstance
      */
-    private TaskInstance findTaskIfExists(String taskName) {
+    private TaskInstance findTaskIfExists(Long taskCode, int taskVersion) {
         List<TaskInstance> taskInstanceList = processService.findValidTaskListByProcessId(this.processInstance.getId());
         for (TaskInstance taskInstance : taskInstanceList) {
-            if (taskInstance.getName().equals(taskName)) {
+            if (taskInstance.getTaskCode() == taskCode && taskInstance.getTaskDefinitionVersion() == taskVersion) {
                 return taskInstance;
             }
         }
@@ -465,20 +467,19 @@ public class MasterExecThread implements Runnable {
      * encapsulation task
      *
      * @param processInstance process instance
-     * @param nodeName node name
+     * @param taskNode taskNode
      * @return TaskInstance
      */
-    private TaskInstance createTaskInstance(ProcessInstance processInstance, String nodeName,
-                                            TaskNode taskNode) {
+    private TaskInstance createTaskInstance(ProcessInstance processInstance, TaskNode taskNode) {
         //update processInstance for update the globalParams
         this.processInstance = this.processService.findProcessInstanceById(this.processInstance.getId());
-        TaskInstance taskInstance = findTaskIfExists(nodeName);
+        TaskInstance taskInstance = findTaskIfExists(taskNode.getCode(), taskNode.getVersion());
         if (taskInstance == null) {
             taskInstance = new TaskInstance();
             taskInstance.setTaskCode(taskNode.getCode());
             taskInstance.setTaskDefinitionVersion(taskNode.getVersion());
             // task name
-            taskInstance.setName(nodeName);
+            taskInstance.setName(taskNode.getName());
             // task instance state
             taskInstance.setState(ExecutionStatus.SUBMITTED_SUCCESS);
             // process instance id
@@ -518,27 +519,28 @@ public class MasterExecThread implements Runnable {
             } else {
                 taskInstance.setWorkerGroup(taskWorkerGroup);
             }
-            //get process global
-            setProcessGlobal(taskNode, taskInstance);
+            if (TaskType.DEPENDENT.getDesc().equalsIgnoreCase(taskNode.getType())) {
+                taskInstance.setTaskParams(taskNode.getDependence());
+            } else {
+                taskInstance.setTaskParams(globalParamToTaskParams(taskNode.getParams()));
+            }
             // delay execution time
             taskInstance.setDelayTime(taskNode.getDelayTime());
         }
         return taskInstance;
     }
 
-    private void setProcessGlobal(TaskNode taskNode, TaskInstance taskInstance) {
+    private String globalParamToTaskParams(String params) {
         String globalParams = this.processInstance.getGlobalParams();
-        if (StringUtils.isNotEmpty(globalParams)) {
-            Map<String, String> globalMap = processService.getGlobalParamMap(globalParams);
-            if (globalMap != null && globalMap.size() != 0) {
-                setGlobalMapToTask(taskNode, taskInstance, globalMap);
-            }
+        if (StringUtils.isBlank(globalParams)) {
+            return params;
         }
-    }
-
-    private void setGlobalMapToTask(TaskNode taskNode, TaskInstance taskInstance, Map<String, String> globalMap) {
-        // the param save in localParams
-        Map<String, Object> result = JSONUtils.toMap(taskNode.getParams(), String.class, Object.class);
+        Map<String, String> globalMap = processService.getGlobalParamMap(globalParams);
+        if (globalMap == null || globalMap.size() == 0) {
+            return params;
+        }
+        // the process global param save in localParams
+        Map<String, Object> result = JSONUtils.toMap(params, String.class, Object.class);
         Object localParams = result.get(LOCAL_PARAMS);
         if (localParams != null) {
             List<Property> allParam = JSONUtils.toList(JSONUtils.toJsonString(localParams), Property.class);
@@ -552,8 +554,8 @@ public class MasterExecThread implements Runnable {
                 }
             }
             result.put(LOCAL_PARAMS, allParam);
-            taskInstance.setTaskParams(JSONUtils.toJsonString(result));
         }
+        return JSONUtils.toJsonString(result);
     }
 
     private void submitPostNode(String parentNodeName) {
@@ -567,9 +569,10 @@ public class MasterExecThread implements Runnable {
                 throw new RuntimeException();
             }
             TaskNode taskNodeObject = dag.getNode(taskNode);
-            VarPoolUtils.setTaskNodeLocalParams(taskNodeObject, propToValue);
-            taskInstances.add(createTaskInstance(processInstance, taskNode,
-                    taskNodeObject));
+            if (!TaskType.DEPENDENT.getDesc().equalsIgnoreCase(taskNodeObject.getType())) {
+                VarPoolUtils.setTaskNodeLocalParams(taskNodeObject, propToValue);
+            }
+            taskInstances.add(createTaskInstance(processInstance, taskNodeObject));
         }
 
         // if previous node success , post node submit

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/utils/DependentExecute.java

@@ -87,7 +87,7 @@ public class DependentExecute {
      */
     private DependResult getDependentResultForItem(DependentItem dependentItem, Date currentTime){
         List<DateInterval> dateIntervals = DependentUtils.getDateIntervalList(currentTime, dependentItem.getDateValue());
-        return calculateResultForTasks(dependentItem, dateIntervals );
+        return calculateResultForTasks(dependentItem, dateIntervals);
     }
 
     /**
@@ -257,9 +257,7 @@ public class DependentExecute {
             }
             dependResultList.add(dependResult);
         }
-        modelDependResult = DependentUtils.getDependResultForRelation(
-                this.relation, dependResultList
-        );
+        modelDependResult = DependentUtils.getDependResultForRelation(this.relation, dependResultList);
         return modelDependResult;
     }
 

dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessService.java

@@ -1247,7 +1247,6 @@ public class ProcessService {
             if (taskInstance.isSubProcess()) {
                 taskInstance.setRetryTimes(taskInstance.getRetryTimes() + 1);
             } else {
-
                 if (processInstanceState != ExecutionStatus.READY_STOP
                         && processInstanceState != ExecutionStatus.READY_PAUSE) {
                     // failure task set invalid