Skip to content

DolphinScheduler
DolphinScheduler

pentaLiker / DolphinScheduler
与 Fork 源项目一致

Fork自 apache / DolphinScheduler

通知 1
Star 1
Fork 0
DolphinScheduler
DolphinScheduler

体验新版 GitCode，发现更多精彩内容 >>

未验证 提交 6ffd0196 编写于 作者: O OS 提交者： GitHub
浏览文件
操作

[cherry-pick][6471-6502]cherry-pick 6471 && 6502 to 2.0-prepare (#6508)

上级 9942d73d
隐藏空白更改
内联 并排
Showing with 235 addition and 169 deletion
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ExecutorServiceImpl.java
浏览文件 @ 6ffd0196
......@@ -283,13 +283,13 @@ public class ExecutorServiceImpl extends BaseServiceImpl implements ExecutorServ
switch (executeType) {
case REPEAT_RUNNING:
result = insertCommand(loginUser, processInstanceId, processDefinition.getCode(), CommandType.REPEAT_RUNNING, startParams);
result = insertCommand(loginUser, processInstanceId, processDefinition.getCode(), processDefinition.getVersion(), CommandType.REPEAT_RUNNING, startParams);
break;
case RECOVER_SUSPENDED_PROCESS:
result = insertCommand(loginUser, processInstanceId, processDefinition.getCode(), CommandType.RECOVER_SUSPENDED_PROCESS, startParams);
result = insertCommand(loginUser, processInstanceId, processDefinition.getCode(), processDefinition.getVersion(), CommandType.RECOVER_SUSPENDED_PROCESS, startParams);
break;
case START_FAILURE_TASK_PROCESS:
result = insertCommand(loginUser, processInstanceId, processDefinition.getCode(), CommandType.START_FAILURE_TASK_PROCESS, startParams);
result = insertCommand(loginUser, processInstanceId, processDefinition.getCode(), processDefinition.getVersion(), CommandType.START_FAILURE_TASK_PROCESS, startParams);
break;
case STOP:
if (processInstance.getState() == ExecutionStatus.READY_STOP) {
......@@ -409,10 +409,11 @@ public class ExecutorServiceImpl extends BaseServiceImpl implements ExecutorServ
* @param loginUser login user
* @param instanceId instance id
* @param processDefinitionCode process definition code
* @param version
* @param commandType command type
* @return insert result code
*/
private Map<String, Object> insertCommand(User loginUser, Integer instanceId, long processDefinitionCode, CommandType commandType, String startParams) {
private Map<String, Object> insertCommand(User loginUser, Integer instanceId, long processDefinitionCode, int processVersion, CommandType commandType, String startParams) {
Map<String, Object> result = new HashMap<>();
//To add startParams only when repeat running is needed
......@@ -427,6 +428,8 @@ public class ExecutorServiceImpl extends BaseServiceImpl implements ExecutorServ
command.setProcessDefinitionCode(processDefinitionCode);
command.setCommandParam(JSONUtils.toJsonString(cmdParam));
command.setExecutorId(loginUser.getId());
command.setProcessDefinitionVersion(processVersion);
command.setProcessInstanceId(instanceId);
if (!processService.verifyIsNeedCreateCommand(command)) {
putMsg(result, Status.PROCESS_INSTANCE_EXECUTING_COMMAND, processDefinitionCode);
......@@ -545,6 +548,11 @@ public class ExecutorServiceImpl extends BaseServiceImpl implements ExecutorServ
command.setWorkerGroup(workerGroup);
command.setEnvironmentCode(environmentCode);
command.setDryRun(dryRun);
ProcessDefinition processDefinition = processService.findProcessDefinitionByCode(processDefineCode);
if (processDefinition != null) {
command.setProcessDefinitionVersion(processDefinition.getVersion());
}
command.setProcessInstanceId(0);
Date start = null;
Date end = null;
......
dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/Command.java
浏览文件 @ 6ffd0196
......@@ -132,6 +132,12 @@ public class Command {
@TableField("dry_run")
private int dryRun;
@TableField("process_instance_id")
private int processInstanceId;
@TableField("process_definition_version")
private int processDefinitionVersion;
public Command() {
this.taskDependType = TaskDependType.TASK_POST;
this.failureStrategy = FailureStrategy.CONTINUE;
......@@ -152,7 +158,10 @@ public class Command {
String workerGroup,
Long environmentCode,
Priority processInstancePriority,
int dryRun) {
int dryRun,
int processInstanceId,
int processDefinitionVersion
) {
this.commandType = commandType;
this.executorId = executorId;
this.processDefinitionCode = processDefinitionCode;
......@@ -168,6 +177,8 @@ public class Command {
this.environmentCode = environmentCode;
this.processInstancePriority = processInstancePriority;
this.dryRun = dryRun;
this.processInstanceId = processInstanceId;
this.processDefinitionVersion = processDefinitionVersion;
}
public TaskDependType getTaskDependType() {
......@@ -298,6 +309,22 @@ public class Command {
this.dryRun = dryRun;
}
public int getProcessInstanceId() {
return processInstanceId;
}
public void setProcessInstanceId(int processInstanceId) {
this.processInstanceId = processInstanceId;
}
public int getProcessDefinitionVersion() {
return processDefinitionVersion;
}
public void setProcessDefinitionVersion(int processDefinitionVersion) {
this.processDefinitionVersion = processDefinitionVersion;
}
@Override
public boolean equals(Object o) {
if (this == o) {
......@@ -353,8 +380,13 @@ public class Command {
if (processInstancePriority != command.processInstancePriority) {
return false;
}
if (processInstanceId != command.processInstanceId) {
return false;
}
if (processDefinitionVersion != command.getProcessDefinitionVersion()) {
return false;
}
return !(updateTime != null ? !updateTime.equals(command.updateTime) : command.updateTime != null);
}
@Override
......@@ -375,6 +407,8 @@ public class Command {
result = 31 * result + (workerGroup != null ? workerGroup.hashCode() : 0);
result = 31 * result + (environmentCode != null ? environmentCode.hashCode() : 0);
result = 31 * result + dryRun;
result = 31 * result + processInstanceId;
result = 31 * result + processDefinitionVersion;
return result;
}
......@@ -397,7 +431,10 @@ public class Command {
+ ", workerGroup='" + workerGroup + '\''
+ ", environmentCode='" + environmentCode + '\''
+ ", dryRun='" + dryRun + '\''
+ ", processInstanceId='" + processInstanceId + '\''
+ ", processDefinitionVersion='" + processDefinitionVersion + '\''
+ '}';
}
}
dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/CommandMapper.java
浏览文件 @ 6ffd0196
......@@ -31,13 +31,6 @@ import java.util.List;
*/
public interface CommandMapper extends BaseMapper<Command> {
/**
* get one command
* @return command
*/
Command getOneToRun();
/**
* count command state
* @param userId userId
......
dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/CommandMapper.xml
浏览文件 @ 6ffd0196
......@@ -18,16 +18,6 @@
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd" >
<mapper namespace="org.apache.dolphinscheduler.dao.mapper.CommandMapper">
<select id="getOneToRun" resultType="org.apache.dolphinscheduler.dao.entity.Command">
select cmd.id, cmd.command_type, cmd.process_definition_code, cmd.command_param, cmd.task_depend_type, cmd.failure_strategy,
cmd.warning_type, cmd.warning_group_id, cmd.schedule_time, cmd.start_time, cmd.executor_id, cmd.update_time,
cmd.process_instance_priority, cmd.worker_group, cmd.environment_code, cmd.dry_run
from t_ds_command cmd
join t_ds_process_definition definition on cmd.process_definition_code = definition.code
where definition.release_state = 1 AND definition.flag = 1
order by cmd.update_time asc
limit 1
</select>
<select id="countCommandState" resultType="org.apache.dolphinscheduler.dao.entity.CommandCount">
select cmd.command_type as command_type, count(1) as count
from t_ds_command cmd, t_ds_process_definition process
......
dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/CommandMapperTest.java
浏览文件 @ 6ffd0196
......@@ -147,7 +147,7 @@ public class CommandMapperTest {
createCommand(CommandType.START_PROCESS, processDefinition.getCode());
Command actualCommand = commandMapper.getOneToRun();
List<Command> actualCommand = commandMapper.queryCommandPage(1,0);
assertNotNull(actualCommand);
}
......@@ -259,6 +259,8 @@ public class CommandMapperTest {
command.setStartTime(DateUtils.stringToDate("2019-12-29 10:10:00"));
command.setUpdateTime(DateUtils.stringToDate("2019-12-29 10:10:00"));
command.setWorkerGroup(Constants.DEFAULT_WORKER_GROUP);
command.setProcessInstanceId(0);
command.setProcessDefinitionVersion(0);
commandMapper.insert(command);
return command;
......
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/config/MasterConfig.java
浏览文件 @ 6ffd0196
......@@ -60,6 +60,9 @@ public class MasterConfig {
@Value("${master.reserved.memory:0.3}")
private double masterReservedMemory;
@Value("${master.cache.process.definition:true}")
private boolean masterCacheProcessDefinition;
public int getListenPort() {
return listenPort;
}
......@@ -150,4 +153,13 @@ public class MasterConfig {
public void setStateWheelInterval(int stateWheelInterval) {
this.stateWheelInterval = stateWheelInterval;
}
public boolean getMasterCacheProcessDefinition() {
return masterCacheProcessDefinition;
}
public void setMasterCacheProcessDefinition(boolean masterCacheProcessDefinition) {
this.masterCacheProcessDefinition = masterCacheProcessDefinition;
}
}
\ No newline at end of file
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterSchedulerService.java
浏览文件 @ 6ffd0196
......@@ -23,6 +23,7 @@ import org.apache.dolphinscheduler.common.thread.ThreadUtils;
import org.apache.dolphinscheduler.common.utils.NetUtils;
import org.apache.dolphinscheduler.common.utils.OSUtils;
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.TaskInstance;
import org.apache.dolphinscheduler.remote.NettyRemotingClient;
......@@ -34,6 +35,7 @@ import org.apache.dolphinscheduler.server.master.registry.ServerNodeManager;
import org.apache.dolphinscheduler.service.alert.ProcessAlertManager;
import org.apache.dolphinscheduler.service.process.ProcessService;
import java.util.HashMap;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ThreadPoolExecutor;
......@@ -92,11 +94,26 @@ public class MasterSchedulerService extends Thread {
*/
private ThreadPoolExecutor masterExecService;
/**
* process instance execution list
*/
private ConcurrentHashMap<Integer, WorkflowExecuteThread> processInstanceExecMaps;
/**
* process timeout check list
*/
ConcurrentHashMap<Integer, ProcessInstance> processTimeoutCheckList = new ConcurrentHashMap<>();
/**
* task time out checkout list
*/
ConcurrentHashMap<Integer, TaskInstance> taskTimeoutCheckList = new ConcurrentHashMap<>();
/**
* key:code-version
* value: processDefinition
*/
HashMap<String, ProcessDefinition> processDefinitionCacheMaps = new HashMap<>();
private StateWheelExecuteThread stateWheelExecuteThread;
/**
......@@ -112,7 +129,6 @@ public class MasterSchedulerService extends Thread {
taskTimeoutCheckList,
this.processInstanceExecMaps,
masterConfig.getStateWheelInterval() * Constants.SLEEP_TIME_MILLIS);
}
@Override
......@@ -165,7 +181,6 @@ public class MasterSchedulerService extends Thread {
*/
private void scheduleProcess() throws Exception {
int activeCount = masterExecService.getActiveCount();
// make sure to scan and delete command table in one transaction
Command command = findOneCommand();
if (command != null) {
......@@ -173,7 +188,12 @@ public class MasterSchedulerService extends Thread {
try {
ProcessInstance processInstance = processService.handleCommand(logger,
getLocalAddress(),
this.masterConfig.getMasterExecThreads() - activeCount, command);
command,
processDefinitionCacheMaps);
if (!masterConfig.getMasterCacheProcessDefinition()
&& processDefinitionCacheMaps.size() > 0) {
processDefinitionCacheMaps.clear();
}
if (processInstance != null) {
WorkflowExecuteThread workflowExecuteThread = new WorkflowExecuteThread(
processInstance
......
dolphinscheduler-server/src/main/resources/master.properties
浏览文件 @ 6ffd0196
......@@ -39,6 +39,9 @@
# master commit task interval, the unit is millisecond
#master.task.commit.interval=1000
# master cache process definition, default: true
#master.cache.process.definition=true
# master max cpuload avg, only higher than the system cpu load average, master server can schedule. default value -1: the number of cpu cores * 2
#master.max.cpuload.avg=-1
......
dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessService.java
浏览文件 @ 6ffd0196
......@@ -208,13 +208,12 @@ public class ProcessService {
*
* @param logger logger
* @param host host
* @param validThreadNum validThreadNum
* @param command found command
* @param processDefinitionCacheMaps
* @return process instance
*/
@Transactional(rollbackFor = Exception.class)
public ProcessInstance handleCommand(Logger logger, String host, int validThreadNum, Command command) {
ProcessInstance processInstance = constructProcessInstance(command, host);
public ProcessInstance handleCommand(Logger logger, String host, Command command, HashMap<String, ProcessDefinition> processDefinitionCacheMaps) {
ProcessInstance processInstance = constructProcessInstance(command, host, processDefinitionCacheMaps);
// cannot construct process instance, return null
if (processInstance == null) {
logger.error("scan command, command parameter is error: {}", command);
......@@ -235,7 +234,6 @@ public class ProcessService {
* @param command command
* @param message message
*/
@Transactional(rollbackFor = Exception.class)
public void moveToErrorCommand(Command command, String message) {
ErrorCommand errorCommand = new ErrorCommand(command, message);
this.errorCommandMapper.insert(errorCommand);
......@@ -286,15 +284,6 @@ public class ProcessService {
return result;
}
/**
* find one command from queue list
*
* @return command
*/
public Command findOneCommand() {
return commandMapper.getOneToRun();
}
/**
* get command page
*
......@@ -547,7 +536,9 @@ public class ProcessService {
processInstance.getWorkerGroup(),
processInstance.getEnvironmentCode(),
processInstance.getProcessInstancePriority(),
processInstance.getDryRun()
processInstance.getDryRun(),
processInstance.getId(),
processInstance.getProcessDefinitionVersion()
);
saveCommand(command);
return;
......@@ -744,90 +735,74 @@ public class ProcessService {
/**
* construct process instance according to one command.
*
* @param command command
* @param host host
* @param command command
* @param host host
* @param processDefinitionCacheMaps
* @return process instance
*/
private ProcessInstance constructProcessInstance(Command command, String host) {
private ProcessInstance constructProcessInstance(Command command, String host, HashMap<String, ProcessDefinition> processDefinitionCacheMaps) {
ProcessInstance processInstance;
ProcessDefinition processDefinition;
CommandType commandType = command.getCommandType();
Map<String, String> cmdParam = JSONUtils.toMap(command.getCommandParam());
ProcessDefinition processDefinition = getProcessDefinitionByCommand(command.getProcessDefinitionCode(), cmdParam);
String key = String.format("%d-%d", command.getProcessDefinitionCode(), command.getProcessDefinitionVersion());
if (processDefinitionCacheMaps.containsKey(key)) {
processDefinition = processDefinitionCacheMaps.get(key);
} else {
processDefinition = this.findProcessDefinition(command.getProcessDefinitionCode(), command.getProcessDefinitionVersion());
if (processDefinition != null) {
processDefinitionCacheMaps.put(key, processDefinition);
}
}
if (processDefinition == null) {
logger.error("cannot find the work process define! define code : {}", command.getProcessDefinitionCode());
return null;
}
Map<String, String> cmdParam = JSONUtils.toMap(command.getCommandParam());
int processInstanceId = command.getProcessInstanceId();
if (processInstanceId == 0) {
processInstance = generateNewProcessInstance(processDefinition, command, cmdParam);
} else {
processInstance = this.findProcessInstanceDetailById(processInstanceId);
if (processInstance == null) {
return processInstance;
}
}
if (cmdParam != null) {
int processInstanceId = 0;
// recover from failure or pause tasks
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.CMD_PARAM_SUB_PROCESS)) {
// sub process map
String pId = cmdParam.get(Constants.CMD_PARAM_SUB_PROCESS);
processInstanceId = Integer.parseInt(pId);
} else if (cmdParam.containsKey(Constants.CMD_PARAM_RECOVERY_WAITING_THREAD)) {
// waiting thread command
String pId = cmdParam.get(Constants.CMD_PARAM_RECOVERY_WAITING_THREAD);
processInstanceId = Integer.parseInt(pId);
CommandType commandTypeIfComplement = getCommandTypeIfComplement(processInstance, command);
// reset global params while repeat running is needed by cmdParam
if (commandTypeIfComplement == CommandType.REPEAT_RUNNING) {
setGlobalParamIfCommanded(processDefinition, cmdParam);
}
if (processInstanceId == 0) {
processInstance = generateNewProcessInstance(processDefinition, command, cmdParam);
} else {
processInstance = this.findProcessInstanceDetailById(processInstanceId);
if (processInstance == null) {
return processInstance;
}
CommandType commandTypeIfComplement = getCommandTypeIfComplement(processInstance, command);
// reset global params while repeat running is needed by cmdParam
if (commandTypeIfComplement == CommandType.REPEAT_RUNNING) {
setGlobalParamIfCommanded(processDefinition, cmdParam);
}
// Recalculate global parameters after rerun.
processInstance.setGlobalParams(ParameterUtils.curingGlobalParams(
// Recalculate global parameters after rerun.
processInstance.setGlobalParams(ParameterUtils.curingGlobalParams(
processDefinition.getGlobalParamMap(),
processDefinition.getGlobalParamList(),
commandTypeIfComplement,
processInstance.getScheduleTime()));
processInstance.setProcessDefinition(processDefinition);
}
//reset command parameter
if (processInstance.getCommandParam() != null) {
Map<String, String> processCmdParam = JSONUtils.toMap(processInstance.getCommandParam());
for (Map.Entry<String, String> entry : processCmdParam.entrySet()) {
if (!cmdParam.containsKey(entry.getKey())) {
cmdParam.put(entry.getKey(), entry.getValue());
}
processInstance.setProcessDefinition(processDefinition);
}
//reset command parameter
if (processInstance.getCommandParam() != null) {
Map<String, String> processCmdParam = JSONUtils.toMap(processInstance.getCommandParam());
for (Map.Entry<String, String> entry : processCmdParam.entrySet()) {
if (!cmdParam.containsKey(entry.getKey())) {
cmdParam.put(entry.getKey(), entry.getValue());
}
}
// reset command parameter if sub process
if (cmdParam.containsKey(Constants.CMD_PARAM_SUB_PROCESS)) {
processInstance.setCommandParam(command.getCommandParam());
}
} else {
// generate one new process instance
processInstance = generateNewProcessInstance(processDefinition, command, cmdParam);
}
// reset command parameter if sub process
if (cmdParam != null && cmdParam.containsKey(Constants.CMD_PARAM_SUB_PROCESS)) {
processInstance.setCommandParam(command.getCommandParam());
}
if (Boolean.FALSE.equals(checkCmdParam(command, cmdParam))) {
logger.error("command parameter check failed!");
return null;
}
if (command.getScheduleTime() != null) {
processInstance.setScheduleTime(command.getScheduleTime());
}
processInstance.setHost(host);
ExecutionStatus runStatus = ExecutionStatus.RUNNING_EXECUTION;
int runTime = processInstance.getRunTimes();
switch (commandType) {
......@@ -846,7 +821,7 @@ public class ProcessService {
initTaskInstance(this.findTaskInstanceById(taskId));
}
cmdParam.put(Constants.CMD_PARAM_RECOVERY_START_NODE_STRING,
String.join(Constants.COMMA, convertIntListToString(failedList)));
String.join(Constants.COMMA, convertIntListToString(failedList)));
processInstance.setCommandParam(JSONUtils.toJsonString(cmdParam));
processInstance.setRunTimes(runTime + 1);
break;
......@@ -859,7 +834,7 @@ public class ProcessService {
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);
ExecutionStatus.KILL);
suspendedNodeList.addAll(stopNodeList);
for (Integer taskId : suspendedNodeList) {
// initialize the pause state
......@@ -1274,7 +1249,7 @@ public class ProcessService {
}
}
String processParam = getSubWorkFlowParam(instanceMap, parentProcessInstance, fatherParams);
int subProcessInstanceId = childInstance == null ? 0 : childInstance.getId();
return new Command(
commandType,
TaskDependType.TASK_POST,
......@@ -1288,7 +1263,9 @@ public class ProcessService {
task.getWorkerGroup(),
task.getEnvironmentCode(),
parentProcessInstance.getProcessInstancePriority(),
parentProcessInstance.getDryRun()
parentProcessInstance.getDryRun(),
subProcessInstanceId,
parentProcessInstance.getProcessDefinitionVersion()
);
}
......@@ -1886,6 +1863,8 @@ public class ProcessService {
//2 insert into recover command
Command cmd = new Command();
cmd.setProcessDefinitionCode(processDefinition.getCode());
cmd.setProcessDefinitionVersion(processDefinition.getVersion());
cmd.setProcessInstanceId(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);
......
dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/quartz/ProcessScheduleJob.java
浏览文件 @ 6ffd0196
......@@ -101,6 +101,7 @@ public class ProcessScheduleJob implements Job {
command.setWorkerGroup(workerGroup);
command.setWarningType(schedule.getWarningType());
command.setProcessInstancePriority(schedule.getProcessInstancePriority());
command.setProcessDefinitionVersion(processDefinition.getVersion());
getProcessService().createCommand(command);
}
......
dolphinscheduler-service/src/test/java/org/apache/dolphinscheduler/service/process/ProcessServiceTest.java
浏览文件 @ 6ffd0196
......@@ -82,7 +82,6 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.fasterxml.jackson.databind.JsonNode;
import com.google.common.collect.Lists;
/**
* process service test
......@@ -119,6 +118,8 @@ public class ProcessServiceTest {
@Mock
private ResourceMapper resourceMapper;
private HashMap<String, ProcessDefinition> processDefinitionCacheMaps = new HashMap<>();
@Test
public void testCreateSubCommand() {
ProcessInstance parentInstance = new ProcessInstance();
......@@ -240,56 +241,65 @@ public class ProcessServiceTest {
//cannot construct process instance, return null;
String host = "127.0.0.1";
int validThreadNum = 1;
Command command = new Command();
command.setProcessDefinitionCode(222);
command.setCommandType(CommandType.REPEAT_RUNNING);
command.setCommandParam("{\"" + CMD_PARAM_RECOVER_PROCESS_ID_STRING + "\":\"111\",\""
+ CMD_PARAM_SUB_PROCESS_DEFINE_ID + "\":\"222\"}");
Assert.assertNull(processService.handleCommand(logger, host, validThreadNum, command));
Assert.assertNull(processService.handleCommand(logger, host, command, processDefinitionCacheMaps));
int definitionVersion = 1;
long definitionCode = 123;
int processInstanceId = 222;
//there is not enough thread for this command
Command command1 = new Command();
command1.setProcessDefinitionCode(123);
command1.setProcessDefinitionCode(definitionCode);
command1.setProcessDefinitionVersion(definitionVersion);
command1.setCommandParam("{\"ProcessInstanceId\":222}");
command1.setCommandType(CommandType.START_PROCESS);
ProcessDefinition processDefinition = new ProcessDefinition();
processDefinition.setId(123);
processDefinition.setName("test");
processDefinition.setVersion(1);
processDefinition.setCode(11L);
processDefinition.setVersion(definitionVersion);
processDefinition.setCode(definitionCode);
processDefinition.setGlobalParams("[{\"prop\":\"startParam1\",\"direct\":\"IN\",\"type\":\"VARCHAR\",\"value\":\"\"}]");
ProcessInstance processInstance = new ProcessInstance();
processInstance.setId(222);
processInstance.setProcessDefinitionCode(11L);
processInstance.setProcessDefinitionVersion(1);
Mockito.when(processDefineMapper.queryByCode(command1.getProcessDefinitionCode())).thenReturn(processDefinition);
processInstance.setId(processInstanceId);
processInstance.setProcessDefinitionCode(definitionCode);
processInstance.setProcessDefinitionVersion(definitionVersion);
Mockito.when(processDefineLogMapper.queryByDefinitionCodeAndVersion(processInstance.getProcessDefinitionCode(),
processInstance.getProcessDefinitionVersion())).thenReturn(new ProcessDefinitionLog(processDefinition));
Mockito.when(processInstanceMapper.queryDetailById(222)).thenReturn(processInstance);
Assert.assertNotNull(processService.handleCommand(logger, host, validThreadNum, command1));
Assert.assertNotNull(processService.handleCommand(logger, host, command1, processDefinitionCacheMaps));
Command command2 = new Command();
command2.setCommandParam("{\"ProcessInstanceId\":222,\"StartNodeIdList\":\"n1,n2\"}");
command2.setProcessDefinitionCode(123);
command2.setProcessDefinitionCode(definitionCode);
command2.setProcessDefinitionVersion(definitionVersion);
command2.setCommandType(CommandType.RECOVER_SUSPENDED_PROCESS);
command2.setProcessInstanceId(processInstanceId);
Assert.assertNotNull(processService.handleCommand(logger, host, validThreadNum, command2));
Assert.assertNotNull(processService.handleCommand(logger, host, command2, processDefinitionCacheMaps));
Command command3 = new Command();
command3.setProcessDefinitionCode(123);
command3.setProcessDefinitionCode(definitionCode);
command3.setProcessDefinitionVersion(definitionVersion);
command3.setProcessInstanceId(processInstanceId);
command3.setCommandParam("{\"WaitingThreadInstanceId\":222}");
command3.setCommandType(CommandType.START_FAILURE_TASK_PROCESS);
Assert.assertNotNull(processService.handleCommand(logger, host, validThreadNum, command3));
Assert.assertNotNull(processService.handleCommand(logger, host, command3, processDefinitionCacheMaps));
Command command4 = new Command();
command4.setProcessDefinitionCode(123);
command4.setProcessDefinitionCode(definitionCode);
command4.setProcessDefinitionVersion(definitionVersion);
command4.setCommandParam("{\"WaitingThreadInstanceId\":222,\"StartNodeIdList\":\"n1,n2\"}");
command4.setCommandType(CommandType.REPEAT_RUNNING);
Assert.assertNotNull(processService.handleCommand(logger, host, validThreadNum, command4));
command4.setProcessInstanceId(processInstanceId);
Assert.assertNotNull(processService.handleCommand(logger, host, command4, processDefinitionCacheMaps));
Command command5 = new Command();
command5.setProcessDefinitionCode(123);
command5.setProcessDefinitionCode(definitionCode);
command5.setProcessDefinitionVersion(definitionVersion);
HashMap<String, String> startParams = new HashMap<>();
startParams.put("startParam1", "testStartParam1");
HashMap<String, String> commandParams = new HashMap<>();
......@@ -297,7 +307,7 @@ public class ProcessServiceTest {
command5.setCommandParam(JSONUtils.toJsonString(commandParams));
command5.setCommandType(CommandType.START_PROCESS);
command5.setDryRun(Constants.DRY_RUN_FLAG_NO);
ProcessInstance processInstance1 = processService.handleCommand(logger, host, validThreadNum, command5);
ProcessInstance processInstance1 = processService.handleCommand(logger, host, command5, processDefinitionCacheMaps);
Assert.assertTrue(processInstance1.getGlobalParams().contains("\"testStartParam1\""));
}
......
sql/dolphinscheduler_h2.sql
浏览文件 @ 6ffd0196
......@@ -328,6 +328,8 @@ CREATE TABLE t_ds_command
worker_group varchar(64),
environment_code bigint(20) DEFAULT '-1',
dry_run int NULL DEFAULT 0,
process_instance_id int(11) DEFAULT 0,
process_definition_version int(11) DEFAULT 0,
PRIMARY KEY (id),
KEY priority_id_index (process_instance_priority, id)
);
......@@ -381,6 +383,8 @@ CREATE TABLE t_ds_error_command
environment_code bigint(20) DEFAULT '-1',
message text,
dry_run int NULL DEFAULT 0,
process_instance_id int(11) DEFAULT 0,
process_definition_version int(11) DEFAULT 0,
PRIMARY KEY (id)
);
......
sql/dolphinscheduler_mysql.sql
浏览文件 @ 6ffd0196
......@@ -317,22 +317,24 @@ CREATE TABLE `t_ds_alertgroup`(
-- ----------------------------
DROP TABLE IF EXISTS `t_ds_command`;
CREATE TABLE `t_ds_command` (
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'key',
`command_type` tinyint(4) DEFAULT NULL COMMENT 'Command type: 0 start workflow, 1 start execution from current node, 2 resume fault-tolerant workflow, 3 resume pause process, 4 start execution from failed node, 5 complement, 6 schedule, 7 rerun, 8 pause, 9 stop, 10 resume waiting thread',
`process_definition_code` bigint(20) DEFAULT NULL COMMENT 'process definition code',
`command_param` text COMMENT 'json command parameters',
`task_depend_type` tinyint(4) DEFAULT NULL COMMENT 'Node dependency type: 0 current node, 1 forward, 2 backward',
`failure_strategy` tinyint(4) DEFAULT '0' COMMENT 'Failed policy: 0 end, 1 continue',
`warning_type` tinyint(4) DEFAULT '0' COMMENT 'Alarm type: 0 is not sent, 1 process is sent successfully, 2 process is sent failed, 3 process is sent successfully and all failures are sent',
`warning_group_id` int(11) DEFAULT NULL COMMENT 'warning group',
`schedule_time` datetime DEFAULT NULL COMMENT 'schedule time',
`start_time` datetime DEFAULT NULL COMMENT 'start time',
`executor_id` int(11) DEFAULT NULL COMMENT 'executor id',
`update_time` datetime DEFAULT NULL COMMENT 'update time',
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'key',
`command_type` tinyint(4) DEFAULT NULL COMMENT 'Command type: 0 start workflow, 1 start execution from current node, 2 resume fault-tolerant workflow, 3 resume pause process, 4 start execution from failed node, 5 complement, 6 schedule, 7 rerun, 8 pause, 9 stop, 10 resume waiting thread',
`process_definition_code` bigint(20) DEFAULT NULL COMMENT 'process definition code',
`command_param` text COMMENT 'json command parameters',
`task_depend_type` tinyint(4) DEFAULT NULL COMMENT 'Node dependency type: 0 current node, 1 forward, 2 backward',
`failure_strategy` tinyint(4) DEFAULT '0' COMMENT 'Failed policy: 0 end, 1 continue',
`warning_type` tinyint(4) DEFAULT '0' COMMENT 'Alarm type: 0 is not sent, 1 process is sent successfully, 2 process is sent failed, 3 process is sent successfully and all failures are sent',
`warning_group_id` int(11) DEFAULT NULL COMMENT 'warning group',
`schedule_time` datetime DEFAULT NULL COMMENT 'schedule time',
`start_time` datetime DEFAULT NULL COMMENT 'start time',
`executor_id` int(11) DEFAULT NULL COMMENT 'executor id',
`update_time` datetime DEFAULT NULL COMMENT 'update time',
`process_instance_priority` int(11) DEFAULT NULL COMMENT 'process instance priority: 0 Highest,1 High,2 Medium,3 Low,4 Lowest',
`worker_group` varchar(64) COMMENT 'worker group',
`environment_code` bigint(20) DEFAULT '-1' COMMENT 'environment code',
`dry_run` int NULL DEFAULT 0 COMMENT 'dry run flag：0 normal, 1 dry run',
`worker_group` varchar(64) COMMENT 'worker group',
`environment_code` bigint(20) DEFAULT '-1' COMMENT 'environment code',
`dry_run` int NULL DEFAULT 0 COMMENT 'dry run flag：0 normal, 1 dry run',
`process_instance_id` int(11) DEFAULT 0 COMMENT 'process instance id',
`process_definition_version` int(11) DEFAULT 0 COMMENT 'process definition version',
PRIMARY KEY (`id`),
KEY `priority_id_index` (`process_instance_priority`,`id`) USING BTREE
) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8;
......@@ -384,6 +386,8 @@ CREATE TABLE `t_ds_error_command` (
`environment_code` bigint(20) DEFAULT '-1' COMMENT 'environment code',
`message` text COMMENT 'message',
`dry_run` int NULL DEFAULT NULL COMMENT 'dry run flag: 0 normal, 1 dry run',
`process_instance_id` int(11) DEFAULT 0 COMMENT 'process instance id: 0',
`process_definition_version` int(11) DEFAULT 0 COMMENT 'process definition version',
PRIMARY KEY (`id`) USING BTREE
) ENGINE=InnoDB DEFAULT CHARSET=utf8 ROW_FORMAT=DYNAMIC;
......
sql/dolphinscheduler_postgre.sql
浏览文件 @ 6ffd0196
......@@ -240,21 +240,23 @@ CREATE TABLE t_ds_alertgroup(
DROP TABLE IF EXISTS t_ds_command;
CREATE TABLE t_ds_command (
id int NOT NULL ,
command_type int DEFAULT NULL ,
process_definition_code bigint NOT NULL ,
command_param text ,
task_depend_type int DEFAULT NULL ,
failure_strategy int DEFAULT '0' ,
warning_type int DEFAULT '0' ,
warning_group_id int DEFAULT NULL ,
schedule_time timestamp DEFAULT NULL ,
start_time timestamp DEFAULT NULL ,
executor_id int DEFAULT NULL ,
update_time timestamp DEFAULT NULL ,
command_type int DEFAULT NULL ,
process_definition_code bigint NOT NULL ,
command_param text ,
task_depend_type int DEFAULT NULL ,
failure_strategy int DEFAULT '0' ,
warning_type int DEFAULT '0' ,
warning_group_id int DEFAULT NULL ,
schedule_time timestamp DEFAULT NULL ,
start_time timestamp DEFAULT NULL ,
executor_id int DEFAULT NULL ,
update_time timestamp DEFAULT NULL ,
process_instance_priority int DEFAULT NULL ,
worker_group varchar(64),
environment_code bigint DEFAULT '-1',
dry_run int DEFAULT '0' ,
worker_group varchar(64),
environment_code bigint DEFAULT '-1',
dry_run int DEFAULT '0' ,
process_instance_id int DEFAULT 0,
process_definition_version int DEFAULT 0,
PRIMARY KEY (id)
) ;
......@@ -284,23 +286,24 @@ CREATE TABLE t_ds_datasource (
DROP TABLE IF EXISTS t_ds_error_command;
CREATE TABLE t_ds_error_command (
id int NOT NULL ,
command_type int DEFAULT NULL ,
executor_id int DEFAULT NULL ,
process_definition_code bigint NOT NULL ,
command_param text ,
task_depend_type int DEFAULT NULL ,
failure_strategy int DEFAULT '0' ,
warning_type int DEFAULT '0' ,
warning_group_id int DEFAULT NULL ,
schedule_time timestamp DEFAULT NULL ,
start_time timestamp DEFAULT NULL ,
update_time timestamp DEFAULT NULL ,
id int NOT NULL ,
command_type int DEFAULT NULL ,
process_definition_code bigint NOT NULL ,
command_param text ,
task_depend_type int DEFAULT NULL ,
failure_strategy int DEFAULT '0' ,
warning_type int DEFAULT '0' ,
warning_group_id int DEFAULT NULL ,
schedule_time timestamp DEFAULT NULL ,
start_time timestamp DEFAULT NULL ,
executor_id int DEFAULT NULL ,
update_time timestamp DEFAULT NULL ,
process_instance_priority int DEFAULT NULL ,
worker_group varchar(64),
environment_code bigint DEFAULT '-1',
message text ,
dry_ru int DEFAULT '0' ,
worker_group varchar(64),
environment_code bigint DEFAULT '-1',
dry_run int DEFAULT '0' ,
process_instance_id int DEFAULT 0,
process_definition_version int DEFAULT 0,
PRIMARY KEY (id)
);
--
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或