补批和重跑逻辑修改

byit-myth-core/myth-admin-core/src/main/java/com/byit/flowservice/impl/NodeVerificationImpl.java

@@ -3,6 +3,7 @@ package com.byit.flowservice.impl;
 import cn.hutool.core.collection.CollectionUtil;
 import com.byit.enums.NodePropertyEnum;
 import com.byit.enums.NodeRunStatusPropertyEnum;
+import com.byit.enums.ScheduleTypeEnum;
 import com.byit.exceptions.SuperiorNodeRunException;
 import com.byit.flowservice.NodeVerification;
 import com.byit.job.exceptions.BusinessException;
@@ -54,6 +55,10 @@ public class NodeVerificationImpl implements NodeVerification {
     public boolean superiorNodeStatus(JobTask thisJobTask) {
         //获取该节点的运行标识
         String runId = thisJobTask.getRunId();
+        if(ScheduleTypeEnum.REPEAT.getCode().equals(thisJobTask.getScheduleType())) {
+            runId = thisJobTask.getReRunId();
+        }
+
         //查询该节点的依赖节点
         String nodeDepend = thisJobTask.getNodeDepend();
         if(StringUtils.isBlank(nodeDepend)){

byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/helper/JobTaskThreadRunHelper.java

@@ -16,6 +16,7 @@ import com.byit.service.mapservice.TaskAndLogServer;
 import com.byit.service.mapservice.TaskAndScheduleService;
 import com.byit.thread.BaseThreadRunHelper;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
 import org.springframework.beans.BeanUtils;
 import org.springframework.stereotype.Component;
 
@@ -115,8 +116,12 @@ public class JobTaskThreadRunHelper extends BaseThreadRunHelper {
                     }else {
                         //普通节点
                         try {
-                            //断定上级节点是否执行成功
-                            if (nodeVerification.superiorNodeStatus(jobTask)) {
+                            //如果是补批节点 而且没有依赖节点的话，按照开始节点处理
+                            if(ScheduleTypeEnum.REPAIR.getCode().equals(jobTask.getScheduleType()) &&
+                                    StringUtils.isBlank(jobTask.getNodeDepend())){
+                                startNodeOperating(jobTask,jobTaskSchedules);
+                            }else if (nodeVerification.superiorNodeStatus(jobTask)) {
+                                //断定上级节点是否执行成功
                                 runJobTask(jobTask,jobTaskSchedules);
                             }
                         }catch (SuperiorNodeRunException se) {