失败重试

byit-myth-core/myth-admin-core/src/main/java/com/byit/mapper/JobTaskRunLogMapper.java

@@ -22,6 +22,12 @@ public interface JobTaskRunLogMapper {
    List<JobTaskRunLog> findJobTaskRunLogNotEndNodeByRunCodeCount(@Param("nodeIds") List<Integer> nodeIds, @Param("runId") String runId);
 
     /**
+     * 查询失败且重试次数大于0的节点
+     * @return
+     */
+    List<JobTaskRunLog> findErrorNode();
+
+    /**
      * 查根据flowId和RunId查询一批节点
      * @param flowId
      * @param runId

byit-myth-core/myth-admin-core/src/main/java/com/byit/model/JobTask.java

@@ -201,6 +201,12 @@ public class JobTask implements Serializable {
     private String reRunId;
 
     /**
+     * 日志表id
+     */
+    @ApiModelProperty("日志表id")
+    private Integer logId;
+
+    /**
      */
     private static final long serialVersionUID = 1L;
 }
\ No newline at end of file

byit-myth-core/myth-admin-core/src/main/java/com/byit/model/JobTaskRunLog.java

@@ -175,6 +175,11 @@ public class JobTaskRunLog implements Serializable {
     private String superSuccessRun;
 
     /**
+     * 执行次数
+     */
+    @ApiModelProperty("执行次数")
+    private Integer runCount;
+    /**
      *
      */
     private static final long serialVersionUID = 1L;

byit-myth-core/myth-admin-core/src/main/java/com/byit/service/JobTaskRunLogAndJobTaskService.java

+package com.byit.service;
+
+import com.byit.model.JobTaskRunLog;
+import org.springframework.transaction.annotation.Transactional;
+
+/**
+ * @author huangfu
+ * 日志表和任务表的事务映射
+ */
+public interface JobTaskRunLogAndJobTaskService {
+    /**
+     * 修改日志文件和保存任务表
+     * @param jobTaskRunLog
+     */
+    void updateLogAndSaveJobTask(JobTaskRunLog jobTaskRunLog);
+}

byit-myth-core/myth-admin-core/src/main/java/com/byit/service/JobTaskRunLogService.java

@@ -20,6 +20,12 @@ public interface JobTaskRunLogService {
      * @return
      */
     List<JobTaskRunLog> findJobTaskRunLogNotEndNodeByRunCodeCount(List<Integer> nodIds, String runId);
+
+    /**
+     * 查询失败且重试次数大于0的节点
+     * @return
+     */
+    List<JobTaskRunLog> findErrorNode();
     /**
      * 查根据flowId和RunId查询一批节点
      * @param flowId

byit-myth-core/myth-admin-core/src/main/java/com/byit/service/NodeService.java

@@ -18,6 +18,13 @@ public interface NodeService {
      */
     List<Node> findNodeByFlowIdAndVersionName(Integer flowId);
 
+    /**
+     * 根据节点ID查询节点
+     * @param id
+     * @return
+     */
+    Node findNodeById(Integer id);
+
 
 
 

byit-myth-core/myth-admin-core/src/main/java/com/byit/service/impl/JobTaskRunLogAndJobTaskServiceImpl.java

+package com.byit.service.impl;
+
+import com.byit.model.JobTask;
+import com.byit.model.JobTaskRunLog;
+import com.byit.model.Node;
+import com.byit.service.JobTaskRunLogAndJobTaskService;
+import com.byit.service.JobTaskRunLogService;
+import com.byit.service.JobTaskService;
+import com.byit.service.NodeService;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.BeanUtils;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+/**
+ * @author huangfu
+ * 日志表和任务表的事务映射
+ */
+@Transactional(rollbackFor = Exception.class)
+@Service
+@Slf4j
+public class JobTaskRunLogAndJobTaskServiceImpl implements JobTaskRunLogAndJobTaskService {
+    private final JobTaskRunLogService jobTaskRunLogService;
+    private final NodeService nodeService;
+    private final JobTaskService jobTaskService;
+
+    public JobTaskRunLogAndJobTaskServiceImpl(JobTaskRunLogService jobTaskRunLogService, NodeService nodeService, JobTaskService jobTaskService) {
+        this.jobTaskRunLogService = jobTaskRunLogService;
+        this.nodeService = nodeService;
+        this.jobTaskService = jobTaskService;
+    }
+
+    @Override
+    public void updateLogAndSaveJobTask(JobTaskRunLog jobTaskRunLog) {
+        log.debug("-------------【{}节点失败重试------------------】",jobTaskRunLog);
+        jobTaskRunLog.setFailedRemainingCount(jobTaskRunLog.getFailedRemainingCount()-1);
+        jobTaskRunLog.setRunCount(jobTaskRunLog.getRunCount()+1);
+        jobTaskRunLogService.updateJobTaskRunLog(jobTaskRunLog);
+
+        //查询节点
+        Node node = nodeService.findNodeById(jobTaskRunLog.getNodeId());
+        //获取重试间隔
+        Long failedRetryInterval = node.getFailedRetryInterval();
+        //获取总共的重试次数
+        Integer failedRetryCount = node.getFailedRetryCount();
+        //获取剩余失败重试间隔
+        Integer failedRemainingCount = jobTaskRunLog.getFailedRemainingCount();
+        //计算重试间隔时间
+        Long nextReTime = (failedRetryCount-failedRemainingCount)*failedRetryInterval;
+        //获取本次执行完的时间
+        long endTime = jobTaskRunLog.getEndTime().getTime();
+        //获取重试时间
+        Long nextTime = endTime+nextReTime;
+
+        //插入job_task表
+        JobTask jobTask = new JobTask();
+        jobTask.setTriggerStatus("1");
+        BeanUtils.copyProperties(node,jobTask);
+        jobTask.setTriggerTime(nextTime);
+        jobTask.setRunId(jobTaskRunLog.getRunId());
+        jobTask.setLogId(jobTaskRunLog.getLogId());
+        jobTask.setFlowName(jobTaskRunLog.getFlowName());
+        jobTaskService.addMythJobTask(jobTask);
+    }
+}

byit-myth-core/myth-admin-core/src/main/java/com/byit/service/impl/JobTaskRunLogServiceImpl.java

@@ -36,6 +36,11 @@ public class JobTaskRunLogServiceImpl implements JobTaskRunLogService {
     }
 
     @Override
+    public List<JobTaskRunLog> findErrorNode() {
+        return jobTaskRunLogMapper.findErrorNode();
+    }
+
+    @Override
     public List<JobTaskRunLogWithBLOBs> findJobTaskRunLogWithBLOBsByFlowIdAndRunId(Integer flowId, String runId) {
         return jobTaskRunLogMapper.findJobTaskRunLogWithBLOBsByFlowIdAndRunId(flowId,runId);
     }

byit-myth-core/myth-admin-core/src/main/java/com/byit/service/impl/NodeServiceImpl.java

@@ -4,6 +4,7 @@ import com.byit.mapper.NodeMapper;
 import com.byit.model.Node;
 import com.byit.service.NodeService;
 import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
 
 import javax.annotation.Resource;
 import java.util.List;
@@ -14,6 +15,7 @@ import java.util.List;
  * @create: 2019-12-24 10:39
  */
 @Service
+@Transactional(rollbackFor = Exception.class)
 public class NodeServiceImpl implements NodeService {
 
     @Resource
@@ -23,4 +25,9 @@ public class NodeServiceImpl implements NodeService {
     public List<Node> findNodeByFlowIdAndVersionName(Integer flowId) {
         return nodeMapper.findNodeByFlowIdAndVersionName(flowId);
     }
+
+    @Override
+    public Node findNodeById(Integer id) {
+        return nodeMapper.getById(id);
+    }
 }

byit-myth-core/myth-admin-core/src/main/java/com/byit/task/JavaBeanJobTask.java

@@ -68,9 +68,20 @@ public class JavaBeanJobTask implements TimerTask {
 
     private void saveLog(JobTaskSchedule mythJobTaskSchedule, String url, String result){
         DispatchResponseDto dispatchResponseDto = JSON.parseObject(result, DispatchResponseDto.class);
-
         JobTaskRunLogServiceImpl jobTaskRunLogService = SpringUtil.getBean(JobTaskRunLogServiceImpl.class);
+
+        JobTaskRunLogWithBLOBs jobTaskRunLogById = jobTaskRunLogService.findJobTaskRunLogById(mythJobTaskSchedule.getLogId());
+
         JobTaskRunLogWithBLOBs jobTaskRunLog = new JobTaskRunLogWithBLOBs();
+
+        if (jobTaskRunLogById.getRunCount()>1) {
+            //第一次调度的日志
+            String triggerMsg = jobTaskRunLogById.getTriggerMsg();
+            jobTaskRunLog.setTriggerMsg(triggerMsg+"|"+dispatchResponseDto.getMsg());
+        }else{
+            jobTaskRunLog.setTriggerMsg(dispatchResponseDto.getMsg());
+        }
+
         jobTaskRunLog.setLogId(mythJobTaskSchedule.getLogId());
         jobTaskRunLog.setVersionName(mythJobTaskSchedule.getVersionName());
         jobTaskRunLog.setRunType("2");
@@ -78,14 +89,19 @@ public class JavaBeanJobTask implements TimerTask {
         jobTaskRunLog.setHandlerName(mythJobTaskSchedule.getHandlerName());
         jobTaskRunLog.setTriggerTime(new Date());
         jobTaskRunLog.setTriggerCode(dispatchResponseDto.getCode());
-        jobTaskRunLog.setTriggerMsg(dispatchResponseDto.getMsg());
+
         if(!("1".equals(dispatchResponseDto.getCode()))){
             Date thisTime = new Date();
             jobTaskRunLog.setStartTime(thisTime);
             jobTaskRunLog.setEndTime(thisTime);
             jobTaskRunLog.setRunCode("2");
             jobTaskRunLog.setAlertEnd("1");
-            jobTaskRunLog.setRunMsg(dispatchResponseDto.getMsg());
+            if (jobTaskRunLogById.getRunCount()>1) {
+                //上一次的执行日志
+                String runMsg = jobTaskRunLogById.getRunMsg();
+                jobTaskRunLog.setRunMsg(runMsg+"|"+"调度失败");
+            }
+
         }
         jobTaskRunLogService.updateJobTaskRunLogWithBLOBs(jobTaskRunLog);
 

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

@@ -175,13 +175,14 @@ public class JobScheduleHelper{
                                         if(dependIdByNodeId.size() == jobTaskRunLogList.size()){
                                             //过滤失败的节点
                                             List<JobTaskRunLog> errorJobLog = jobTaskRunLogList.stream().
-                                                    filter(jobTaskRunLog -> ("2".equals(jobTaskRunLog.getRunCode()) || "4".equals(jobTaskRunLog.getRunCode())))
+                                                    filter(jobTaskRunLog -> ("2".equals(jobTaskRunLog.getRunCode()) || "4".equals(jobTaskRunLog.getRunCode()) || "6".equals(jobTaskRunLog.getRunCode())))
                                                     .collect(Collectors.toList());
                                             //判断剩余执行次数是否为0
                                             if (parentNodeErrorCount(errorJobLog)) {
                                                 log.debug("--------------【{}的上级节点的失败节点已经全部重试完毕】------------------",jobTask);
                                                 //该节点如果为弱引用
                                                 if ("1".equals(jobTask.getSuperSuccessRun())) {
+                                                    log.info("-------------【查询到有弱引用节点】-----------------");
                                                     //执行代码
                                                     runJobTask(jobTask,jobTaskSchedules);
                                                 }else{
@@ -323,9 +324,14 @@ public class JobScheduleHelper{
                     if(CollectionUtil.isNotEmpty(jobTaskSchedules)){
                         //循环遍历添加任务
                         jobTaskSchedules.forEach(mythJobTaskSchedule ->{
+                            //如果是重跑就有logId
+                            Integer logId = mythJobTaskSchedule.getLogId();
                             if ("JAVA".equals(mythJobTaskSchedule.getJobType())) {
-                                Integer logId = saveLog(mythJobTaskSchedule);
+                                if(logId == null){
+                                    logId = saveLog(mythJobTaskSchedule);
+                                }
                                 mythJobTaskSchedule.setLogId(logId);
+                                //构建调度执行器
                                 JavaBeanJobTask javaBeanJobTask = new JavaBeanJobTask(mythJobTaskSchedule);
                                 jobTaskScheduleService.delete(mythJobTaskSchedule.getId());
                                 WorkRoulette.addJob(javaBeanJobTask,mythJobTaskSchedule.getTriggerTime());

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

@@ -27,13 +27,21 @@ public class LogCallbackThread implements Runnable {
         log.debug("--------------------任务执行完成---------------------");
         JobTaskRunLogServiceImpl mythJobTaskRunLogService = SpringUtil.getBean(JobTaskRunLogServiceImpl.class);
         JobTaskRunLogWithBLOBs jobTaskRunLog = new JobTaskRunLogWithBLOBs();
+
+        JobTaskRunLogWithBLOBs jobTaskRunLogById = mythJobTaskRunLogService.findJobTaskRunLogById(jobRunResultDto.getLogId());
+
+        if (jobTaskRunLogById.getRunCount()>1) {
+            String runMsg = jobTaskRunLogById.getRunMsg()+"|"+jobRunResultDto.getReturnResult().getMsg();
+            jobTaskRunLog.setRunMsg(runMsg);
+        }else{
+            jobTaskRunLog.setRunMsg(jobRunResultDto.getReturnResult().getMsg());
+        }
+
         jobTaskRunLog.setLogId(jobRunResultDto.getLogId());
         jobTaskRunLog.setStartTime(jobRunResultDto.getStartTime());
-        System.out.println(DateUtil.format(jobRunResultDto.getStartTime(),"yyyy-MM-dd HH:mm:ss"));
         jobTaskRunLog.setEndTime(jobRunResultDto.getEndTime());
-        System.out.println(DateUtil.format(jobRunResultDto.getEndTime(),"yyyy-MM-dd HH:mm:ss"));
         jobTaskRunLog.setRunCode(jobRunResultDto.getReturnResult().getCode());
-        jobTaskRunLog.setRunMsg(jobRunResultDto.getReturnResult().getMsg());
+
         jobTaskRunLog.setAlertEnd("0");
         mythJobTaskRunLogService.updateJobTaskRunLogWithBLOBs(jobTaskRunLog);
 

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

@@ -2,14 +2,12 @@ package com.byit.thread;
 
 import cn.hutool.core.collection.CollectionUtil;
 import com.byit.enums.RunRecordingEnum;
-import com.byit.model.EmailAlarm;
 import com.byit.model.JobTaskRunLog;
 import com.byit.model.RunRecording;
-import com.byit.service.EmailAlarmService;
+import com.byit.service.JobTaskRunLogAndJobTaskService;
 import com.byit.service.JobTaskRunLogService;
 import com.byit.service.RunRecordingService;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.commons.lang3.StringUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
@@ -33,10 +31,12 @@ public class LogScanHelper {
     private DataSource dataSource;
     private final JobTaskRunLogService jobTaskRunLogService;
     private final RunRecordingService runRecordingService;
+    private final JobTaskRunLogAndJobTaskService jobTaskRunLogAndJobTaskService;
     @Autowired
-    public LogScanHelper(JobTaskRunLogService jobTaskRunLogService, RunRecordingService runRecordingService) {
+    public LogScanHelper(JobTaskRunLogService jobTaskRunLogService, RunRecordingService runRecordingService, JobTaskRunLogAndJobTaskService jobTaskRunLogAndJobTaskService) {
         this.jobTaskRunLogService = jobTaskRunLogService;
         this.runRecordingService = runRecordingService;
+        this.jobTaskRunLogAndJobTaskService = jobTaskRunLogAndJobTaskService;
     }
 
     @Autowired
@@ -54,6 +54,15 @@ public class LogScanHelper {
      */
     private volatile boolean notAlarmedNodeScanIsStop = false;
     /**
+     * 扫描失败的节点的线程是否停止
+     */
+    private volatile boolean errorNodeScanIsStop = false;
+
+    /**
+     * 扫描失败的节点的线程是否停止
+     */
+    private Thread errorNodeScanThread = null;
+    /**
      * 扫描虚节点的线程定义
      */
     private Thread virtualNodeScanThread = null;
@@ -65,6 +74,7 @@ public class LogScanHelper {
     public void start(){
         virtualNodeScanMethod();
         notAlarmedNodeScanMethod();
+        errorNodeScan();
     }
 
     /**
@@ -252,6 +262,93 @@ public class LogScanHelper {
         notAlarmedNodeScanThread.start();
     }
 
+    /**
+     * 失败的节点扫描 执行失败重试
+     * 扫描到失败节点，过滤引上级节点失败的情况
+     */
+    public void errorNodeScan(){
+        errorNodeScanThread = new Thread(()->{
+            dateAligned(5000);
+            log.info("---------------------【com.byit.thread.LogScanHelper#errorNodeScan】init success------------------------");
+            while (!errorNodeScanIsStop) {
+                //是否需要睡眠
+                boolean isSleep = false;
+                Connection conn = null;
+                Boolean connAutoCommit = null;
+                PreparedStatement preparedStatement = null;
+                try {
+                    conn = dataSource.getConnection();
+                    connAutoCommit = conn.getAutoCommit();
+                    conn.setAutoCommit(false);
+                    preparedStatement = conn.prepareStatement("SELECT * FROM JOB_LOCK WHERE LOCK_NAME = 'log_error_lock' FOR UPDATE ");
+                    preparedStatement.execute();
+
+                    //查询又重试次数的失败节点
+                    List<JobTaskRunLog> errorNodes = jobTaskRunLogService.findErrorNode();
+                    if(CollectionUtil.isNotEmpty(errorNodes)){
+                        errorNodes.forEach(errorNode ->{
+                            log.debug("-----------------【操作失败节点{}】-----------------",errorNode);
+                            jobTaskRunLogAndJobTaskService.updateLogAndSaveJobTask(errorNode);
+                        });
+                    }else{
+                        isSleep = true;
+                    }
+
+
+                } catch (Exception e) {
+                    e.printStackTrace();
+                } finally {
+                    //释放资源
+                    if (conn != null) {
+                        try {
+                            conn.commit();
+                        } catch (SQLException e) {
+                            if (!virtualNodeScanIsStop) {
+                                log.error("--------------------【提交行锁出错】---------------------");
+                            }
+                        }
+                    }
+
+                    try {
+                        if (conn != null) {
+                            conn.setAutoCommit(connAutoCommit);
+                        }
+                    } catch (SQLException e) {
+                        if (!virtualNodeScanIsStop) {
+                            log.error("--------------------【恢复自动提交出错】---------------------");
+                        }
+                    }
+
+                    if (preparedStatement != null) {
+                        try {
+                            preparedStatement.close();
+                        } catch (SQLException e) {
+                            if (!notAlarmedNodeScanIsStop) {
+                                log.error("--------------------【关闭执行器出错】---------------------");
+                            }
+                        }
+                    }
+
+                    try {
+                        conn.close();
+                    } catch (SQLException e) {
+                        if (!virtualNodeScanIsStop) {
+                            log.error("--------------------【关闭链接出错】---------------------");
+                        }
+                    }
+                }
+
+                if (isSleep) {
+                    dateAligned(20000);
+                }
+            }
+        });
+        errorNodeScanThread.setName("myth-job#【LogScanHelper】#errorNodeScan");
+        errorNodeScanThread.setDaemon(true);
+        errorNodeScanThread.start();
+
+    }
+
 
     public void doStop(){
         this.virtualNodeScanIsStop = true;
@@ -287,6 +384,22 @@ public class LogScanHelper {
         }
         log.warn("---------------【日志扫描线程被注销】-----------------------");
 
+        this.errorNodeScanIsStop = true;
+        try {
+            TimeUnit.SECONDS.sleep(1);
+        } catch (InterruptedException e) {
+            e.printStackTrace( );
+        }
+        if (errorNodeScanThread.getState() != Thread.State.TERMINATED) {
+            errorNodeScanThread.interrupt();
+            try {
+                errorNodeScanThread.join();
+            } catch (InterruptedException e) {
+                e.printStackTrace( );
+            }
+        }
+        log.warn("---------------【失败重试节点扫描线程被注销】-----------------------");
+
 
     }
 

byit-myth-core/myth-admin-core/src/main/resources/mapper/JobTaskMapper.xml

@@ -33,6 +33,7 @@
     <result column="flow_name" jdbcType="VARCHAR" property="flowName" />
     <result column="super_success_run" jdbcType="CHAR" property="superSuccessRun"/>
     <result column="re_run_id" jdbcType="VARCHAR" property="reRunId"/>
+    <result column="log_id" jdbcType="INTEGER" property="logId"/>
   </resultMap>
   <resultMap extends="BaseResultMap" id="ResultMapWithBLOBs" type="com.byit.model.JobTask">
     <!-- generated @mbg.generated date: 2019-12-25 -->
@@ -44,7 +45,7 @@
     job_type, handler_name, node_desc, node_name, map_flow_id, node_timeout, is_virtual, 
     plugin_urls, priority, failed_retry_interval, routing_strategy, run_id, run_param, 
     run_source_desc, script_urls, source_principal, trigger_time, trigger_status, version_name, 
-    run_command,flow_name, super_success_run, re_run_id
+    run_command,flow_name, super_success_run, re_run_id,log_id
   </sql>
   <sql id="Blob_Column_List">
     run_source
@@ -166,6 +167,9 @@
       <if test="reRunId != null">
         re_run_id,
       </if>
+      <if test="logId != null">
+        log_id,
+      </if>
     </trim>
     <trim prefix="values (" suffix=")" suffixOverrides=",">
       <if test="id != null">
@@ -261,6 +265,9 @@
       <if test="reRunId != null">
         #{reRunId,jdbcType=VARCHAR},
       </if>
+      <if test="logId != null">
+        #{logId,jdbcType=INTEGER},
+      </if>
     </trim>
   </insert>
 
@@ -270,7 +277,7 @@
     job_type, handler_name, node_desc, node_name, map_flow_id, node_timeout, is_virtual,
     plugin_urls, priority, failed_retry_interval, routing_strategy, run_id, run_param,
     run_source_desc, script_urls, source_principal, trigger_time, trigger_status, version_name,
-    run_command,run_source,flow_name,super_success_run, re_run_id
+    run_command,run_source,flow_name,super_success_run, re_run_id ,log_id
     ) values
     <foreach collection="jobTasks" item="jobTask" separator =",">
       (
@@ -282,7 +289,8 @@
       #{jobTask.runParam,jdbcType=VARCHAR},#{jobTask.runSourceDesc,jdbcType=VARCHAR},#{jobTask.scriptUrls,jdbcType=VARCHAR},
       #{jobTask.sourcePrincipal,jdbcType=VARCHAR},#{jobTask.triggerTime,jdbcType=BIGINT},#{jobTask.triggerStatus,jdbcType=CHAR},
       #{jobTask.versionName,jdbcType=VARCHAR},#{jobTask.runCommand,jdbcType=VARCHAR},#{jobTask.runSource,jdbcType=LONGVARCHAR},
-      #{jobTask.flowName,jdbcType=VARCHAR},#{jobTask.superSuccessRun,jdbcType=CHAR},#{jobTask.reRunId,jdbcType=VARCHAR}
+      #{jobTask.flowName,jdbcType=VARCHAR},#{jobTask.superSuccessRun,jdbcType=CHAR},#{jobTask.reRunId,jdbcType=VARCHAR},
+      #{jobTask.logId,jdbcType=INTEGER}
 
       )
     </foreach>
@@ -382,6 +390,9 @@
       <if test="reRunId != null">
         re_run_id = #{reRunId,jdbcType=VARCHAR},
       </if>
+      <if test="logId != null">
+        logId = #{logId,jdbcType=INTEGER},
+      </if>
     </set>
     where id = #{id,jdbcType=INTEGER}
   </update>

byit-myth-core/myth-admin-core/src/main/resources/mapper/JobTaskRunLogMapper.xml

@@ -28,6 +28,7 @@
     <result column="re_run_id" jdbcType="VARCHAR" property="reRunId" />
     <result column="log_file_name" jdbcType="VARCHAR" property="logFileName" />
     <result column="super_success_run" jdbcType="CHAR" property="superSuccessRun"/>
+    <result column="run_count" jdbcType="INTEGER" property="runCount"/>
   </resultMap>
   <resultMap extends="BaseResultMap" id="ResultMapWithBLOBs" type="com.byit.model.JobTaskRunLogWithBLOBs">
     <result column="run_msg" jdbcType="LONGVARCHAR" property="runMsg" />
@@ -37,7 +38,7 @@
     log_id, failed_remaining_count, version_name, flow_id, flow_name, job_group_id, handler_name, 
     node_name, is_virtual, run_code, run_params, start_time, run_type, trigger_code, 
     trigger_time, job_group_ip, map_flow_id, run_command, end_time, node_id,job_type,alert_end,run_id,re_run_id
-    ,log_file_name, super_success_run
+    ,log_file_name, super_success_run, run_count
   </sql>
   <sql id="Blob_Column_List">
     run_msg, trigger_msg
@@ -56,6 +57,13 @@
           </foreach>
   </select>
 
+  <select id="findErrorNode" resultMap="BaseResultMap">
+    select <include refid="Base_Column_List" />
+    from job_task_run_log
+    where run_code != '6' and (run_code = '2' || run_code = '4') and failed_remaining_count > 0
+    and is_virtual = '1'
+  </select>
+
   <select id="findByRunIdAndNodeId" resultMap="BaseResultMap">
     select <include refid="Base_Column_List"/>
     from job_task_run_log
@@ -183,6 +191,9 @@
       <if test="superSuccessRun != null">
         super_success_run,
       </if>
+      <if test="runCount != null">
+        run_count,
+      </if>
     </trim>
     <trim prefix="values (" suffix=")" suffixOverrides=",">
       <if test="logId != null">
@@ -269,6 +280,9 @@
       <if test="superSuccessRun != null">
         #{superSuccessRun,jdbcType=CHAR},
       </if>
+      <if test="runCount != null">
+        #{runCount,jdbcType=INTEGER},
+      </if>
     </trim>
   </insert>
   <update id="updateJobTaskRunLogWithBLOBs" parameterType="com.byit.model.JobTaskRunLogWithBLOBs">
@@ -355,6 +369,9 @@
       <if test="superSuccessRun != null">
         super_success_run = #{superSuccessRun,jdbcType=CHAR},
       </if>
+      <if test="superSuccessRun != null">
+        run_count = #{runCount,jdbcType=INTEGER},
+      </if>
     </set>
     where log_id = #{logId,jdbcType=INTEGER}
   </update>
@@ -436,6 +453,9 @@
       <if test="superSuccessRun != null">
         super_success_run = #{superSuccessRun,jdbcType=CHAR},
       </if>
+      <if test="runCount != null">
+        run_count = #{runCount,jdbcType=INTEGER},
+      </if>
     </set>
     where log_id = #{logId,jdbcType=INTEGER}
   </update>