|
|
@ -222,7 +222,8 @@ public class ExecutorServiceImpl extends BaseServiceImpl implements ExecutorServ |
|
|
|
Long environmentCode, Integer timeout, |
|
|
|
Long environmentCode, Integer timeout, |
|
|
|
Map<String, String> startParams, Integer expectedParallelismNumber, |
|
|
|
Map<String, String> startParams, Integer expectedParallelismNumber, |
|
|
|
int dryRun, int testFlag, |
|
|
|
int dryRun, int testFlag, |
|
|
|
ComplementDependentMode complementDependentMode, Integer version) { |
|
|
|
ComplementDependentMode complementDependentMode, Integer version, |
|
|
|
|
|
|
|
boolean allLevelDependent) { |
|
|
|
Project project = projectMapper.queryByCode(projectCode); |
|
|
|
Project project = projectMapper.queryByCode(projectCode); |
|
|
|
// check user access for project
|
|
|
|
// check user access for project
|
|
|
|
Map<String, Object> result = |
|
|
|
Map<String, Object> result = |
|
|
@ -264,7 +265,7 @@ public class ExecutorServiceImpl extends BaseServiceImpl implements ExecutorServ |
|
|
|
cronTime, warningType, loginUser.getId(), warningGroupId, runMode, processInstancePriority, |
|
|
|
cronTime, warningType, loginUser.getId(), warningGroupId, runMode, processInstancePriority, |
|
|
|
workerGroup, tenantCode, |
|
|
|
workerGroup, tenantCode, |
|
|
|
environmentCode, startParams, expectedParallelismNumber, dryRun, testFlag, |
|
|
|
environmentCode, startParams, expectedParallelismNumber, dryRun, testFlag, |
|
|
|
complementDependentMode); |
|
|
|
complementDependentMode, allLevelDependent); |
|
|
|
|
|
|
|
|
|
|
|
if (create > 0) { |
|
|
|
if (create > 0) { |
|
|
|
processDefinition.setWarningGroupId(warningGroupId); |
|
|
|
processDefinition.setWarningGroupId(warningGroupId); |
|
|
@ -427,8 +428,6 @@ public class ExecutorServiceImpl extends BaseServiceImpl implements ExecutorServ |
|
|
|
|
|
|
|
|
|
|
|
/** |
|
|
|
/** |
|
|
|
* do action to workflow instance:pause, stop, repeat, recover from pause, recover from stop,rerun failed task |
|
|
|
* do action to workflow instance:pause, stop, repeat, recover from pause, recover from stop,rerun failed task |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
* |
|
|
|
* |
|
|
|
* @param loginUser login user |
|
|
|
* @param loginUser login user |
|
|
|
* @param workflowInstanceId workflow instance id |
|
|
|
* @param workflowInstanceId workflow instance id |
|
|
@ -740,6 +739,7 @@ public class ExecutorServiceImpl extends BaseServiceImpl implements ExecutorServ |
|
|
|
* @param workerGroup workerGroup |
|
|
|
* @param workerGroup workerGroup |
|
|
|
* @param testFlag testFlag |
|
|
|
* @param testFlag testFlag |
|
|
|
* @param environmentCode environmentCode |
|
|
|
* @param environmentCode environmentCode |
|
|
|
|
|
|
|
* @param allLevelDependent allLevelDependent |
|
|
|
* @return command id |
|
|
|
* @return command id |
|
|
|
*/ |
|
|
|
*/ |
|
|
|
private int createCommand(Long triggerCode, CommandType commandType, long processDefineCode, TaskDependType nodeDep, |
|
|
|
private int createCommand(Long triggerCode, CommandType commandType, long processDefineCode, TaskDependType nodeDep, |
|
|
@ -748,7 +748,8 @@ public class ExecutorServiceImpl extends BaseServiceImpl implements ExecutorServ |
|
|
|
Priority processInstancePriority, String workerGroup, String tenantCode, |
|
|
|
Priority processInstancePriority, String workerGroup, String tenantCode, |
|
|
|
Long environmentCode, |
|
|
|
Long environmentCode, |
|
|
|
Map<String, String> startParams, Integer expectedParallelismNumber, int dryRun, |
|
|
|
Map<String, String> startParams, Integer expectedParallelismNumber, int dryRun, |
|
|
|
int testFlag, ComplementDependentMode complementDependentMode) { |
|
|
|
int testFlag, ComplementDependentMode complementDependentMode, |
|
|
|
|
|
|
|
boolean allLevelDependent) { |
|
|
|
|
|
|
|
|
|
|
|
/** |
|
|
|
/** |
|
|
|
* instantiate command schedule instance |
|
|
|
* instantiate command schedule instance |
|
|
@ -807,7 +808,7 @@ public class ExecutorServiceImpl extends BaseServiceImpl implements ExecutorServ |
|
|
|
log.info("Start to create {} command, processDefinitionCode:{}.", |
|
|
|
log.info("Start to create {} command, processDefinitionCode:{}.", |
|
|
|
command.getCommandType().getDescp(), processDefineCode); |
|
|
|
command.getCommandType().getDescp(), processDefineCode); |
|
|
|
return createComplementCommandList(triggerCode, schedule, runMode, command, expectedParallelismNumber, |
|
|
|
return createComplementCommandList(triggerCode, schedule, runMode, command, expectedParallelismNumber, |
|
|
|
complementDependentMode); |
|
|
|
complementDependentMode, allLevelDependent); |
|
|
|
} catch (CronParseException cronParseException) { |
|
|
|
} catch (CronParseException cronParseException) { |
|
|
|
// We catch the exception here just to make compiler happy, since we have already validated the schedule
|
|
|
|
// We catch the exception here just to make compiler happy, since we have already validated the schedule
|
|
|
|
// cron expression before
|
|
|
|
// cron expression before
|
|
|
@ -834,7 +835,8 @@ public class ExecutorServiceImpl extends BaseServiceImpl implements ExecutorServ |
|
|
|
protected int createComplementCommandList(Long triggerCode, String scheduleTimeParam, RunMode runMode, |
|
|
|
protected int createComplementCommandList(Long triggerCode, String scheduleTimeParam, RunMode runMode, |
|
|
|
Command command, |
|
|
|
Command command, |
|
|
|
Integer expectedParallelismNumber, |
|
|
|
Integer expectedParallelismNumber, |
|
|
|
ComplementDependentMode complementDependentMode) throws CronParseException { |
|
|
|
ComplementDependentMode complementDependentMode, |
|
|
|
|
|
|
|
boolean allLevelDependent) throws CronParseException { |
|
|
|
int createCount = 0; |
|
|
|
int createCount = 0; |
|
|
|
String startDate = null; |
|
|
|
String startDate = null; |
|
|
|
String endDate = null; |
|
|
|
String endDate = null; |
|
|
@ -894,7 +896,8 @@ public class ExecutorServiceImpl extends BaseServiceImpl implements ExecutorServ |
|
|
|
log.info( |
|
|
|
log.info( |
|
|
|
"Complement dependent mode is all dependent and Scheduler is not empty, need create complement dependent command, processDefinitionCode:{}.", |
|
|
|
"Complement dependent mode is all dependent and Scheduler is not empty, need create complement dependent command, processDefinitionCode:{}.", |
|
|
|
command.getProcessDefinitionCode()); |
|
|
|
command.getProcessDefinitionCode()); |
|
|
|
dependentProcessDefinitionCreateCount += createComplementDependentCommand(schedules, command); |
|
|
|
dependentProcessDefinitionCreateCount += |
|
|
|
|
|
|
|
createComplementDependentCommand(schedules, command, allLevelDependent); |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
if (createCount > 0) { |
|
|
|
if (createCount > 0) { |
|
|
@ -963,7 +966,7 @@ public class ExecutorServiceImpl extends BaseServiceImpl implements ExecutorServ |
|
|
|
"Complement dependent mode is all dependent and Scheduler is not empty, need create complement dependent command, processDefinitionCode:{}.", |
|
|
|
"Complement dependent mode is all dependent and Scheduler is not empty, need create complement dependent command, processDefinitionCode:{}.", |
|
|
|
command.getProcessDefinitionCode()); |
|
|
|
command.getProcessDefinitionCode()); |
|
|
|
dependentProcessDefinitionCreateCount += |
|
|
|
dependentProcessDefinitionCreateCount += |
|
|
|
createComplementDependentCommand(schedules, command); |
|
|
|
createComplementDependentCommand(schedules, command, allLevelDependent); |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
@ -1004,7 +1007,7 @@ public class ExecutorServiceImpl extends BaseServiceImpl implements ExecutorServ |
|
|
|
/** |
|
|
|
/** |
|
|
|
* create complement dependent command |
|
|
|
* create complement dependent command |
|
|
|
*/ |
|
|
|
*/ |
|
|
|
public int createComplementDependentCommand(List<Schedule> schedules, Command command) { |
|
|
|
public int createComplementDependentCommand(List<Schedule> schedules, Command command, boolean allLevelDependent) { |
|
|
|
int dependentProcessDefinitionCreateCount = 0; |
|
|
|
int dependentProcessDefinitionCreateCount = 0; |
|
|
|
Command dependentCommand; |
|
|
|
Command dependentCommand; |
|
|
|
|
|
|
|
|
|
|
@ -1017,7 +1020,8 @@ public class ExecutorServiceImpl extends BaseServiceImpl implements ExecutorServ |
|
|
|
|
|
|
|
|
|
|
|
List<DependentProcessDefinition> dependentProcessDefinitionList = |
|
|
|
List<DependentProcessDefinition> dependentProcessDefinitionList = |
|
|
|
getComplementDependentDefinitionList(dependentCommand.getProcessDefinitionCode(), |
|
|
|
getComplementDependentDefinitionList(dependentCommand.getProcessDefinitionCode(), |
|
|
|
CronUtils.getMaxCycle(schedules.get(0).getCrontab()), dependentCommand.getWorkerGroup()); |
|
|
|
CronUtils.getMaxCycle(schedules.get(0).getCrontab()), dependentCommand.getWorkerGroup(), |
|
|
|
|
|
|
|
allLevelDependent); |
|
|
|
dependentCommand.setTaskDependType(TaskDependType.TASK_POST); |
|
|
|
dependentCommand.setTaskDependType(TaskDependType.TASK_POST); |
|
|
|
for (DependentProcessDefinition dependentProcessDefinition : dependentProcessDefinitionList) { |
|
|
|
for (DependentProcessDefinition dependentProcessDefinition : dependentProcessDefinitionList) { |
|
|
|
// If the id is Integer, the auto-increment id will be obtained by mybatis-plus
|
|
|
|
// If the id is Integer, the auto-increment id will be obtained by mybatis-plus
|
|
|
@ -1041,12 +1045,38 @@ public class ExecutorServiceImpl extends BaseServiceImpl implements ExecutorServ |
|
|
|
*/ |
|
|
|
*/ |
|
|
|
private List<DependentProcessDefinition> getComplementDependentDefinitionList(long processDefinitionCode, |
|
|
|
private List<DependentProcessDefinition> getComplementDependentDefinitionList(long processDefinitionCode, |
|
|
|
CycleEnum processDefinitionCycle, |
|
|
|
CycleEnum processDefinitionCycle, |
|
|
|
String workerGroup) { |
|
|
|
String workerGroup, |
|
|
|
|
|
|
|
boolean allLevelDependent) { |
|
|
|
List<DependentProcessDefinition> dependentProcessDefinitionList = |
|
|
|
List<DependentProcessDefinition> dependentProcessDefinitionList = |
|
|
|
processService.queryDependentProcessDefinitionByProcessDefinitionCode(processDefinitionCode); |
|
|
|
checkDependentProcessDefinitionValid( |
|
|
|
|
|
|
|
processService.queryDependentProcessDefinitionByProcessDefinitionCode(processDefinitionCode), |
|
|
|
|
|
|
|
processDefinitionCycle, workerGroup, |
|
|
|
|
|
|
|
processDefinitionCode); |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
if (dependentProcessDefinitionList.isEmpty()) { |
|
|
|
|
|
|
|
return dependentProcessDefinitionList; |
|
|
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
return checkDependentProcessDefinitionValid(dependentProcessDefinitionList, processDefinitionCycle, |
|
|
|
if (allLevelDependent) { |
|
|
|
workerGroup, processDefinitionCode); |
|
|
|
List<DependentProcessDefinition> childList = new ArrayList<>(dependentProcessDefinitionList); |
|
|
|
|
|
|
|
while (true) { |
|
|
|
|
|
|
|
List<DependentProcessDefinition> childDependentList = childList |
|
|
|
|
|
|
|
.stream() |
|
|
|
|
|
|
|
.flatMap(dependentProcessDefinition -> checkDependentProcessDefinitionValid( |
|
|
|
|
|
|
|
processService.queryDependentProcessDefinitionByProcessDefinitionCode( |
|
|
|
|
|
|
|
dependentProcessDefinition.getProcessDefinitionCode()), |
|
|
|
|
|
|
|
processDefinitionCycle, |
|
|
|
|
|
|
|
workerGroup, |
|
|
|
|
|
|
|
dependentProcessDefinition.getProcessDefinitionCode()).stream()) |
|
|
|
|
|
|
|
.collect(Collectors.toList()); |
|
|
|
|
|
|
|
if (childDependentList.isEmpty()) { |
|
|
|
|
|
|
|
break; |
|
|
|
|
|
|
|
} |
|
|
|
|
|
|
|
dependentProcessDefinitionList.addAll(childDependentList); |
|
|
|
|
|
|
|
childList = new ArrayList<>(childDependentList); |
|
|
|
|
|
|
|
} |
|
|
|
|
|
|
|
} |
|
|
|
|
|
|
|
return dependentProcessDefinitionList; |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
/** |
|
|
|
/** |
|
|
|