Browse Source

fix bug#1336 serial complement data can have multiple process instances (#3317)

Co-authored-by: lenboo <baoliang@analysys.com.cn>
pull/3/MERGE
bao liang 4 years ago committed by GitHub
parent
commit
624f0aeab9
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
  1. 18
      dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterExecThread.java

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

@ -252,6 +252,9 @@ public class MasterExecThread implements Runnable {
} }
while(Stopper.isRunning()){ while(Stopper.isRunning()){
logger.info("process {} start to complement {} data",
processInstance.getId(), DateUtils.dateToString(scheduleDate));
// prepare dag and other info // prepare dag and other info
prepareProcess(); prepareProcess();
@ -266,13 +269,13 @@ public class MasterExecThread implements Runnable {
// execute process ,waiting for end // execute process ,waiting for end
runProcess(); runProcess();
endProcess();
// process instance failure ,no more complements // process instance failure ,no more complements
if(!processInstance.getState().typeIsSuccess()){ if(!processInstance.getState().typeIsSuccess()){
logger.info("process {} state {}, complement not completely!", logger.info("process {} state {}, complement not completely!",
processInstance.getId(), processInstance.getState()); processInstance.getId(), processInstance.getState());
break; break;
} }
// current process instance success ,next execute // current process instance success ,next execute
if(null == iterator){ if(null == iterator){
// loop by day // loop by day
@ -291,9 +294,7 @@ public class MasterExecThread implements Runnable {
} }
scheduleDate = iterator.next(); scheduleDate = iterator.next();
} }
// flow end
logger.info("process {} start to complement {} data",
processInstance.getId(), DateUtils.dateToString(scheduleDate));
// execute next process instance complement data // execute next process instance complement data
processInstance.setScheduleTime(scheduleDate); processInstance.setScheduleTime(scheduleDate);
if(cmdParam.containsKey(Constants.CMDPARAM_RECOVERY_START_NODE_STRING)){ if(cmdParam.containsKey(Constants.CMDPARAM_RECOVERY_START_NODE_STRING)){
@ -301,22 +302,15 @@ public class MasterExecThread implements Runnable {
processInstance.setCommandParam(JSONUtils.toJsonString(cmdParam)); processInstance.setCommandParam(JSONUtils.toJsonString(cmdParam));
} }
List<TaskInstance> taskInstanceList = processService.findValidTaskListByProcessId(processInstance.getId());
for(TaskInstance taskInstance : taskInstanceList){
taskInstance.setFlag(Flag.NO);
processService.updateTaskInstance(taskInstance);
}
processInstance.setState(ExecutionStatus.RUNNING_EXEUTION); processInstance.setState(ExecutionStatus.RUNNING_EXEUTION);
processInstance.setGlobalParams(ParameterUtils.curingGlobalParams( processInstance.setGlobalParams(ParameterUtils.curingGlobalParams(
processInstance.getProcessDefinition().getGlobalParamMap(), processInstance.getProcessDefinition().getGlobalParamMap(),
processInstance.getProcessDefinition().getGlobalParamList(), processInstance.getProcessDefinition().getGlobalParamList(),
CommandType.COMPLEMENT_DATA, processInstance.getScheduleTime())); CommandType.COMPLEMENT_DATA, processInstance.getScheduleTime()));
processInstance.setId(0);
processService.saveProcessInstance(processInstance); processService.saveProcessInstance(processInstance);
} }
// flow end
endProcess();
} }

Loading…
Cancel
Save