github
/
DolphinScheduler
mirror of https://gitee.com/dolphinscheduler/DolphinScheduler.git
1
0
Fork 0
Code Issues Releases Wiki Activity
Browse Source

[Improvement][Code style] FIX SPELL WAITTING TO WAITING , etc. (#4118) 

* FIX SPELL

* FIX SPELL AND  Optimizing code conventions

* add ut  cannot construct process instance, return null;

* add ut testExportProcessMetaData

* add ut testExportProcessMetaData

* add ut testImportProcessSchedule

* add ut MasterExecThreadTest

* add ut MasterExecThreadTest

* add ut testSubProcessViewTree

* add ut testComplementWithStartNodeList

* add ut testRecurseFindSubProcessId

* add ut testRecurseFindSubProcessId

* add ut testRecurseFindSubProcessId
pull/3/MERGE
felix.wang 4 years ago committed by GitHub
parent
b46044d592
commit
3e411d075f
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
9 changed files with 849 additions and 386 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 193
      dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/ExecutorService.java
  2. 30
      dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ProcessDefinitionServiceImpl.java
  3. 28
      dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/ExecutorService2Test.java
  4. 170
      dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/ProcessDefinitionServiceTest.java
  5. 16
      dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/Constants.java
  6. 406
      dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterExecThread.java
  7. 117
      dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/master/MasterExecThreadTest.java
  8. 64
      dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessService.java
  9. 211
      dolphinscheduler-service/src/test/java/org/apache/dolphinscheduler/service/process/ProcessServiceTest.java

193
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/ExecutorService.java
Unescape View File

@ -14,39 +14,62 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dolphinscheduler.api.service;
import static org.apache.dolphinscheduler.common.Constants.CMDPARAM_COMPLEMENT_DATA_END_DATE;
import static org.apache.dolphinscheduler.common.Constants.CMDPARAM_COMPLEMENT_DATA_START_DATE;
import static org.apache.dolphinscheduler.common.Constants.CMD_PARAM_RECOVER_PROCESS_ID_STRING;
import static org.apache.dolphinscheduler.common.Constants.CMD_PARAM_START_NODE_NAMES;
import static org.apache.dolphinscheduler.common.Constants.MAX_TASK_TIMEOUT;
import org.apache.dolphinscheduler.api.enums.ExecuteType;
import org.apache.dolphinscheduler.api.enums.Status;
import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.common.enums.*;
import org.apache.dolphinscheduler.common.enums.CommandType;
import org.apache.dolphinscheduler.common.enums.ExecutionStatus;
import org.apache.dolphinscheduler.common.enums.FailureStrategy;
import org.apache.dolphinscheduler.common.enums.Priority;
import org.apache.dolphinscheduler.common.enums.ReleaseState;
import org.apache.dolphinscheduler.common.enums.RunMode;
import org.apache.dolphinscheduler.common.enums.TaskDependType;
import org.apache.dolphinscheduler.common.enums.WarningType;
import org.apache.dolphinscheduler.common.model.Server;
import org.apache.dolphinscheduler.common.utils.CollectionUtils;
import org.apache.dolphinscheduler.common.utils.DateUtils;
import org.apache.dolphinscheduler.common.utils.*;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.common.utils.StringUtils;
import org.apache.dolphinscheduler.dao.entity.*;
import org.apache.dolphinscheduler.dao.entity.Command;
import org.apache.dolphinscheduler.dao.entity.ProcessDefinition;
import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
import org.apache.dolphinscheduler.dao.entity.Project;
import org.apache.dolphinscheduler.dao.entity.Schedule;
import org.apache.dolphinscheduler.dao.entity.Tenant;
import org.apache.dolphinscheduler.dao.entity.User;
import org.apache.dolphinscheduler.dao.mapper.ProcessDefinitionMapper;
import org.apache.dolphinscheduler.dao.mapper.ProcessInstanceMapper;
import org.apache.dolphinscheduler.dao.mapper.ProjectMapper;
import org.apache.dolphinscheduler.service.process.ProcessService;
import org.apache.dolphinscheduler.service.quartz.cron.CronUtils;
import java.text.ParseException;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.text.ParseException;
import java.util.*;
import static org.apache.dolphinscheduler.common.Constants.*;
/**
* executor service
*/
@Service
public class ExecutorService extends BaseService{
public class ExecutorService extends BaseService {
private static final Logger logger = LoggerFactory.getLogger(ExecutorService.class);
@ -73,22 +96,22 @@ public class ExecutorService extends BaseService{
/**
* execute process instance
*
* @param loginUser login user
* @param projectName project name
* @param processDefinitionId process Definition Id
* @param cronTime cron time
* @param commandType command type
* @param failureStrategy failuer strategy
* @param startNodeList start nodelist
* @param taskDependType node dependency type
* @param warningType warning type
* @param warningGroupId notify group id
* @param receivers receivers
* @param receiversCc receivers cc
* @param loginUser login user
* @param projectName project name
* @param processDefinitionId process Definition Id
* @param cronTime cron time
* @param commandType command type
* @param failureStrategy failuer strategy
* @param startNodeList start nodelist
* @param taskDependType node dependency type
* @param warningType warning type
* @param warningGroupId notify group id
* @param receivers receivers
* @param receiversCc receivers cc
* @param processInstancePriority process instance priority
* @param workerGroup worker group name
* @param runMode run mode
* @param timeout timeout
* @param timeout timeout
* @return execute process instance code
* @throws ParseException Parse Exception
*/
@ -101,23 +124,23 @@ public class ExecutorService extends BaseService{
Map<String, Object> result = new HashMap<>();
// timeout is invalid
if (timeout <= 0 || timeout > MAX_TASK_TIMEOUT) {
putMsg(result,Status.TASK_TIMEOUT_PARAMS_ERROR);
putMsg(result, Status.TASK_TIMEOUT_PARAMS_ERROR);
return result;
}
Project project = projectMapper.queryByName(projectName);
Map<String, Object> checkResultAndAuth = checkResultAndAuth(loginUser, projectName, project);
if (checkResultAndAuth != null){
if (checkResultAndAuth != null) {
return checkResultAndAuth;
}
// check process define release state
ProcessDefinition processDefinition = processDefinitionMapper.selectById(processDefinitionId);
result = checkProcessDefinitionValid(processDefinition, processDefinitionId);
if(result.get(Constants.STATUS) != Status.SUCCESS){
if (result.get(Constants.STATUS) != Status.SUCCESS) {
return result;
}
if (!checkTenantSuitable(processDefinition)){
if (!checkTenantSuitable(processDefinition)) {
logger.error("there is not any valid tenant for the process definition: id:{},name:{}, ",
processDefinition.getId(), processDefinition.getName());
putMsg(result, Status.TENANT_NOT_SUITABLE);
@ -129,14 +152,13 @@ public class ExecutorService extends BaseService{
return result;
}
/**
* create command
*/
int create = this.createCommand(commandType, processDefinitionId,
taskDependType, failureStrategy, startNodeList, cronTime, warningType, loginUser.getId(),
warningGroupId, runMode,processInstancePriority, workerGroup);
if(create > 0 ){
warningGroupId, runMode, processInstancePriority, workerGroup);
if (create > 0) {
/**
* according to the process definition ID updateProcessInstance and CC recipient
*/
@ -152,6 +174,7 @@ public class ExecutorService extends BaseService{
/**
* check whether master exists
*
* @param result result
* @return master exists return true , otherwise return false
*/
@ -167,7 +190,6 @@ public class ExecutorService extends BaseService{
return true;
}
/**
* check whether the process definition can be executed
*
@ -175,22 +197,20 @@ public class ExecutorService extends BaseService{
* @param processDefineId process definition id
* @return check result code
*/
public Map<String, Object> checkProcessDefinitionValid(ProcessDefinition processDefinition, int processDefineId){
public Map<String, Object> checkProcessDefinitionValid(ProcessDefinition processDefinition, int processDefineId) {
Map<String, Object> result = new HashMap<>();
if (processDefinition == null) {
// check process definition exists
putMsg(result, Status.PROCESS_DEFINE_NOT_EXIST,processDefineId);
putMsg(result, Status.PROCESS_DEFINE_NOT_EXIST, processDefineId);
} else if (processDefinition.getReleaseState() != ReleaseState.ONLINE) {
// check process definition online
putMsg(result, Status.PROCESS_DEFINE_NOT_RELEASE,processDefineId);
}else{
putMsg(result, Status.PROCESS_DEFINE_NOT_RELEASE, processDefineId);
} else {
result.put(Constants.STATUS, Status.SUCCESS);
}
return result;
}
/**
* do action to process instance：pause, stop, repeat, recover from pause, recover from stop
*
@ -214,7 +234,6 @@ public class ExecutorService extends BaseService{
return result;
}
ProcessInstance processInstance = processService.findProcessInstanceDetailById(processInstanceId);
if (processInstance == null) {
putMsg(result, Status.PROCESS_INSTANCE_NOT_EXIST, processInstanceId);
@ -222,7 +241,7 @@ public class ExecutorService extends BaseService{
}
ProcessDefinition processDefinition = processService.findProcessDefineById(processInstance.getProcessDefinitionId());
if(executeType != ExecuteType.STOP && executeType != ExecuteType.PAUSE){
if (executeType != ExecuteType.STOP && executeType != ExecuteType.PAUSE) {
result = checkProcessDefinitionValid(processDefinition, processInstance.getProcessDefinitionId());
if (result.get(Constants.STATUS) != Status.SUCCESS) {
return result;
@ -234,7 +253,7 @@ public class ExecutorService extends BaseService{
if (status != Status.SUCCESS) {
return checkResult;
}
if (!checkTenantSuitable(processDefinition)){
if (!checkTenantSuitable(processDefinition)) {
logger.error("there is not any valid tenant for the process definition: id:{},name:{}, ",
processDefinition.getId(), processDefinition.getName());
putMsg(result, Status.TENANT_NOT_SUITABLE);
@ -275,6 +294,7 @@ public class ExecutorService extends BaseService{
/**
* check tenant suitable
*
* @param processDefinition process definition
* @return true if tenant suitable, otherwise return false
*/
@ -315,7 +335,7 @@ public class ExecutorService extends BaseService{
}
break;
case RECOVER_SUSPENDED_PROCESS:
if (executionStatus.typeIsPause()|| executionStatus.typeIsCancel()) {
if (executionStatus.typeIsPause() || executionStatus.typeIsCancel()) {
checkResult = true;
}
break;
@ -323,7 +343,7 @@ public class ExecutorService extends BaseService{
break;
}
if (!checkResult) {
putMsg(result,Status.PROCESS_INSTANCE_STATE_OPERATION_ERROR, processInstance.getName(), executionStatus.toString(), executeType.toString());
putMsg(result, Status.PROCESS_INSTANCE_STATE_OPERATION_ERROR, processInstance.getName(), executionStatus.toString(), executeType.toString());
} else {
putMsg(result, Status.SUCCESS);
}
@ -331,7 +351,7 @@ public class ExecutorService extends BaseService{
}
/**
* prepare to update process instance command type and status
* prepare to update process instance command type and status
*
* @param processInstance process instance
* @param commandType command type
@ -370,11 +390,11 @@ public class ExecutorService extends BaseService{
command.setCommandType(commandType);
command.setProcessDefinitionId(processDefinitionId);
command.setCommandParam(String.format("{\"%s\":%d}",
CMDPARAM_RECOVER_PROCESS_ID_STRING, instanceId));
CMD_PARAM_RECOVER_PROCESS_ID_STRING, instanceId));
command.setExecutorId(loginUser.getId());
if(!processService.verifyIsNeedCreateCommand(command)){
putMsg(result, Status.PROCESS_INSTANCE_EXECUTING_COMMAND,processDefinitionId);
if (!processService.verifyIsNeedCreateCommand(command)) {
putMsg(result, Status.PROCESS_INSTANCE_EXECUTING_COMMAND, processDefinitionId);
return result;
}
@ -391,28 +411,29 @@ public class ExecutorService extends BaseService{
/**
* check if sub processes are offline before starting process definition
*
* @param processDefineId process definition id
* @return check result code
*/
public Map<String, Object> startCheckByProcessDefinedId(int processDefineId) {
Map<String, Object> result = new HashMap<>();
if (processDefineId == 0){
if (processDefineId == 0) {
logger.error("process definition id is null");
putMsg(result,Status.REQUEST_PARAMS_NOT_VALID_ERROR,"process definition id");
putMsg(result, Status.REQUEST_PARAMS_NOT_VALID_ERROR, "process definition id");
}
List<Integer> ids = new ArrayList<>();
processService.recurseFindSubProcessId(processDefineId, ids);
Integer[] idArray = ids.toArray(new Integer[ids.size()]);
if (!ids.isEmpty()){
if (!ids.isEmpty()) {
List<ProcessDefinition> processDefinitionList = processDefinitionMapper.queryDefinitionListByIdList(idArray);
if (processDefinitionList != null){
for (ProcessDefinition processDefinition : processDefinitionList){
if (processDefinitionList != null) {
for (ProcessDefinition processDefinition : processDefinitionList) {
/**
* if there is no online process, exit directly
*/
if (processDefinition.getReleaseState() != ReleaseState.ONLINE){
putMsg(result,Status.PROCESS_DEFINE_NOT_RELEASE, processDefinition.getName());
if (processDefinition.getReleaseState() != ReleaseState.ONLINE) {
putMsg(result, Status.PROCESS_DEFINE_NOT_RELEASE, processDefinition.getName());
logger.info("not release process definition id: {} , name : {}",
processDefinition.getId(), processDefinition.getName());
return result;
@ -431,13 +452,13 @@ public class ExecutorService extends BaseService{
* @param processInstanceId process instance id
* @return receivers cc list
*/
public Map<String, Object> getReceiverCc(Integer processDefineId,Integer processInstanceId) {
public Map<String, Object> getReceiverCc(Integer processDefineId, Integer processInstanceId) {
Map<String, Object> result = new HashMap<>();
logger.info("processInstanceId {}",processInstanceId);
if(processDefineId == null && processInstanceId == null){
logger.info("processInstanceId {}", processInstanceId);
if (processDefineId == null && processInstanceId == null) {
throw new RuntimeException("You must set values for parameters processDefineId or processInstanceId");
}
if(processDefineId == null && processInstanceId != null) {
if (processDefineId == null && processInstanceId != null) {
ProcessInstance processInstance = processInstanceMapper.selectById(processInstanceId);
if (processInstance == null) {
throw new RuntimeException("processInstanceId is not exists");
@ -445,24 +466,24 @@ public class ExecutorService extends BaseService{
processDefineId = processInstance.getProcessDefinitionId();
}
ProcessDefinition processDefinition = processDefinitionMapper.selectById(processDefineId);
if (processDefinition == null){
throw new RuntimeException(String.format("processDefineId %d is not exists",processDefineId));
if (processDefinition == null) {
throw new RuntimeException(String.format("processDefineId %d is not exists", processDefineId));
}
String receivers = processDefinition.getReceivers();
String receiversCc = processDefinition.getReceiversCc();
Map<String,String> dataMap = new HashMap<>();
dataMap.put(Constants.RECEIVERS,receivers);
dataMap.put(Constants.RECEIVERS_CC,receiversCc);
Map<String, String> dataMap = new HashMap<>();
dataMap.put(Constants.RECEIVERS, receivers);
dataMap.put(Constants.RECEIVERS_CC, receiversCc);
result.put(Constants.DATA_LIST, dataMap);
putMsg(result, Status.SUCCESS);
return result;
}
/**
* create command
*
* @param commandType commandType
* @param processDefineId processDefineId
* @param nodeDep nodeDep
@ -476,37 +497,36 @@ public class ExecutorService extends BaseService{
* @param processInstancePriority processInstancePriority
* @param workerGroup workerGroup
* @return command id
* @throws ParseException
*/
private int createCommand(CommandType commandType, int processDefineId,
TaskDependType nodeDep, FailureStrategy failureStrategy,
String startNodeList, String schedule, WarningType warningType,
int executorId, int warningGroupId,
RunMode runMode,Priority processInstancePriority, String workerGroup) throws ParseException {
RunMode runMode, Priority processInstancePriority, String workerGroup) throws ParseException {
/**
* instantiate command schedule instance
*/
Command command = new Command();
Map<String,String> cmdParam = new HashMap<>();
if(commandType == null){
Map<String, String> cmdParam = new HashMap<>();
if (commandType == null) {
command.setCommandType(CommandType.START_PROCESS);
}else{
} else {
command.setCommandType(commandType);
}
command.setProcessDefinitionId(processDefineId);
if(nodeDep != null){
if (nodeDep != null) {
command.setTaskDependType(nodeDep);
}
if(failureStrategy != null){
if (failureStrategy != null) {
command.setFailureStrategy(failureStrategy);
}
if(StringUtils.isNotEmpty(startNodeList)){
cmdParam.put(CMDPARAM_START_NODE_NAMES, startNodeList);
if (StringUtils.isNotEmpty(startNodeList)) {
cmdParam.put(CMD_PARAM_START_NODE_NAMES, startNodeList);
}
if(warningType != null){
if (warningType != null) {
command.setWarningType(warningType);
}
command.setCommandParam(JSONUtils.toJsonString(cmdParam));
@ -517,32 +537,32 @@ public class ExecutorService extends BaseService{
Date start = null;
Date end = null;
if(StringUtils.isNotEmpty(schedule)){
if (StringUtils.isNotEmpty(schedule)) {
String[] interval = schedule.split(",");
if(interval.length == 2){
if (interval.length == 2) {
start = DateUtils.getScheduleDate(interval[0]);
end = DateUtils.getScheduleDate(interval[1]);
}
}
// determine whether to complement
if(commandType == CommandType.COMPLEMENT_DATA){
if (commandType == CommandType.COMPLEMENT_DATA) {
runMode = (runMode == null) ? RunMode.RUN_MODE_SERIAL : runMode;
if(null != start && null != end && !start.after(end)){
if(runMode == RunMode.RUN_MODE_SERIAL){
if (null != start && null != end && !start.after(end)) {
if (runMode == RunMode.RUN_MODE_SERIAL) {
cmdParam.put(CMDPARAM_COMPLEMENT_DATA_START_DATE, DateUtils.dateToString(start));
cmdParam.put(CMDPARAM_COMPLEMENT_DATA_END_DATE, DateUtils.dateToString(end));
command.setCommandParam(JSONUtils.toJsonString(cmdParam));
return processService.createCommand(command);
}else if (runMode == RunMode.RUN_MODE_PARALLEL){
} else if (runMode == RunMode.RUN_MODE_PARALLEL) {
List<Schedule> schedules = processService.queryReleaseSchedulerListByProcessDefinitionId(processDefineId);
List<Date> listDate = new LinkedList<>();
if(!CollectionUtils.isEmpty(schedules)){
if (!CollectionUtils.isEmpty(schedules)) {
for (Schedule item : schedules) {
listDate.addAll(CronUtils.getSelfFireDateList(start, end, item.getCrontab()));
}
}
if(!CollectionUtils.isEmpty(listDate)){
if (!CollectionUtils.isEmpty(listDate)) {
// loop by schedule date
for (Date date : listDate) {
cmdParam.put(CMDPARAM_COMPLEMENT_DATA_START_DATE, DateUtils.dateToString(date));
@ -551,10 +571,10 @@ public class ExecutorService extends BaseService{
processService.createCommand(command);
}
return listDate.size();
}else{
} else {
// loop by day
int runCunt = 0;
while(!start.after(end)) {
while (!start.after(end)) {
runCunt += 1;
cmdParam.put(CMDPARAM_COMPLEMENT_DATA_START_DATE, DateUtils.dateToString(start));
cmdParam.put(CMDPARAM_COMPLEMENT_DATA_END_DATE, DateUtils.dateToString(start));
@ -565,11 +585,11 @@ public class ExecutorService extends BaseService{
return runCunt;
}
}
}else{
} else {
logger.error("there is not valid schedule date for the process definition: id:{},date:{}",
processDefineId, schedule);
}
}else{
} else {
command.setCommandParam(JSONUtils.toJsonString(cmdParam));
return processService.createCommand(command);
}
@ -579,11 +599,6 @@ public class ExecutorService extends BaseService{
/**
* check result and auth
*
* @param loginUser
* @param projectName
* @param project
* @return
*/
private Map<String, Object> checkResultAndAuth(User loginUser, String projectName, Project project) {
// check project auth

30
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ProcessDefinitionServiceImpl.java
Unescape View File

@ -17,7 +17,7 @@
package org.apache.dolphinscheduler.api.service.impl;
import static org.apache.dolphinscheduler.common.Constants.CMDPARAM_SUB_PROCESS_DEFINE_ID;
import static org.apache.dolphinscheduler.common.Constants.CMD_PARAM_SUB_PROCESS_DEFINE_ID;
import org.apache.dolphinscheduler.api.dto.ProcessMeta;
import org.apache.dolphinscheduler.api.dto.treeview.Instance;
@ -78,7 +78,6 @@ import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
@ -86,7 +85,6 @@ import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;
@ -161,13 +159,14 @@ public class ProcessDefinitionServiceImpl extends BaseService implements
* @return create result code
* @throws JsonProcessingException JsonProcessingException
*/
@Override
public Map<String, Object> createProcessDefinition(User loginUser,
String projectName,
String name,
String processDefinitionJson,
String desc,
String locations,
String connects) {
String connects) {
Map<String, Object> result = new HashMap<>();
Project project = projectMapper.queryByName(projectName);
@ -230,6 +229,7 @@ public class ProcessDefinitionServiceImpl extends BaseService implements
/**
* get resource ids
*
* @param processData process data
* @return resource ids
*/
@ -264,6 +264,7 @@ public class ProcessDefinitionServiceImpl extends BaseService implements
}
return sb.toString();
}
/**
* query process definition list
*
@ -271,6 +272,7 @@ public class ProcessDefinitionServiceImpl extends BaseService implements
* @param projectName project name
* @return definition list
*/
@Override
public Map<String, Object> queryProcessDefinitionList(User loginUser, String projectName) {
HashMap<String, Object> result = new HashMap<>(5);
@ -300,6 +302,7 @@ public class ProcessDefinitionServiceImpl extends BaseService implements
* @param userId user id
* @return process definition page
*/
@Override
public Map<String, Object> queryProcessDefinitionListPaging(User loginUser, String projectName, String searchVal, Integer pageNo, Integer pageSize, Integer userId) {
Map<String, Object> result = new HashMap<>();
@ -332,6 +335,7 @@ public class ProcessDefinitionServiceImpl extends BaseService implements
* @param processId process definition id
* @return process definition detail
*/
@Override
public Map<String, Object> queryProcessDefinitionById(User loginUser, String projectName, Integer processId) {
Map<String, Object> result = new HashMap<>();
@ -366,6 +370,7 @@ public class ProcessDefinitionServiceImpl extends BaseService implements
* @param connects connects for nodes
* @return update result code
*/
@Override
public Map<String, Object> updateProcessDefinition(User loginUser,
String projectName,
int id,
@ -455,6 +460,7 @@ public class ProcessDefinitionServiceImpl extends BaseService implements
* @param name name
* @return true if process definition name not exists, otherwise false
*/
@Override
public Map<String, Object> verifyProcessDefinitionName(User loginUser, String projectName, String name) {
Map<String, Object> result = new HashMap<>();
@ -482,6 +488,7 @@ public class ProcessDefinitionServiceImpl extends BaseService implements
* @param processDefinitionId process definition id
* @return delete result code
*/
@Override
@Transactional(rollbackFor = RuntimeException.class)
public Map<String, Object> deleteProcessDefinitionById(User loginUser, String projectName, Integer processDefinitionId) {
@ -513,9 +520,9 @@ public class ProcessDefinitionServiceImpl extends BaseService implements
return result;
}
// check process instances is already running
List<ProcessInstance> processInstances = processInstanceService.queryByProcessDefineIdAndStatus(processDefinitionId, Constants.NOT_TERMINATED_STATES);
List<ProcessInstance> processInstances = processInstanceService.queryByProcessDefineIdAndStatus(processDefinitionId, Constants.NOT_TERMINATED_STATES);
if (CollectionUtils.isNotEmpty(processInstances)) {
putMsg(result, Status.DELETE_PROCESS_DEFINITION_BY_ID_FAIL,processInstances.size());
putMsg(result, Status.DELETE_PROCESS_DEFINITION_BY_ID_FAIL, processInstances.size());
return result;
}
@ -554,6 +561,7 @@ public class ProcessDefinitionServiceImpl extends BaseService implements
* @param releaseState release state
* @return release result code
*/
@Override
@Transactional(rollbackFor = RuntimeException.class)
public Map<String, Object> releaseProcessDefinition(User loginUser, String projectName, int id, int releaseState) {
HashMap<String, Object> result = new HashMap<>();
@ -621,6 +629,7 @@ public class ProcessDefinitionServiceImpl extends BaseService implements
/**
* batch export process definition by ids
*/
@Override
public void batchExportProcessDefinitionByIds(User loginUser, String projectName, String processDefinitionIds, HttpServletResponse response) {
if (StringUtils.isEmpty(processDefinitionIds)) {
@ -699,6 +708,7 @@ public class ProcessDefinitionServiceImpl extends BaseService implements
/**
* get export process metadata string
*
* @param processDefinitionId process definition id
* @param processDefinition process definition
* @return export process metadata string
@ -790,6 +800,7 @@ public class ProcessDefinitionServiceImpl extends BaseService implements
* @param currentProjectName current project name
* @return import process
*/
@Override
@Transactional(rollbackFor = RuntimeException.class)
public Map<String, Object> importProcessDefinition(User loginUser, MultipartFile file, String currentProjectName) {
Map<String, Object> result = new HashMap<>(5);
@ -1123,6 +1134,7 @@ public class ProcessDefinitionServiceImpl extends BaseService implements
* @param processDefinitionJson process definition json
* @return check result code
*/
@Override
public Map<String, Object> checkProcessNodeList(ProcessData processData, String processDefinitionJson) {
Map<String, Object> result = new HashMap<>();
@ -1174,6 +1186,7 @@ public class ProcessDefinitionServiceImpl extends BaseService implements
* @param defineId define id
* @return task node list
*/
@Override
public Map<String, Object> getTaskNodeListByDefinitionId(Integer defineId) {
Map<String, Object> result = new HashMap<>();
@ -1210,6 +1223,7 @@ public class ProcessDefinitionServiceImpl extends BaseService implements
* @param defineIdList define id list
* @return task node list
*/
@Override
public Map<String, Object> getTaskNodeListByDefinitionIdList(String defineIdList) {
Map<String, Object> result = new HashMap<>();
@ -1247,6 +1261,7 @@ public class ProcessDefinitionServiceImpl extends BaseService implements
* @param projectId project id
* @return process definitions in the project
*/
@Override
public Map<String, Object> queryProcessDefinitionAllByProjectId(Integer projectId) {
HashMap<String, Object> result = new HashMap<>(5);
@ -1266,6 +1281,7 @@ public class ProcessDefinitionServiceImpl extends BaseService implements
* @return tree view json data
* @throws Exception exception
*/
@Override
public Map<String, Object> viewTree(Integer processId, Integer limit) throws Exception {
Map<String, Object> result = new HashMap<>();
@ -1350,7 +1366,7 @@ public class ProcessDefinitionServiceImpl extends BaseService implements
String taskJson = taskInstance.getTaskJson();
taskNode = JSONUtils.parseObject(taskJson, TaskNode.class);
subProcessId = Integer.parseInt(JSONUtils.parseObject(
taskNode.getParams()).path(CMDPARAM_SUB_PROCESS_DEFINE_ID).asText());
taskNode.getParams()).path(CMD_PARAM_SUB_PROCESS_DEFINE_ID).asText());
}
treeViewDto.getInstances().add(new Instance(taskInstance.getId(), taskInstance.getName(), taskInstance.getTaskType(), taskInstance.getState().toString()
, taskInstance.getStartTime(), taskInstance.getEndTime(), taskInstance.getHost(), DateUtils.format2Readable(endTime.getTime() - startTime.getTime()), subProcessId));

28
dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/ExecutorService2Test.java
Unescape View File

@ -14,6 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dolphinscheduler.api.service;
import static org.mockito.ArgumentMatchers.any;
@ -137,9 +138,31 @@ public class ExecutorService2Test {
Assert.assertEquals(Status.SUCCESS, result.get(Constants.STATUS));
verify(processService, times(1)).createCommand(any(Command.class));
} catch (Exception e) {
//ignore
}
}
/**
* not complement
*/
@Test
public void testComplementWithStartNodeList() throws ParseException {
try {
Mockito.when(processService.queryReleaseSchedulerListByProcessDefinitionId(processDefinitionId)).thenReturn(zeroSchedulerList());
Map<String, Object> result = executorService.execProcessInstance(loginUser, projectName,
processDefinitionId, cronTime, CommandType.START_PROCESS,
null, "n1,n2",
null, null, 0,
"", "", RunMode.RUN_MODE_SERIAL,
Priority.LOW, Constants.DEFAULT_WORKER_GROUP, 110);
Assert.assertEquals(Status.SUCCESS, result.get(Constants.STATUS));
verify(processService, times(1)).createCommand(any(Command.class));
} catch (Exception e) {
//ignore
}
}
/**
* date error
*/
@ -156,6 +179,7 @@ public class ExecutorService2Test {
Assert.assertEquals(Status.START_PROCESS_INSTANCE_ERROR, result.get(Constants.STATUS));
verify(processService, times(0)).createCommand(any(Command.class));
} catch (Exception e) {
//ignore
}
}
@ -175,6 +199,7 @@ public class ExecutorService2Test {
Assert.assertEquals(Status.SUCCESS, result.get(Constants.STATUS));
verify(processService, times(1)).createCommand(any(Command.class));
} catch (Exception e) {
//ignore
}
}
@ -194,6 +219,7 @@ public class ExecutorService2Test {
Assert.assertEquals(Status.SUCCESS, result.get(Constants.STATUS));
verify(processService, times(31)).createCommand(any(Command.class));
} catch (Exception e) {
//ignore
}
}
@ -213,10 +239,10 @@ public class ExecutorService2Test {
Assert.assertEquals(Status.SUCCESS, result.get(Constants.STATUS));
verify(processService, times(15)).createCommand(any(Command.class));
} catch (Exception e) {
//ignore
}
}
@Test
public void testNoMsterServers() throws ParseException {
Mockito.when(monitorService.getServerListFromZK(true)).thenReturn(new ArrayList<Server>());

170
dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/ProcessDefinitionServiceTest.java
Unescape View File

@ -18,6 +18,8 @@
package org.apache.dolphinscheduler.api.service;
import static org.assertj.core.api.Assertions.assertThat;
import static org.powermock.api.mockito.PowerMockito.mock;
import static org.powermock.api.mockito.PowerMockito.when;
import org.apache.dolphinscheduler.api.dto.ProcessMeta;
import org.apache.dolphinscheduler.api.enums.Status;
@ -66,6 +68,9 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import javax.servlet.ServletOutputStream;
import javax.servlet.http.HttpServletResponse;
import org.junit.Assert;
import org.junit.Test;
import org.junit.runner.RunWith;
@ -83,33 +88,6 @@ import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
@RunWith(MockitoJUnitRunner.class)
public class ProcessDefinitionServiceTest {
@InjectMocks
private ProcessDefinitionServiceImpl processDefinitionService;
@Mock
private ProcessDefinitionMapper processDefineMapper;
@Mock
private ProjectMapper projectMapper;
@Mock
private ProjectServiceImpl projectService;
@Mock
private ScheduleMapper scheduleMapper;
@Mock
private ProcessService processService;
@Mock
private ProcessInstanceService processInstanceService;
@Mock
private TaskInstanceMapper taskInstanceMapper;
@Mock
private ProcessDefinitionVersionService processDefinitionVersionService;
private static final String SHELL_JSON = "{\n"
+ " \"globalParams\": [\n"
+ " \n"
@ -150,7 +128,6 @@ public class ProcessDefinitionServiceTest {
+ " \"tenantId\": 1,\n"
+ " \"timeout\": 0\n"
+ "}";
private static final String CYCLE_SHELL_JSON = "{\n"
+ " \"globalParams\": [\n"
+ " \n"
@ -253,6 +230,24 @@ public class ProcessDefinitionServiceTest {
+ " \"tenantId\": 1,\n"
+ " \"timeout\": 0\n"
+ "}";
@InjectMocks
private ProcessDefinitionServiceImpl processDefinitionService;
@Mock
private ProcessDefinitionMapper processDefineMapper;
@Mock
private ProjectMapper projectMapper;
@Mock
private ProjectServiceImpl projectService;
@Mock
private ScheduleMapper scheduleMapper;
@Mock
private ProcessService processService;
@Mock
private ProcessInstanceService processInstanceService;
@Mock
private TaskInstanceMapper taskInstanceMapper;
@Mock
private ProcessDefinitionVersionService processDefinitionVersionService;
@Test
public void testQueryProcessDefinitionList() {
@ -751,6 +746,70 @@ public class ProcessDefinitionServiceTest {
Mockito.when(taskInstanceMapper.queryByInstanceIdAndName(processInstance.getId(), "shell-1")).thenReturn(taskInstance);
Map<String, Object> taskNotNuLLRes = processDefinitionService.viewTree(46, 10);
Assert.assertEquals(Status.SUCCESS, taskNotNuLLRes.get(Constants.STATUS));
}
@Test
public void testSubProcessViewTree() throws Exception {
ProcessDefinition processDefinition = getProcessDefinition();
processDefinition.setProcessDefinitionJson(SHELL_JSON);
List<ProcessInstance> processInstanceList = new ArrayList<>();
ProcessInstance processInstance = new ProcessInstance();
processInstance.setId(1);
processInstance.setName("test_instance");
processInstance.setState(ExecutionStatus.RUNNING_EXECUTION);
processInstance.setHost("192.168.xx.xx");
processInstance.setStartTime(new Date());
processInstance.setEndTime(new Date());
processInstanceList.add(processInstance);
TaskInstance taskInstance = new TaskInstance();
taskInstance.setStartTime(new Date());
taskInstance.setEndTime(new Date());
taskInstance.setTaskType("SUB_PROCESS");
taskInstance.setId(1);
taskInstance.setName("test_task_instance");
taskInstance.setState(ExecutionStatus.RUNNING_EXECUTION);
taskInstance.setHost("192.168.xx.xx");
taskInstance.setTaskJson("{\n"
+ " \"conditionResult\": {\n"
+ " \"failedNode\": [\n"
+ " \"\"\n"
+ " ],\n"
+ " \"successNode\": [\n"
+ " \"\"\n"
+ " ]\n"
+ " },\n"
+ " \"delayTime\": \"0\",\n"
+ " \"dependence\": {},\n"
+ " \"description\": \"\",\n"
+ " \"id\": \"1\",\n"
+ " \"maxRetryTimes\": \"0\",\n"
+ " \"name\": \"test_task_instance\",\n"
+ " \"params\": {\n"
+ " \"processDefinitionId\": \"222\",\n"
+ " \"resourceList\": []\n"
+ " },\n"
+ " \"preTasks\": [],\n"
+ " \"retryInterval\": \"1\",\n"
+ " \"runFlag\": \"NORMAL\",\n"
+ " \"taskInstancePriority\": \"MEDIUM\",\n"
+ " \"timeout\": {\n"
+ " \"enable\": false,\n"
+ " \"interval\": null,\n"
+ " \"strategy\": \"\"\n"
+ " },\n"
+ " \"type\": \"SUB_PROCESS\",\n"
+ " \"workerGroup\": \"default\"\n"
+ "}");
//task instance exist
Mockito.when(processDefineMapper.selectById(46)).thenReturn(processDefinition);
Mockito.when(processInstanceService.queryByProcessDefineId(46, 10)).thenReturn(processInstanceList);
Mockito.when(taskInstanceMapper.queryByInstanceIdAndName(processInstance.getId(), "shell-1")).thenReturn(taskInstance);
Map<String, Object> taskNotNuLLRes = processDefinitionService.viewTree(46, 10);
Assert.assertEquals(Status.SUCCESS, taskNotNuLLRes.get(Constants.STATUS));
}
@Test
@ -973,7 +1032,7 @@ public class ProcessDefinitionServiceTest {
}
@Test
public void testBatchExportProcessDefinitionByIds() {
public void testBatchExportProcessDefinitionByIds() throws IOException {
processDefinitionService.batchExportProcessDefinitionByIds(
null, null, null, null);
@ -991,6 +1050,28 @@ public class ProcessDefinitionServiceTest {
processDefinitionService.batchExportProcessDefinitionByIds(
loginUser, projectName, "1", null);
ProcessDefinition processDefinition = new ProcessDefinition();
processDefinition.setId(1);
processDefinition.setProcessDefinitionJson("{\"globalParams\":[],\"tasks\":[{\"conditionResult\":"
+ "{\"failedNode\":[\"\"],\"successNode\":[\"\"]},\"delayTime\":\"0\",\"dependence\":{}"
+ ",\"description\":\"\",\"id\":\"tasks-3011\",\"maxRetryTimes\":\"0\",\"name\":\"tsssss\""
+ ",\"params\":{\"localParams\":[],\"rawScript\":\"echo \\\"123123\\\"\",\"resourceList\":[]}"
+ ",\"preTasks\":[],\"retryInterval\":\"1\",\"runFlag\":\"NORMAL\",\"taskInstancePriority\":\"MEDIUM\""
+ ",\"timeout\":{\"enable\":false,\"interval\":null,\"strategy\":\"\"},\"type\":\"SHELL\""
+ ",\"waitStartTimeout\":{},\"workerGroup\":\"default\"}],\"tenantId\":4,\"timeout\":0}");
Map<String, Object> checkResult = new HashMap<>();
checkResult.put(Constants.STATUS, Status.SUCCESS);
Mockito.when(projectMapper.queryByName(projectName)).thenReturn(project);
Mockito.when(projectService.checkProjectAndAuth(loginUser, project, projectName)).thenReturn(checkResult);
Mockito.when(processDefineMapper.queryByDefineId(1)).thenReturn(processDefinition);
HttpServletResponse response = mock(HttpServletResponse.class);
ServletOutputStream outputStream = mock(ServletOutputStream.class);
when(response.getOutputStream()).thenReturn(outputStream);
processDefinitionService.batchExportProcessDefinitionByIds(
loginUser, projectName, "1", response);
}
@Test
@ -1182,4 +1263,35 @@ public class ProcessDefinitionServiceTest {
result.put(Constants.MSG, status.getMsg());
}
}
@Test
public void testExportProcessMetaData() {
Integer processDefinitionId = 111;
ProcessDefinition processDefinition = new ProcessDefinition();
processDefinition.setId(processDefinitionId);
processDefinition.setProcessDefinitionJson("{\"globalParams\":[],\"tasks\":[{\"conditionResult\":"
+ "{\"failedNode\":[\"\"],\"successNode\":"
+ "[\"\"]},\"delayTime\":\"0\",\"dependence\":{},"
+ "\"description\":\"\",\"id\":\"tasks-3011\",\"maxRetryTimes\":\"0\",\"name\":\"tsssss\","
+ "\"params\":{\"localParams\":[],\"rawScript\":\"echo \\\"123123\\\"\",\"resourceList\":[]},"
+ "\"preTasks\":[],\"retryInterval\":\"1\",\"runFlag\":\"NORMAL\",\"taskInstancePriority\":\"MEDIUM\","
+ "\"timeout\":{\"enable\":false,\"interval\":null,\"strategy\":\"\"},\"type\":\"SHELL\","
+ "\"waitStartTimeout\":{},\"workerGroup\":\"default\"}],\"tenantId\":4,\"timeout\":0}");
Assert.assertNotNull(processDefinitionService.exportProcessMetaData(processDefinitionId, processDefinition));
}
@Test
public void testImportProcessSchedule() {
User loginUser = new User();
loginUser.setId(1);
loginUser.setUserType(UserType.ADMIN_USER);
Integer processDefinitionId = 111;
String processDefinitionName = "testProcessDefinition";
String projectName = "project_test1";
Map<String, Object> result = new HashMap<>();
putMsg(result, Status.PROJECT_NOT_FOUNT);
ProcessMeta processMeta = new ProcessMeta();
Assert.assertEquals(0, processDefinitionService.importProcessSchedule(loginUser, projectName, processMeta, processDefinitionName, processDefinitionId));
}
}

16
dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/Constants.java
Unescape View File

@ -442,21 +442,21 @@ public final class Constants {
/**
* command parameter keys
*/
public static final String CMDPARAM_RECOVER_PROCESS_ID_STRING = "ProcessInstanceId";
public static final String CMD_PARAM_RECOVER_PROCESS_ID_STRING = "ProcessInstanceId";
public static final String CMDPARAM_RECOVERY_START_NODE_STRING = "StartNodeIdList";
public static final String CMD_PARAM_RECOVERY_START_NODE_STRING = "StartNodeIdList";
public static final String CMDPARAM_RECOVERY_WAITTING_THREAD = "WaittingThreadInstanceId";
public static final String CMD_PARAM_RECOVERY_WAITING_THREAD = "WaitingThreadInstanceId";
public static final String CMDPARAM_SUB_PROCESS = "processInstanceId";
public static final String CMD_PARAM_SUB_PROCESS = "processInstanceId";
public static final String CMDPARAM_EMPTY_SUB_PROCESS = "0";
public static final String CMD_PARAM_EMPTY_SUB_PROCESS = "0";
public static final String CMDPARAM_SUB_PROCESS_PARENT_INSTANCE_ID = "parentProcessInstanceId";
public static final String CMD_PARAM_SUB_PROCESS_PARENT_INSTANCE_ID = "parentProcessInstanceId";
public static final String CMDPARAM_SUB_PROCESS_DEFINE_ID = "processDefinitionId";
public static final String CMD_PARAM_SUB_PROCESS_DEFINE_ID = "processDefinitionId";
public static final String CMDPARAM_START_NODE_NAMES = "StartNodeNameList";
public static final String CMD_PARAM_START_NODE_NAMES = "StartNodeNameList";
/**
* complement data start date

406
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterExecThread.java
View File

File diff suppressed because it is too large Load Diff

117
dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/master/MasterExecThreadTest.java
Unescape View File

@ -14,19 +14,41 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dolphinscheduler.server.master;
import org.apache.dolphinscheduler.common.utils.*;
import org.apache.dolphinscheduler.common.enums.*;
import static org.apache.dolphinscheduler.common.Constants.CMDPARAM_COMPLEMENT_DATA_END_DATE;
import static org.apache.dolphinscheduler.common.Constants.CMDPARAM_COMPLEMENT_DATA_START_DATE;
import static org.apache.dolphinscheduler.common.Constants.CMD_PARAM_RECOVERY_START_NODE_STRING;
import static org.apache.dolphinscheduler.common.Constants.CMD_PARAM_START_NODE_NAMES;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.powermock.api.mockito.PowerMockito.mock;
import org.apache.dolphinscheduler.common.enums.CommandType;
import org.apache.dolphinscheduler.common.enums.ExecutionStatus;
import org.apache.dolphinscheduler.common.enums.Flag;
import org.apache.dolphinscheduler.common.graph.DAG;
import org.apache.dolphinscheduler.common.utils.DateUtils;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.dao.entity.ProcessDefinition;
import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
import org.apache.dolphinscheduler.dao.entity.Schedule;
import org.apache.dolphinscheduler.dao.entity.TaskInstance;
import org.apache.dolphinscheduler.server.master.config.MasterConfig;
import org.apache.dolphinscheduler.server.master.runner.MasterExecThread;
import org.apache.dolphinscheduler.service.bean.SpringApplicationContext;
import org.apache.dolphinscheduler.service.process.ProcessService;
import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.text.ParseException;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
@ -36,15 +58,6 @@ import org.powermock.api.mockito.PowerMockito;
import org.powermock.core.classloader.annotations.PrepareForTest;
import org.powermock.modules.junit4.PowerMockRunner;
import org.springframework.context.ApplicationContext;
import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.text.ParseException;
import java.util.*;
import static org.apache.dolphinscheduler.common.Constants.CMDPARAM_COMPLEMENT_DATA_END_DATE;
import static org.apache.dolphinscheduler.common.Constants.CMDPARAM_COMPLEMENT_DATA_START_DATE;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.powermock.api.mockito.PowerMockito.mock;
/**
* test for MasterExecThread
@ -66,7 +79,7 @@ public class MasterExecThreadTest {
private ApplicationContext applicationContext;
@Before
public void init() throws Exception{
public void init() throws Exception {
processService = mock(ProcessService.class);
applicationContext = mock(ApplicationContext.class);
@ -92,7 +105,7 @@ public class MasterExecThreadTest {
masterExecThread = PowerMockito.spy(new MasterExecThread(
processInstance
, processService
,null, null, config));
, null, null, config));
// prepareProcess init dag
Field dag = MasterExecThread.class.getDeclaredField("dag");
dag.setAccessible(true);
@ -106,18 +119,17 @@ public class MasterExecThreadTest {
/**
* without schedule
* @throws ParseException
*/
@Test
public void testParallelWithOutSchedule() throws ParseException {
try{
try {
Mockito.when(processService.queryReleaseSchedulerListByProcessDefinitionId(processDefinitionId)).thenReturn(zeroSchedulerList());
Method method = MasterExecThread.class.getDeclaredMethod("executeComplementProcess");
method.setAccessible(true);
method.invoke(masterExecThread);
// one create save, and 1-30 for next save, and last day 20 no save
verify(processService, times(20)).saveProcessInstance(processInstance);
}catch (Exception e){
} catch (Exception e) {
e.printStackTrace();
Assert.fail();
}
@ -125,27 +137,86 @@ public class MasterExecThreadTest {
/**
* with schedule
* @throws ParseException
*/
@Test
public void testParallelWithSchedule() throws ParseException {
try{
public void testParallelWithSchedule() {
try {
Mockito.when(processService.queryReleaseSchedulerListByProcessDefinitionId(processDefinitionId)).thenReturn(oneSchedulerList());
Method method = MasterExecThread.class.getDeclaredMethod("executeComplementProcess");
method.setAccessible(true);
method.invoke(masterExecThread);
// one create save, and 9(1 to 20 step 2) for next save, and last day 31 no save
verify(processService, times(9)).saveProcessInstance(processInstance);
}catch (Exception e){
} catch (Exception e) {
Assert.fail();
}
}
@Test
public void testParseStartNodeName() throws ParseException {
try {
Map<String, String> cmdParam = new HashMap<>();
cmdParam.put(CMD_PARAM_START_NODE_NAMES, "t1,t2,t3");
Mockito.when(processInstance.getCommandParam()).thenReturn(JSONUtils.toJsonString(cmdParam));
Class<MasterExecThread> masterExecThreadClass = MasterExecThread.class;
Method method = masterExecThreadClass.getDeclaredMethod("parseStartNodeName", String.class);
method.setAccessible(true);
List<String> nodeNames = (List<String>) method.invoke(masterExecThread, JSONUtils.toJsonString(cmdParam));
Assert.assertEquals(3, nodeNames.size());
} catch (Exception e) {
Assert.fail();
}
}
@Test
public void testRetryTaskIntervalOverTime() {
try {
TaskInstance taskInstance = new TaskInstance();
taskInstance.setId(0);
taskInstance.setMaxRetryTimes(0);
taskInstance.setRetryInterval(0);
taskInstance.setState(ExecutionStatus.FAILURE);
Class<MasterExecThread> masterExecThreadClass = MasterExecThread.class;
Method method = masterExecThreadClass.getDeclaredMethod("retryTaskIntervalOverTime", TaskInstance.class);
method.setAccessible(true);
Assert.assertTrue((Boolean) method.invoke(masterExecThread, taskInstance));
} catch (Exception e) {
Assert.fail();
}
}
@Test
public void testGetStartTaskInstanceList() {
try {
TaskInstance taskInstance1 = new TaskInstance();
taskInstance1.setId(1);
TaskInstance taskInstance2 = new TaskInstance();
taskInstance2.setId(2);
TaskInstance taskInstance3 = new TaskInstance();
taskInstance3.setId(3);
TaskInstance taskInstance4 = new TaskInstance();
taskInstance4.setId(4);
Map<String, String> cmdParam = new HashMap<>();
cmdParam.put(CMD_PARAM_RECOVERY_START_NODE_STRING, "1,2,3,4");
Mockito.when(processService.findTaskInstanceById(1)).thenReturn(taskInstance1);
Mockito.when(processService.findTaskInstanceById(2)).thenReturn(taskInstance2);
Mockito.when(processService.findTaskInstanceById(3)).thenReturn(taskInstance3);
Mockito.when(processService.findTaskInstanceById(4)).thenReturn(taskInstance4);
Class<MasterExecThread> masterExecThreadClass = MasterExecThread.class;
Method method = masterExecThreadClass.getDeclaredMethod("getStartTaskInstanceList", String.class);
method.setAccessible(true);
List<TaskInstance> taskInstances = (List<TaskInstance>) method.invoke(masterExecThread, JSONUtils.toJsonString(cmdParam));
Assert.assertEquals(4, taskInstances.size());
} catch (Exception e) {
Assert.fail();
}
}
private List<Schedule> zeroSchedulerList(){
private List<Schedule> zeroSchedulerList() {
return Collections.EMPTY_LIST;
}
private List<Schedule> oneSchedulerList(){
private List<Schedule> oneSchedulerList() {
List<Schedule> schedulerList = new LinkedList<>();
Schedule schedule = new Schedule();
schedule.setCrontab("0 0 0 1/2 * ?");

64
dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessService.java
Unescape View File

@ -19,11 +19,11 @@ package org.apache.dolphinscheduler.service.process;
import static org.apache.dolphinscheduler.common.Constants.CMDPARAM_COMPLEMENT_DATA_END_DATE;
import static org.apache.dolphinscheduler.common.Constants.CMDPARAM_COMPLEMENT_DATA_START_DATE;
import static org.apache.dolphinscheduler.common.Constants.CMDPARAM_EMPTY_SUB_PROCESS;
import static org.apache.dolphinscheduler.common.Constants.CMDPARAM_RECOVER_PROCESS_ID_STRING;
import static org.apache.dolphinscheduler.common.Constants.CMDPARAM_SUB_PROCESS;
import static org.apache.dolphinscheduler.common.Constants.CMDPARAM_SUB_PROCESS_DEFINE_ID;
import static org.apache.dolphinscheduler.common.Constants.CMDPARAM_SUB_PROCESS_PARENT_INSTANCE_ID;
import static org.apache.dolphinscheduler.common.Constants.CMD_PARAM_EMPTY_SUB_PROCESS;
import static org.apache.dolphinscheduler.common.Constants.CMD_PARAM_RECOVER_PROCESS_ID_STRING;
import static org.apache.dolphinscheduler.common.Constants.CMD_PARAM_SUB_PROCESS;
import static org.apache.dolphinscheduler.common.Constants.CMD_PARAM_SUB_PROCESS_DEFINE_ID;
import static org.apache.dolphinscheduler.common.Constants.CMD_PARAM_SUB_PROCESS_PARENT_INSTANCE_ID;
import static org.apache.dolphinscheduler.common.Constants.YYYY_MM_DD_HH_MM_SS;
import static java.util.stream.Collectors.toSet;
@ -266,14 +266,14 @@ public class ProcessService {
if (cmdTypeMap.containsKey(commandType)) {
ObjectNode cmdParamObj = JSONUtils.parseObject(command.getCommandParam());
int processInstanceId = cmdParamObj.path(CMDPARAM_RECOVER_PROCESS_ID_STRING).asInt();
int processInstanceId = cmdParamObj.path(CMD_PARAM_RECOVER_PROCESS_ID_STRING).asInt();
List<Command> commands = commandMapper.selectList(null);
// for all commands
for (Command tmpCommand : commands) {
if (cmdTypeMap.containsKey(tmpCommand.getCommandType())) {
ObjectNode tempObj = JSONUtils.parseObject(tmpCommand.getCommandParam());
if (tempObj != null && processInstanceId == tempObj.path(CMDPARAM_RECOVER_PROCESS_ID_STRING).asInt()) {
if (tempObj != null && processInstanceId == tempObj.path(CMD_PARAM_RECOVER_PROCESS_ID_STRING).asInt()) {
isNeedCreate = false;
break;
}
@ -439,7 +439,7 @@ public class ProcessService {
for (TaskNode taskNode : taskNodeList) {
String parameter = taskNode.getParams();
ObjectNode parameterJson = JSONUtils.parseObject(parameter);
if (parameterJson.get(CMDPARAM_SUB_PROCESS_DEFINE_ID) != null) {
if (parameterJson.get(CMD_PARAM_SUB_PROCESS_DEFINE_ID) != null) {
SubProcessParameters subProcessParam = JSONUtils.parseObject(parameter, SubProcessParameters.class);
ids.add(subProcessParam.getProcessDefinitionId());
recurseFindSubProcessId(subProcessParam.getProcessDefinitionId(), ids);
@ -468,7 +468,7 @@ public class ProcessService {
return;
}
Map<String, String> cmdParam = new HashMap<>();
cmdParam.put(Constants.CMDPARAM_RECOVERY_WAITTING_THREAD, String.valueOf(processInstance.getId()));
cmdParam.put(Constants.CMD_PARAM_RECOVERY_WAITING_THREAD, String.valueOf(processInstance.getId()));
// process instance quit by "waiting thread" state
if (originCommand == null) {
Command command = new Command(
@ -613,8 +613,8 @@ public class ProcessService {
private Boolean checkCmdParam(Command command, Map<String, String> cmdParam) {
if (command.getTaskDependType() == TaskDependType.TASK_ONLY || command.getTaskDependType() == TaskDependType.TASK_PRE) {
if (cmdParam == null
|| !cmdParam.containsKey(Constants.CMDPARAM_START_NODE_NAMES)
|| cmdParam.get(Constants.CMDPARAM_START_NODE_NAMES).isEmpty()) {
|| !cmdParam.containsKey(Constants.CMD_PARAM_START_NODE_NAMES)
|| cmdParam.get(Constants.CMD_PARAM_START_NODE_NAMES).isEmpty()) {
logger.error("command node depend type is {}, but start nodes is null ", command.getTaskDependType());
return false;
}
@ -647,20 +647,20 @@ public class ProcessService {
if (cmdParam != null) {
Integer processInstanceId = 0;
// recover from failure or pause tasks
if (cmdParam.containsKey(Constants.CMDPARAM_RECOVER_PROCESS_ID_STRING)) {
String processId = cmdParam.get(Constants.CMDPARAM_RECOVER_PROCESS_ID_STRING);
if (cmdParam.containsKey(Constants.CMD_PARAM_RECOVER_PROCESS_ID_STRING)) {
String processId = cmdParam.get(Constants.CMD_PARAM_RECOVER_PROCESS_ID_STRING);
processInstanceId = Integer.parseInt(processId);
if (processInstanceId == 0) {
logger.error("command parameter is error, [ ProcessInstanceId ] is 0");
return null;
}
} else if (cmdParam.containsKey(Constants.CMDPARAM_SUB_PROCESS)) {
} else if (cmdParam.containsKey(Constants.CMD_PARAM_SUB_PROCESS)) {
// sub process map
String pId = cmdParam.get(Constants.CMDPARAM_SUB_PROCESS);
String pId = cmdParam.get(Constants.CMD_PARAM_SUB_PROCESS);
processInstanceId = Integer.parseInt(pId);
} else if (cmdParam.containsKey(Constants.CMDPARAM_RECOVERY_WAITTING_THREAD)) {
} else if (cmdParam.containsKey(Constants.CMD_PARAM_RECOVERY_WAITING_THREAD)) {
// waiting thread command
String pId = cmdParam.get(Constants.CMDPARAM_RECOVERY_WAITTING_THREAD);
String pId = cmdParam.get(Constants.CMD_PARAM_RECOVERY_WAITING_THREAD);
processInstanceId = Integer.parseInt(pId);
}
if (processInstanceId == 0) {
@ -681,7 +681,7 @@ public class ProcessService {
}
}
// reset command parameter if sub process
if (cmdParam.containsKey(Constants.CMDPARAM_SUB_PROCESS)) {
if (cmdParam.containsKey(Constants.CMD_PARAM_SUB_PROCESS)) {
processInstance.setCommandParam(command.getCommandParam());
}
} else {
@ -708,14 +708,14 @@ public class ProcessService {
List<Integer> failedList = this.findTaskIdByInstanceState(processInstance.getId(), ExecutionStatus.FAILURE);
List<Integer> toleranceList = this.findTaskIdByInstanceState(processInstance.getId(), ExecutionStatus.NEED_FAULT_TOLERANCE);
List<Integer> killedList = this.findTaskIdByInstanceState(processInstance.getId(), ExecutionStatus.KILL);
cmdParam.remove(Constants.CMDPARAM_RECOVERY_START_NODE_STRING);
cmdParam.remove(Constants.CMD_PARAM_RECOVERY_START_NODE_STRING);
failedList.addAll(killedList);
failedList.addAll(toleranceList);
for (Integer taskId : failedList) {
initTaskInstance(this.findTaskInstanceById(taskId));
}
cmdParam.put(Constants.CMDPARAM_RECOVERY_START_NODE_STRING,
cmdParam.put(Constants.CMD_PARAM_RECOVERY_START_NODE_STRING,
String.join(Constants.COMMA, convertIntListToString(failedList)));
processInstance.setCommandParam(JSONUtils.toJsonString(cmdParam));
processInstance.setRunTimes(runTime + 1);
@ -726,7 +726,7 @@ public class ProcessService {
break;
case RECOVER_SUSPENDED_PROCESS:
// find pause tasks and init task's state
cmdParam.remove(Constants.CMDPARAM_RECOVERY_START_NODE_STRING);
cmdParam.remove(Constants.CMD_PARAM_RECOVERY_START_NODE_STRING);
List<Integer> suspendedNodeList = this.findTaskIdByInstanceState(processInstance.getId(), ExecutionStatus.PAUSE);
List<Integer> stopNodeList = findTaskIdByInstanceState(processInstance.getId(),
ExecutionStatus.KILL);
@ -735,7 +735,7 @@ public class ProcessService {
// initialize the pause state
initTaskInstance(this.findTaskInstanceById(taskId));
}
cmdParam.put(Constants.CMDPARAM_RECOVERY_START_NODE_STRING, String.join(",", convertIntListToString(suspendedNodeList)));
cmdParam.put(Constants.CMD_PARAM_RECOVERY_START_NODE_STRING, String.join(",", convertIntListToString(suspendedNodeList)));
processInstance.setCommandParam(JSONUtils.toJsonString(cmdParam));
processInstance.setRunTimes(runTime + 1);
break;
@ -755,8 +755,8 @@ public class ProcessService {
break;
case REPEAT_RUNNING:
// delete the recover task names from command parameter
if (cmdParam.containsKey(Constants.CMDPARAM_RECOVERY_START_NODE_STRING)) {
cmdParam.remove(Constants.CMDPARAM_RECOVERY_START_NODE_STRING);
if (cmdParam.containsKey(Constants.CMD_PARAM_RECOVERY_START_NODE_STRING)) {
cmdParam.remove(Constants.CMD_PARAM_RECOVERY_START_NODE_STRING);
processInstance.setCommandParam(JSONUtils.toJsonString(cmdParam));
}
// delete all the valid tasks when repeat running
@ -835,16 +835,16 @@ public class ProcessService {
}
Map<String, String> paramMap = JSONUtils.toMap(cmdParam);
// write sub process id into cmd param.
if (paramMap.containsKey(CMDPARAM_SUB_PROCESS)
&& CMDPARAM_EMPTY_SUB_PROCESS.equals(paramMap.get(CMDPARAM_SUB_PROCESS))) {
paramMap.remove(CMDPARAM_SUB_PROCESS);
paramMap.put(CMDPARAM_SUB_PROCESS, String.valueOf(subProcessInstance.getId()));
if (paramMap.containsKey(CMD_PARAM_SUB_PROCESS)
&& CMD_PARAM_EMPTY_SUB_PROCESS.equals(paramMap.get(CMD_PARAM_SUB_PROCESS))) {
paramMap.remove(CMD_PARAM_SUB_PROCESS);
paramMap.put(CMD_PARAM_SUB_PROCESS, String.valueOf(subProcessInstance.getId()));
subProcessInstance.setCommandParam(JSONUtils.toJsonString(paramMap));
subProcessInstance.setIsSubProcess(Flag.YES);
this.saveProcessInstance(subProcessInstance);
}
// copy parent instance user def params to sub process..
String parentInstanceId = paramMap.get(CMDPARAM_SUB_PROCESS_PARENT_INSTANCE_ID);
String parentInstanceId = paramMap.get(CMD_PARAM_SUB_PROCESS_PARENT_INSTANCE_ID);
if (StringUtils.isNotEmpty(parentInstanceId)) {
ProcessInstance parentInstance = findProcessInstanceDetailById(Integer.parseInt(parentInstanceId));
if (parentInstance != null) {
@ -1058,7 +1058,7 @@ public class ProcessService {
CommandType commandType = getSubCommandType(parentProcessInstance, childInstance);
TaskNode taskNode = JSONUtils.parseObject(task.getTaskJson(), TaskNode.class);
Map<String, String> subProcessParam = JSONUtils.toMap(taskNode.getParams());
Integer childDefineId = Integer.parseInt(subProcessParam.get(Constants.CMDPARAM_SUB_PROCESS_DEFINE_ID));
Integer childDefineId = Integer.parseInt(subProcessParam.get(Constants.CMD_PARAM_SUB_PROCESS_DEFINE_ID));
String processParam = getSubWorkFlowParam(instanceMap, parentProcessInstance);
return new Command(
@ -1443,7 +1443,7 @@ public class ProcessService {
* @return create process instance result
*/
public int createWorkProcessInstanceMap(ProcessInstanceMap processInstanceMap) {
Integer count = 0;
int count = 0;
if (processInstanceMap != null) {
return processInstanceMapMapper.insert(processInstanceMap);
}
@ -1647,7 +1647,7 @@ public class ProcessService {
//2 insert into recover command
Command cmd = new Command();
cmd.setProcessDefinitionId(processInstance.getProcessDefinitionId());
cmd.setCommandParam(String.format("{\"%s\":%d}", Constants.CMDPARAM_RECOVER_PROCESS_ID_STRING, processInstance.getId()));
cmd.setCommandParam(String.format("{\"%s\":%d}", Constants.CMD_PARAM_RECOVER_PROCESS_ID_STRING, processInstance.getId()));
cmd.setExecutorId(processInstance.getExecutorId());
cmd.setCommandType(CommandType.RECOVER_TOLERANCE_FAULT_PROCESS);
createCommand(cmd);

211
dolphinscheduler-service/src/test/java/org/apache/dolphinscheduler/service/process/ProcessServiceTest.java
Unescape View File

@ -17,30 +17,75 @@
package org.apache.dolphinscheduler.service.process;
import static org.apache.dolphinscheduler.common.Constants.CMD_PARAM_RECOVER_PROCESS_ID_STRING;
import static org.apache.dolphinscheduler.common.Constants.CMD_PARAM_SUB_PROCESS_DEFINE_ID;
import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.common.enums.CommandType;
import org.apache.dolphinscheduler.common.enums.Flag;
import org.apache.dolphinscheduler.common.enums.WarningType;
import org.apache.dolphinscheduler.common.utils.DateUtils;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.dao.entity.Command;
import org.apache.dolphinscheduler.dao.entity.ProcessDefinition;
import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
import org.apache.dolphinscheduler.dao.entity.ProcessInstanceMap;
import org.apache.dolphinscheduler.dao.entity.TaskInstance;
import org.apache.dolphinscheduler.dao.entity.User;
import org.apache.dolphinscheduler.dao.mapper.CommandMapper;
import org.apache.dolphinscheduler.dao.mapper.ErrorCommandMapper;
import org.apache.dolphinscheduler.dao.mapper.ProcessDefinitionMapper;
import org.apache.dolphinscheduler.dao.mapper.ProcessInstanceMapper;
import org.apache.dolphinscheduler.dao.mapper.TaskInstanceMapper;
import org.apache.dolphinscheduler.dao.mapper.UserMapper;
import org.apache.dolphinscheduler.service.quartz.cron.CronUtilsTest;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.junit.Assert;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.InjectMocks;
import org.mockito.Mock;
import org.mockito.Mockito;
import org.mockito.junit.MockitoJUnitRunner;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.fasterxml.jackson.databind.JsonNode;
/**
* process service test
*/
@RunWith(MockitoJUnitRunner.class)
public class ProcessServiceTest {
private static final Logger logger = LoggerFactory.getLogger(CronUtilsTest.class);
@InjectMocks
private ProcessService processService;
@Mock
private CommandMapper commandMapper;
@Mock
private ErrorCommandMapper errorCommandMapper;
@Mock
private ProcessDefinitionMapper processDefineMapper;
@Mock
private ProcessInstanceMapper processInstanceMapper;
@Mock
private UserMapper userMapper;
@Mock
TaskInstanceMapper taskInstanceMapper;
@Test
public void testCreateSubCommand() {
ProcessService processService = new ProcessService();
@ -89,7 +134,7 @@ public class ProcessServiceTest {
String endString = "2020-01-10 00:00:00";
parentInstance.setCommandType(CommandType.START_FAILURE_TASK_PROCESS);
parentInstance.setHistoryCmd("COMPLEMENT_DATA,START_FAILURE_TASK_PROCESS");
Map<String,String> complementMap = new HashMap<>();
Map<String, String> complementMap = new HashMap<>();
complementMap.put(Constants.CMDPARAM_COMPLEMENT_DATA_START_DATE, startString);
complementMap.put(Constants.CMDPARAM_COMPLEMENT_DATA_END_DATE, endString);
parentInstance.setCommandParam(JSONUtils.toJsonString(complementMap));
@ -113,4 +158,168 @@ public class ProcessServiceTest {
);
Assert.assertEquals(CommandType.START_FAILURE_TASK_PROCESS, command.getCommandType());
}
@Test
public void testVerifyIsNeedCreateCommand() {
List<Command> commands = new ArrayList<>();
Command command = new Command();
command.setCommandType(CommandType.REPEAT_RUNNING);
command.setCommandParam("{\"" + CMD_PARAM_RECOVER_PROCESS_ID_STRING + "\":\"111\"}");
commands.add(command);
Mockito.when(commandMapper.selectList(null)).thenReturn(commands);
Assert.assertFalse(processService.verifyIsNeedCreateCommand(command));
Command command1 = new Command();
command1.setCommandType(CommandType.REPEAT_RUNNING);
command1.setCommandParam("{\"" + CMD_PARAM_RECOVER_PROCESS_ID_STRING + "\":\"222\"}");
Assert.assertTrue(processService.verifyIsNeedCreateCommand(command1));
Command command2 = new Command();
command2.setCommandType(CommandType.PAUSE);
Assert.assertTrue(processService.verifyIsNeedCreateCommand(command2));
}
@Test
public void testCreateRecoveryWaitingThreadCommand() {
int id = 123;
Mockito.when(commandMapper.deleteById(id)).thenReturn(1);
ProcessInstance subProcessInstance = new ProcessInstance();
subProcessInstance.setIsSubProcess(Flag.YES);
Command originCommand = new Command();
originCommand.setId(id);
processService.createRecoveryWaitingThreadCommand(originCommand, subProcessInstance);
ProcessInstance processInstance = new ProcessInstance();
processInstance.setId(111);
processService.createRecoveryWaitingThreadCommand(null, subProcessInstance);
Command recoverCommand = new Command();
recoverCommand.setCommandType(CommandType.RECOVER_WAITTING_THREAD);
processService.createRecoveryWaitingThreadCommand(recoverCommand, subProcessInstance);
Command repeatRunningCommand = new Command();
recoverCommand.setCommandType(CommandType.REPEAT_RUNNING);
processService.createRecoveryWaitingThreadCommand(repeatRunningCommand, subProcessInstance);
ProcessInstance subProcessInstance2 = new ProcessInstance();
subProcessInstance2.setId(111);
subProcessInstance2.setIsSubProcess(Flag.NO);
processService.createRecoveryWaitingThreadCommand(repeatRunningCommand, subProcessInstance2);
}
@Test
public void testHandleCommand() {
//cannot construct process instance, return null;
String host = "127.0.0.1";
int validThreadNum = 1;
Command command = new Command();
command.setProcessDefinitionId(222);
command.setCommandType(CommandType.REPEAT_RUNNING);
command.setCommandParam("{\"" + CMD_PARAM_RECOVER_PROCESS_ID_STRING + "\":\"111\",\""
+ CMD_PARAM_SUB_PROCESS_DEFINE_ID + "\":\"222\"}");
Mockito.when(processDefineMapper.selectById(command.getProcessDefinitionId())).thenReturn(null);
Assert.assertNull(processService.handleCommand(logger, host, validThreadNum, command));
//there is not enough thread for this command
Command command1 = new Command();
command1.setProcessDefinitionId(123);
command1.setCommandParam("{\"ProcessInstanceId\":222}");
command1.setCommandType(CommandType.START_PROCESS);
ProcessDefinition processDefinition = new ProcessDefinition();
processDefinition.setId(123);
processDefinition.setName("test");
processDefinition.setVersion(1);
processDefinition.setProcessDefinitionJson("{\"globalParams\":[],\"tasks\":[{\"conditionResult\":"
+ "{\"failedNode\":[\"\"],\"successNode\":[\"\"]},\"delayTime\":\"0\",\"dependence\":{}"
+ ",\"description\":\"\",\"id\":\"tasks-3011\",\"maxRetryTimes\":\"0\",\"name\":\"tsssss\""
+ ",\"params\":{\"localParams\":[],\"rawScript\":\"echo \\\"123123\\\"\",\"resourceList\":[]}"
+ ",\"preTasks\":[],\"retryInterval\":\"1\",\"runFlag\":\"NORMAL\",\"taskInstancePriority\":\"MEDIUM\""
+ ",\"timeout\":{\"enable\":false,\"interval\":null,\"strategy\":\"\"},\"type\":\"SHELL\""
+ ",\"waitStartTimeout\":{},\"workerGroup\":\"default\"}],\"tenantId\":4,\"timeout\":0}");
ProcessInstance processInstance = new ProcessInstance();
processInstance.setId(222);
Mockito.when(processDefineMapper.selectById(command1.getProcessDefinitionId())).thenReturn(processDefinition);
Mockito.when(processInstanceMapper.queryDetailById(222)).thenReturn(processInstance);
Assert.assertNotNull(processService.handleCommand(logger, host, validThreadNum, command1));
Command command2 = new Command();
command2.setCommandParam("{\"ProcessInstanceId\":222,\"StartNodeIdList\":\"n1,n2\"}");
command2.setProcessDefinitionId(123);
command2.setCommandType(CommandType.RECOVER_SUSPENDED_PROCESS);
Assert.assertNotNull(processService.handleCommand(logger, host, validThreadNum, command2));
Command command3 = new Command();
command3.setProcessDefinitionId(123);
command3.setCommandParam("{\"WaitingThreadInstanceId\":222}");
command3.setCommandType(CommandType.START_FAILURE_TASK_PROCESS);
Assert.assertNotNull(processService.handleCommand(logger, host, validThreadNum, command3));
Command command4 = new Command();
command4.setProcessDefinitionId(123);
command4.setCommandParam("{\"WaitingThreadInstanceId\":222,\"StartNodeIdList\":\"n1,n2\"}");
command4.setCommandType(CommandType.REPEAT_RUNNING);
Assert.assertNotNull(processService.handleCommand(logger, host, validThreadNum, command4));
}
@Test
public void testGetUserById() {
User user = new User();
user.setId(123);
Mockito.when(userMapper.selectById(123)).thenReturn(user);
Assert.assertEquals(user, processService.getUserById(123));
}
@Test
public void testFormatTaskAppId() {
TaskInstance taskInstance = new TaskInstance();
taskInstance.setId(333);
taskInstance.setProcessDefinitionId(111);
taskInstance.setProcessInstanceId(222);
Mockito.when(processService.findProcessDefineById(taskInstance.getProcessDefinitionId())).thenReturn(null);
Mockito.when(processService.findProcessInstanceById(taskInstance.getProcessInstanceId())).thenReturn(null);
Assert.assertEquals("", processService.formatTaskAppId(taskInstance));
ProcessDefinition processDefinition = new ProcessDefinition();
processDefinition.setId(111);
ProcessInstance processInstance = new ProcessInstance();
processInstance.setId(222);
Mockito.when(processService.findProcessDefineById(taskInstance.getProcessDefinitionId())).thenReturn(processDefinition);
Mockito.when(processService.findProcessInstanceById(taskInstance.getProcessInstanceId())).thenReturn(processInstance);
Assert.assertEquals("111_222_333", processService.formatTaskAppId(taskInstance));
}
@Test
public void testRecurseFindSubProcessId() {
ProcessDefinition processDefinition = new ProcessDefinition();
processDefinition.setProcessDefinitionJson("{\"globalParams\":[],\"tasks\":[{\"conditionResult\":"
+ "{\"failedNode\":[\"\"],\"successNode\":[\"\"]},\"delayTime\":\"0\""
+ ",\"dependence\":{},\"description\":\"\",\"id\":\"tasks-76544\""
+ ",\"maxRetryTimes\":\"0\",\"name\":\"test\",\"params\":{\"localParams\":[],"
+ "\"rawScript\":\"echo \\\"123123\\\"\",\"resourceList\":[],\"processDefinitionId\""
+ ":\"222\"},\"preTasks\":[],\"retryInterval\":\"1\",\"runFlag\":\"NORMAL\","
+ "\"taskInstancePriority\":\"MEDIUM\",\"timeout\":{\"enable\":false,\"interval\":"
+ "null,\"strategy\":\"\"},\"type\":\"SHELL\",\"waitStartTimeout\":{},\"workerGroup\":\"default\"}],"
+ "\"tenantId\":4,\"timeout\":0}");
int parentId = 111;
List<Integer> ids = new ArrayList<>();
ProcessDefinition processDefinition2 = new ProcessDefinition();
processDefinition2.setProcessDefinitionJson("{\"globalParams\":[],\"tasks\":[{\"conditionResult\""
+ ":{\"failedNode\":[\"\"],\"successNode\":[\"\"]},\"delayTime\":\"0\",\"dependence\":{},"
+ "\"description\":\"\",\"id\":\"tasks-76544\",\"maxRetryTimes\":\"0\",\"name\":\"test\","
+ "\"params\":{\"localParams\":[],\"rawScript\":\"echo \\\"123123\\\"\",\"resourceList\":[]},"
+ "\"preTasks\":[],\"retryInterval\":\"1\",\"runFlag\":\"NORMAL\",\"taskInstancePriority\":"
+ "\"MEDIUM\",\"timeout\":{\"enable\":false,\"interval\":null,\"strategy\":\"\"},\"type\":"
+ "\"SHELL\",\"waitStartTimeout\":{},\"workerGroup\":\"default\"}],\"tenantId\":4,\"timeout\":0}");
Mockito.when(processDefineMapper.selectById(parentId)).thenReturn(processDefinition);
Mockito.when(processDefineMapper.selectById(222)).thenReturn(processDefinition2);
processService.recurseFindSubProcessId(parentId, ids);
}
}

Write Preview
Loading…
Cancel
Save