Merge remote-tracking branch 'origin/developer' into developer

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

@@ -280,7 +280,9 @@ public class ApiFlowServiceImpl implements ApiFlowService {
             //设置失败重试
             if (null != ((PluginNode)pluginNode).getConfig().getFailedRetryCount()){
                 node.setFailedRetryCount(((PluginNode)pluginNode).getConfig().getFailedRetryCount());
-                node.setFailedRetryInterval(((PluginNode)pluginNode).getConfig().getFailedRetryInterval());
+                node.setFailedRetryInterval(null == ((PluginNode)pluginNode).getConfig().getFailedRetryInterval() ? 3000 : ((PluginNode)pluginNode).getConfig().getFailedRetryInterval());
+            }else {
+                node.setFailedRetryCount(0);
             }
             //设置剩余执行次数
             if (null != ((PluginNode)pluginNode).getConfig().getRepeatCount() && !"-1".equals(((PluginNode)pluginNode).getConfig().getRepeatCount())){

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

@@ -125,4 +125,12 @@ public interface JobTaskRunLogMapper {
                                                @Param("flowIdList")List<Integer> flowIdList);
 
     StatisticData findStatisticDataByRunIdAndFlowId(@Param("runId")String runId, @Param("flowId")Integer flowId);
+
+    /**
+     * 根据runid和flowidlist获取运行日志
+     * @param runId
+     * @param flowIdList
+     * @return
+     */
+    List<JobTaskRunLog> findbyRunIdAndFlowIdList(@Param("runId")String runId, @Param("flowIdList")List<Integer> flowIdList);
 }
\ No newline at end of file

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

@@ -137,7 +137,7 @@ public interface RunRecordingMapper {
      * @param runId
      * @return
      */
-    RunRecording findUnFinishByRunId(@Param("runId")String runId, @Param("flowId")Integer flowId);
+    List<RunRecording> findUnFinishByRunId(@Param("runId")String runId, @Param("flowIdList")List<Integer> flowIdList);
 
     List<RunRecording> findByStartAndEndTime(@Param("startDate")Date startDate,
                                              @Param("endDate")Date endDate,

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

@@ -364,27 +364,45 @@ public class FlowServiceImpl implements FlowService {
         ValidationUtil.dataNotBank(flowName, "工作流名称不允许为空！");
         Flow flow = flowMapper.getByWorkSpaceAndName(workspace.getWorkspaceId(), flowName);
         ValidationUtil.dataNotNull(flow, flowName + "工作流不存在");
+
+        List<Integer> flowIdList = new ArrayList<>();
+        Queue<Integer> queue = new LinkedList<>();
+        flowIdList.add(flow.getFlowId());
+        queue.offer(flow.getFlowId());
+        while(!queue.isEmpty()){
+            List<Node> innerFlowList = nodeMapper.findVirtualByFlowId(queue.poll());
+            if (null != innerFlowList && innerFlowList.size() > 0){
+                innerFlowList.forEach(node -> {
+                    flowIdList.add(node.getMapFlowId());
+                    queue.offer(node.getMapFlowId());
+                });
+            }
+        }
         //杀死运行实例
-        RunRecording runRecording = runRecordingMapper.findUnFinishByRunId(runId, flow.getFlowId());
-        ValidationUtil.dataNotNull(runRecording, "该运行id未找到或已经运行完成无需杀死！");
+        List<RunRecording> runRecordingList = runRecordingMapper.findUnFinishByRunId(runId, flowIdList);
+        ValidationUtil.dataNotNull(runRecordingList, "该运行id未找到或已经运行完成无需杀死！");
         Boolean result = true;
-        //将未完成的运行实例设置为完成且杀死
-        runRecording.setFlowStatus("4");
-        runRecording.setFlowRunResult("5");
-        runRecordingMapper.updateRunRecordingById(runRecording);
+        runRecordingList.forEach(runRecording -> {
+            //将未完成的运行实例设置为完成且杀死
+            runRecording.setFlowStatus("4");
+            runRecording.setFlowRunResult("5");
+            runRecordingMapper.updateRunRecordingById(runRecording);
+        });
         //杀死所有的任务
-        List<JobTaskRunLogWithBLOBs> jobTaskRunLogList = jobTaskRunLogMapper.findJobTaskRunLogWithBLOBsByFlowIdAndRunId(flow.getFlowId(), runId);
-        for (JobTaskRunLogWithBLOBs jobTaskRunLog : jobTaskRunLogList){
-            if ("0".equals(jobTaskRunLog.getRunCode())){
-                if (StringUtils.isEmpty(jobTaskRunLog.getJobGroupIp())){
-                    log.warn("已经调度成功但是还未返回具体的调用机器的ip地址");
-                    Thread.sleep(3000);
-                    jobTaskRunLog = jobTaskRunLogMapper.findJobTaskRunLogByLogId(jobTaskRunLog.getLogId());
-                    ValidationUtil.isTrueValidation(!"0".equals(jobTaskRunLog.getRunCode()), "该任务已经运行结束！");
-                    ValidationUtil.dataNotBank(jobTaskRunLog.getJobGroupIp(), "尚未分配执行机请稍后再试");
-                }
-                if (! killJob(jobTaskRunLog.getLogId(), jobTaskRunLog.getJobGroupIp())){
-                    result = false;
+        List<JobTaskRunLog> jobTaskRunLogList = jobTaskRunLogMapper.findbyRunIdAndFlowIdList(runId, flowIdList);
+        if (null != jobTaskRunLogList && jobTaskRunLogList.size() > 0){
+            for (JobTaskRunLog jobTaskRunLog : jobTaskRunLogList){
+                if ("0".equals(jobTaskRunLog.getRunCode())){
+                    if (StringUtils.isEmpty(jobTaskRunLog.getJobGroupIp())){
+                        log.warn("已经调度成功但是还未返回具体的调用机器的ip地址");
+                        Thread.sleep(3000);
+                        jobTaskRunLog = jobTaskRunLogMapper.findJobTaskRunLogByLogId(jobTaskRunLog.getLogId());
+                        ValidationUtil.isTrueValidation(!"0".equals(jobTaskRunLog.getRunCode()), "该任务已经运行结束！");
+                        ValidationUtil.dataNotBank(jobTaskRunLog.getJobGroupIp(), "尚未分配执行机请稍后再试");
+                    }
+                    if (! killJob(jobTaskRunLog.getLogId(), jobTaskRunLog.getJobGroupIp())){
+                        result = false;
+                    }
                 }
             }
         }

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

@@ -217,7 +217,7 @@
   <insert id="saveList" parameterType="com.byit.model.FlowStatusSnapshoot">
     INSERT INTO flow_status_snapshoot
     (
-        status_id, workspace_id, flow_id,
+        workspace_id, flow_id,
         flow_name, `day`, `hour`,
         flow_status, snapshoot_time, unstart_node,
         runing_node, success_node, fail_node,
@@ -226,7 +226,6 @@
     VALUES
     <foreach collection="flowStatusSnapshootList" item="flowStatusSnapshoot" separator=",">
     (
-      #{flowStatusSnapshoot.statusId,jdbcType=INTEGER},
       #{flowStatusSnapshoot.workspaceId,jdbcType=INTEGER},
       #{flowStatusSnapshoot.flowId,jdbcType=INTEGER},
       #{flowStatusSnapshoot.flowName,jdbcType=VARCHAR},

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

@@ -148,8 +148,20 @@
       where run_id = #{runId}
             and flow_id = #{flowId}
     </select>
+  <select id="findbyRunIdAndFlowIdList" resultMap="BaseResultMap">
+    select  <include refid="Base_Column_List" />
+    from job_task_run_log
+    where run_id = #{runId}
+      <if test="flowIdList != null">
+        and flow_id in (
+        <foreach collection="flowIdList" item="flowId" separator=",">
+          #{flowId}
+        </foreach>
+        )
+      </if>
+  </select>
 
-    <delete id="deleteById" parameterType="java.lang.Integer">
+  <delete id="deleteById" parameterType="java.lang.Integer">
     delete from job_task_run_log
     where log_id = #{logId,jdbcType=INTEGER}
   </delete>

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

@@ -103,8 +103,14 @@
     <include refid="Base_Column_List" />
     from run_recording
     where run_id = #{runId,jdbcType=VARCHAR}
-        and flow_id = #{flowId,jdbcType=INTEGER}
         and flow_status != '4'
+        <if test="flowIdList != null">
+          and flow_id in (
+          <foreach collection="flowIdList" item="flowId" separator=",">
+            #{flowId}
+          </foreach>
+          )
+        </if>
   </select>
   <select id="findByStartAndEndTime" resultMap="BaseResultMap">
     select

byit-myth-executor/myth-executor-plugin/src/main/java/com/byit/utils/JobUtils.java

@@ -268,9 +268,13 @@ public class JobUtils {
      * @param runId
      * @return
      */
-    public static ResponseResult killFlow(String runId){
+    public static ResponseResult killFlow(String runId, String flowName, String workspaceName){
+        Map<String,String> map = new HashMap<>(5);
+        map.put("runId", runId);
+        map.put("flowName", flowName);
+        map.put("workspaceName", workspaceName);
         //发送请求 添加任务
-        String response = createHttpRequest(REQUEST_FLOW_KILL_FLOW, "runId=" + runId);
+        String response = createHttpRequest(REQUEST_FLOW_KILL_FLOW, "param="+JSON.toJSONString(map,WriteClassName));
         log.info("--------------------杀死工作流接口调用成功，结果为：{}------------------------",response);
         return JSON.parseObject(response, ResponseResult.class);
     }