natas_hw
/
dolphinscheduler0701
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Add WorkflowExecuteContext (#14544) 

(cherry picked from commit e6d94632b9)
This commit is contained in:
Wenjun Ruan 2023-07-14 18:40:01 +08:00 committed by Jay Chung
parent 65de246b11
commit adb106f47c
39 changed files with 1035 additions and 753 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

48
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/dto/gantt/GanttDto.java
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.dto.gantt;
import java.util.ArrayList;
@ -21,9 +22,9 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
/**
* gantt DTO
*/
import lombok.Data;
@Data
public class GanttDto {
/**
@ -37,9 +38,9 @@ public class GanttDto {
private List<Task> tasks = new ArrayList<>();
/**
* task name list
* task code list
*/
private List<String> taskNames;
private List<Long> taskNames;
/**
* task status map
@ -50,48 +51,19 @@ public class GanttDto {
this.taskStatus = new HashMap<>();
taskStatus.put("success", "success");
}
public GanttDto(int height, List<Task> tasks, List<String> taskNames) {
public GanttDto(int height, List<Task> tasks, List<Long> taskNames) {
this();
this.height = height;
this.tasks = tasks;
this.taskNames = taskNames;
}
public GanttDto(int height, List<Task> tasks, List<String> taskNames, Map<String, String> taskStatus) {
public GanttDto(int height, List<Task> tasks, List<Long> taskNames, Map<String, String> taskStatus) {
this.height = height;
this.tasks = tasks;
this.taskNames = taskNames;
this.taskStatus = taskStatus;
}
public int getHeight() {
return height;
}
public void setHeight(int height) {
this.height = height;
}
public List<Task> getTasks() {
return tasks;
}
public void setTasks(List<Task> tasks) {
this.tasks = tasks;
}
public List<String> getTaskNames() {
return taskNames;
}
public void setTaskNames(List<String> taskNames) {
this.taskNames = taskNames;
}
public Map<String, String> getTaskStatus() {
return taskStatus;
}
public void setTaskStatus(Map<String, String> taskStatus) {
this.taskStatus = taskStatus;
}
}

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

@ -1894,12 +1894,12 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
putMsg(result, Status.PROCESS_DEFINE_NOT_EXIST, String.valueOf(code));
return result;
}
DAG<String, TaskNode, TaskNodeRelation> dag = processService.genDagGraph(processDefinition);
DAG<Long, TaskNode, TaskNodeRelation> dag = processService.genDagGraph(processDefinition);
// nodes that are running
Map<String, List<TreeViewDto>> runningNodeMap = new ConcurrentHashMap<>();
Map<Long, List<TreeViewDto>> runningNodeMap = new ConcurrentHashMap<>();
// nodes that are waiting to run
Map<String, List<TreeViewDto>> waitingRunningNodeMap = new ConcurrentHashMap<>();
Map<Long, List<TreeViewDto>> waitingRunningNodeMap = new ConcurrentHashMap<>();
// List of process instances
List<ProcessInstance> processInstanceList = processInstanceService.queryByProcessDefineCode(code, limit);
@ -1937,16 +1937,16 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
List<TreeViewDto> parentTreeViewDtoList = new ArrayList<>();
parentTreeViewDtoList.add(parentTreeViewDto);
// Here is the encapsulation task instance
for (String startNode : dag.getBeginNode()) {
for (Long startNode : dag.getBeginNode()) {
runningNodeMap.put(startNode, parentTreeViewDtoList);
}
while (!ServerLifeCycleManager.isStopped()) {
Set<String> postNodeList;
Iterator<Map.Entry<String, List<TreeViewDto>>> iter = runningNodeMap.entrySet().iterator();
Set<Long> postNodeList;
Iterator<Map.Entry<Long, List<TreeViewDto>>> iter = runningNodeMap.entrySet().iterator();
while (iter.hasNext()) {
Map.Entry<String, List<TreeViewDto>> en = iter.next();
String nodeCode = en.getKey();
Map.Entry<Long, List<TreeViewDto>> en = iter.next();
Long nodeCode = en.getKey();
parentTreeViewDtoList = en.getValue();
TreeViewDto treeViewDto = new TreeViewDto();
@ -1957,8 +1957,8 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
// set treeViewDto instances
for (int i = limit - 1; i >= 0; i--) {
ProcessInstance processInstance = processInstanceList.get(i);
TaskInstance taskInstance = taskInstanceMapper.queryByInstanceIdAndCode(processInstance.getId(),
Long.parseLong(nodeCode));
TaskInstance taskInstance =
taskInstanceMapper.queryByInstanceIdAndCode(processInstance.getId(), nodeCode);
if (taskInstance == null) {
treeViewDto.getInstances().add(new Instance(-1, "not running", 0, "null"));
} else {
@ -1985,7 +1985,7 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
}
postNodeList = dag.getSubsequentNodes(nodeCode);
if (CollectionUtils.isNotEmpty(postNodeList)) {
for (String nextNodeCode : postNodeList) {
for (Long nextNodeCode : postNodeList) {
List<TreeViewDto> treeViewDtoList = waitingRunningNodeMap.get(nextNodeCode);
if (CollectionUtils.isEmpty(treeViewDtoList)) {
treeViewDtoList = new ArrayList<>();

12
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ProcessInstanceServiceImpl.java
View File

@ -1016,22 +1016,20 @@ public class ProcessInstanceServiceImpl extends BaseServiceImpl implements Proce
return result;
}
GanttDto ganttDto = new GanttDto();
DAG<String, TaskNode, TaskNodeRelation> dag = processService.genDagGraph(processDefinition);
DAG<Long, TaskNode, TaskNodeRelation> dag = processService.genDagGraph(processDefinition);
// topological sort
List<String> nodeList = dag.topologicalSort();
List<Long> nodeList = dag.topologicalSort();
ganttDto.setTaskNames(nodeList);
List<Task> taskList = new ArrayList<>();
if (CollectionUtils.isNotEmpty(nodeList)) {
List<Long> taskCodes = nodeList.stream().map(Long::parseLong).collect(Collectors.toList());
List<TaskInstance> taskInstances = taskInstanceMapper.queryByProcessInstanceIdsAndTaskCodes(
Collections.singletonList(processInstanceId), taskCodes);
for (String node : nodeList) {
Collections.singletonList(processInstanceId), nodeList);
for (Long node : nodeList) {
TaskInstance taskInstance = null;
for (TaskInstance instance : taskInstances) {
if (instance.getProcessInstanceId() == processInstanceId
&& instance.getTaskCode() == Long.parseLong(node)) {
if (instance.getProcessInstanceId() == processInstanceId && instance.getTaskCode() == node) {
taskInstance = instance;
break;
}

6
dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/ProcessInstanceServiceTest.java
View File

@ -745,9 +745,9 @@ public class ProcessInstanceServiceTest {
processInstance.getProcessDefinitionVersion())).thenReturn(new ProcessDefinitionLog());
when(processInstanceMapper.queryDetailById(1)).thenReturn(processInstance);
when(taskInstanceMapper.queryByInstanceIdAndName(Mockito.anyInt(), Mockito.any())).thenReturn(taskInstance);
DAG<String, TaskNode, TaskNodeRelation> graph = new DAG<>();
for (int i = 1; i <= 7; ++i) {
graph.addNode(i + "", new TaskNode());
DAG<Long, TaskNode, TaskNodeRelation> graph = new DAG<>();
for (long i = 1; i <= 7; ++i) {
graph.addNode(i, new TaskNode());
}
when(processService.genDagGraph(Mockito.any(ProcessDefinition.class)))

60
dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/model/TaskNodeRelation.java
View File

@ -17,63 +17,23 @@
package org.apache.dolphinscheduler.common.model;
import java.util.Objects;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
@Data
@AllArgsConstructor
@NoArgsConstructor
public class TaskNodeRelation {
/**
* task start node name
* task start node Code
*/
private String startNode;
private Long startNode;
/**
* task end node name
* task end node Code
*/
private String endNode;
private Long endNode;
public TaskNodeRelation() {
}
public TaskNodeRelation(String startNode, String endNode) {
this.startNode = startNode;
this.endNode = endNode;
}
public String getStartNode() {
return startNode;
}
public void setStartNode(String startNode) {
this.startNode = startNode;
}
public String getEndNode() {
return endNode;
}
public void setEndNode(String endNode) {
this.endNode = endNode;
}
@Override
public boolean equals(Object o) {
if (!(o instanceof TaskNodeRelation)) {
return false;
}
TaskNodeRelation relation = (TaskNodeRelation) o;
return (relation.getStartNode().equals(this.startNode) && relation.getEndNode().equals(this.endNode));
}
@Override
public String toString() {
return "TaskNodeRelation{"
+ "startNode='" + startNode + '\''
+ ", endNode='" + endNode + '\''
+ '}';
}
@Override
public int hashCode() {
return Objects.hash(startNode, endNode);
}
}

11
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/event/WorkflowStartEventHandler.java
View File

@ -25,7 +25,7 @@ import org.apache.dolphinscheduler.server.master.metrics.ProcessInstanceMetrics;
import org.apache.dolphinscheduler.server.master.runner.StateWheelExecuteThread;
import org.apache.dolphinscheduler.server.master.runner.WorkflowExecuteRunnable;
import org.apache.dolphinscheduler.server.master.runner.WorkflowExecuteThreadPool;
import org.apache.dolphinscheduler.server.master.runner.WorkflowSubmitStatus;
import org.apache.dolphinscheduler.server.master.runner.WorkflowStartStatus;
import java.util.concurrent.CompletableFuture;
@ -56,17 +56,18 @@ public class WorkflowStartEventHandler implements WorkflowEventHandler {
throw new WorkflowEventHandleError(
"The workflow start event is invalid, cannot find the workflow instance from cache");
}
ProcessInstance processInstance = workflowExecuteRunnable.getProcessInstance();
ProcessInstance processInstance =
workflowExecuteRunnable.getWorkflowExecuteContext().getWorkflowInstance();
ProcessInstanceMetrics.incProcessInstanceByStateAndProcessDefinitionCode("submit",
processInstance.getProcessDefinitionCode().toString());
CompletableFuture.supplyAsync(workflowExecuteRunnable::call, workflowExecuteThreadPool)
.thenAccept(workflowSubmitStatus -> {
if (WorkflowSubmitStatus.SUCCESS == workflowSubmitStatus) {
.thenAccept(workflowStartStatus -> {
if (WorkflowStartStatus.SUCCESS == workflowStartStatus) {
log.info("Success submit the workflow instance");
if (processInstance.getTimeout() > 0) {
stateWheelExecuteThread.addProcess4TimeoutCheck(processInstance);
}
} else if (WorkflowSubmitStatus.FAILED == workflowSubmitStatus) {
} else if (WorkflowStartStatus.FAILED == workflowStartStatus) {
log.error(
"Failed to submit the workflow instance, will resend the workflow start event: {}",
workflowEvent);

3
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/event/WorkflowStateEventHandler.java
View File

@ -35,7 +35,8 @@ public class WorkflowStateEventHandler implements StateEventHandler {
public boolean handleStateEvent(WorkflowExecuteRunnable workflowExecuteRunnable,
StateEvent stateEvent) throws StateEventHandleException {
WorkflowStateEvent workflowStateEvent = (WorkflowStateEvent) stateEvent;
ProcessInstance processInstance = workflowExecuteRunnable.getProcessInstance();
ProcessInstance processInstance =
workflowExecuteRunnable.getWorkflowExecuteContext().getWorkflowInstance();
ProcessDefinition processDefinition = processInstance.getProcessDefinition();
measureProcessState(workflowStateEvent, processInstance.getProcessDefinitionCode().toString());

3
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/event/WorkflowTimeoutStateEventHandler.java
View File

@ -33,7 +33,8 @@ public class WorkflowTimeoutStateEventHandler implements StateEventHandler {
@Override
public boolean handleStateEvent(WorkflowExecuteRunnable workflowExecuteRunnable, StateEvent stateEvent) {
log.info("Handle workflow instance timeout event");
ProcessInstance processInstance = workflowExecuteRunnable.getProcessInstance();
ProcessInstance processInstance =
workflowExecuteRunnable.getWorkflowExecuteContext().getWorkflowInstance();
ProcessInstanceMetrics.incProcessInstanceByStateAndProcessDefinitionCode("timeout",
processInstance.getProcessDefinitionCode().toString());
workflowExecuteRunnable.processTimeout();

33
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/graph/IWorkflowGraph.java Normal file
View File

@ -0,0 +1,33 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dolphinscheduler.server.master.graph;
import org.apache.dolphinscheduler.common.graph.DAG;
import org.apache.dolphinscheduler.common.model.TaskNodeRelation;
import org.apache.dolphinscheduler.service.model.TaskNode;
public interface IWorkflowGraph {
TaskNode getTaskNodeByCode(Long taskCode);
// todo: refactor DAG class
DAG<Long, TaskNode, TaskNodeRelation> getDag();
boolean isForbiddenTask(Long taskCode);
}

69
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/graph/WorkflowGraph.java Normal file
View File

@ -0,0 +1,69 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dolphinscheduler.server.master.graph;
import static com.google.common.base.Preconditions.checkNotNull;
import org.apache.dolphinscheduler.common.graph.DAG;
import org.apache.dolphinscheduler.common.model.TaskNodeRelation;
import org.apache.dolphinscheduler.service.model.TaskNode;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.function.Function;
import java.util.stream.Collectors;
public class WorkflowGraph implements IWorkflowGraph {
private final Map<Long, TaskNode> taskNodeMap;
private final DAG<Long, TaskNode, TaskNodeRelation> dag;
private final Set<Long> forbiddenTaskCodes;
public WorkflowGraph(List<TaskNode> taskNodes,
DAG<Long, TaskNode, TaskNodeRelation> dag) {
checkNotNull(taskNodes, "taskNodes can not be null");
checkNotNull(dag, "dag can not be null");
this.taskNodeMap = taskNodes.stream().collect(Collectors.toMap(TaskNode::getCode, Function.identity()));
this.dag = dag;
forbiddenTaskCodes =
taskNodes.stream().filter(TaskNode::isForbidden).map(TaskNode::getCode).collect(Collectors.toSet());
}
@Override
public TaskNode getTaskNodeByCode(Long taskCode) {
TaskNode taskNode = taskNodeMap.get(taskCode);
if (taskNode == null) {
throw new IllegalArgumentException("task node not found, taskCode: " + taskCode);
}
return taskNode;
}
@Override
public DAG<Long, TaskNode, TaskNodeRelation> getDag() {
return dag;
}
@Override
public boolean isForbiddenTask(Long taskCode) {
return forbiddenTaskCodes.contains(taskCode);
}
}

132
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/graph/WorkflowGraphFactory.java Normal file
View File

@ -0,0 +1,132 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dolphinscheduler.server.master.graph;
import static org.apache.dolphinscheduler.common.constants.CommandKeyConstants.CMD_PARAM_RECOVERY_START_NODE_STRING;
import static org.apache.dolphinscheduler.common.constants.CommandKeyConstants.CMD_PARAM_START_NODES;
import static org.apache.dolphinscheduler.common.constants.Constants.COMMA;
import org.apache.dolphinscheduler.common.constants.Constants;
import org.apache.dolphinscheduler.common.graph.DAG;
import org.apache.dolphinscheduler.common.model.TaskNodeRelation;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
import org.apache.dolphinscheduler.dao.entity.ProcessTaskRelation;
import org.apache.dolphinscheduler.dao.entity.TaskDefinitionLog;
import org.apache.dolphinscheduler.dao.entity.TaskInstance;
import org.apache.dolphinscheduler.dao.repository.TaskDefinitionLogDao;
import org.apache.dolphinscheduler.dao.repository.TaskInstanceDao;
import org.apache.dolphinscheduler.service.model.TaskNode;
import org.apache.dolphinscheduler.service.process.ProcessDag;
import org.apache.dolphinscheduler.service.process.ProcessService;
import org.apache.dolphinscheduler.service.utils.DagHelper;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
@Slf4j
@Component
public class WorkflowGraphFactory {
@Autowired
private ProcessService processService;
@Autowired
private TaskInstanceDao taskInstanceDao;
@Autowired
private TaskDefinitionLogDao taskDefinitionLogDao;
public IWorkflowGraph createWorkflowGraph(ProcessInstance workflowInstance) throws Exception {
List<ProcessTaskRelation> processTaskRelations =
processService.findRelationByCode(workflowInstance.getProcessDefinitionCode(),
workflowInstance.getProcessDefinitionVersion());
List<TaskDefinitionLog> taskDefinitionLogs =
taskDefinitionLogDao.queryTaskDefineLogList(processTaskRelations);
List<TaskNode> taskNodeList = processService.transformTask(processTaskRelations, taskDefinitionLogs);
// generate process to get DAG info
List<Long> recoveryTaskNodeCodeList = getRecoveryTaskNodeCodeList(workflowInstance.getCommandParam());
List<Long> startNodeNameList = parseStartNodeName(workflowInstance.getCommandParam());
ProcessDag processDag = DagHelper.generateFlowDag(taskNodeList, startNodeNameList, recoveryTaskNodeCodeList,
workflowInstance.getTaskDependType());
if (processDag == null) {
log.error("ProcessDag is null");
throw new IllegalArgumentException("Create WorkflowGraph failed, ProcessDag is null");
}
// generate process dag
DAG<Long, TaskNode, TaskNodeRelation> dagGraph = DagHelper.buildDagGraph(processDag);
log.debug("Build dag success, dag: {}", dagGraph);
return new WorkflowGraph(taskNodeList, dagGraph);
}
/**
* generate start node code list from parsing command param;
* if "StartNodeIdList" exists in command param, return StartNodeIdList
*
* @return recovery node code list
*/
private List<Long> getRecoveryTaskNodeCodeList(String cmdParam) {
Map<String, String> paramMap = JSONUtils.toMap(cmdParam);
// todo: Can we use a better way to set the recover taskInstanceId list? rather then use the cmdParam
if (paramMap != null && paramMap.containsKey(CMD_PARAM_RECOVERY_START_NODE_STRING)) {
List<Integer> startTaskInstanceIds = Arrays.stream(paramMap.get(CMD_PARAM_RECOVERY_START_NODE_STRING)
.split(COMMA))
.filter(StringUtils::isNotEmpty)
.map(Integer::valueOf)
.collect(Collectors.toList());
if (CollectionUtils.isNotEmpty(startTaskInstanceIds)) {
return taskInstanceDao.queryByIds(startTaskInstanceIds).stream().map(TaskInstance::getTaskCode)
.collect(Collectors.toList());
}
}
return Collections.emptyList();
}
private List<Long> parseStartNodeName(String cmdParam) {
List<Long> startNodeNameList = new ArrayList<>();
Map<String, String> paramMap = JSONUtils.toMap(cmdParam);
if (paramMap == null) {
return startNodeNameList;
}
if (paramMap.containsKey(CMD_PARAM_START_NODES)) {
startNodeNameList = Arrays.asList(paramMap.get(CMD_PARAM_START_NODES).split(Constants.COMMA))
.stream()
.map(String::trim)
.map(Long::valueOf)
.collect(Collectors.toList());
}
return startNodeNameList;
}
}

3
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/EventExecuteService.java
View File

@ -78,7 +78,8 @@ public class EventExecuteService extends BaseDaemonThread {
private void workflowEventHandler() {
for (WorkflowExecuteRunnable workflowExecuteThread : this.processInstanceExecCacheManager.getAll()) {
try {
LogUtils.setWorkflowInstanceIdMDC(workflowExecuteThread.getProcessInstance().getId());
LogUtils.setWorkflowInstanceIdMDC(
workflowExecuteThread.getWorkflowExecuteContext().getWorkflowInstance().getId());
workflowExecuteThreadPool.executeEvent(workflowExecuteThread);
} finally {

33
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/IWorkflowExecuteContext.java Normal file
View File

@ -0,0 +1,33 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dolphinscheduler.server.master.runner;
import org.apache.dolphinscheduler.dao.entity.ProcessDefinition;
import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
import org.apache.dolphinscheduler.server.master.graph.IWorkflowGraph;
// todo: Add method to manage the task instance
public interface IWorkflowExecuteContext {
ProcessDefinition getWorkflowDefinition();
ProcessInstance getWorkflowInstance();
IWorkflowGraph getWorkflowGraph();
}

32
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/IWorkflowExecuteRunnable.java Normal file
View File

@ -0,0 +1,32 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dolphinscheduler.server.master.runner;
import java.util.concurrent.Callable;
public interface IWorkflowExecuteRunnable extends Callable<WorkflowStartStatus> {
// todo: add control method to manage the workflow runnable e.g. pause/stop ....
@Override
default WorkflowStartStatus call() {
return startWorkflow();
}
WorkflowStartStatus startWorkflow();
}

3
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterSchedulerBootstrap.java
View File

@ -134,7 +134,8 @@ public class MasterSchedulerBootstrap extends BaseDaemonThread implements AutoCl
try {
WorkflowExecuteRunnable workflowExecuteRunnable =
workflowExecuteRunnableFactory.createWorkflowExecuteRunnable(command);
ProcessInstance processInstance = workflowExecuteRunnable.getProcessInstance();
ProcessInstance processInstance = workflowExecuteRunnable
.getWorkflowExecuteContext().getWorkflowInstance();
if (processInstanceExecCacheManager.contains(processInstance.getId())) {
log.error(
"The workflow instance is already been cached, this case shouldn't be happened");

6
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/StateWheelExecuteThread.java
View File

@ -141,7 +141,8 @@ public class StateWheelExecuteThread extends BaseDaemonThread {
processInstanceTimeoutCheckList.remove(processInstanceId);
continue;
}
ProcessInstance processInstance = workflowExecuteThread.getProcessInstance();
ProcessInstance processInstance =
workflowExecuteThread.getWorkflowExecuteContext().getWorkflowInstance();
if (processInstance == null) {
log.warn("Check workflow timeout failed, the workflowInstance is null");
continue;
@ -284,7 +285,8 @@ public class StateWheelExecuteThread extends BaseDaemonThread {
Optional<TaskInstance> taskInstanceOptional =
workflowExecuteThread.getRetryTaskInstanceByTaskCode(taskCode);
ProcessInstance processInstance = workflowExecuteThread.getProcessInstance();
ProcessInstance processInstance =
workflowExecuteThread.getWorkflowExecuteContext().getWorkflowInstance();
if (processInstance.getState().isReadyStop()) {
log.warn(

54
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteContext.java Normal file
View File

@ -0,0 +1,54 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dolphinscheduler.server.master.runner;
import static com.google.common.base.Preconditions.checkNotNull;
import org.apache.dolphinscheduler.dao.entity.ProcessDefinition;
import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
import org.apache.dolphinscheduler.server.master.graph.IWorkflowGraph;
import lombok.Getter;
public class WorkflowExecuteContext implements IWorkflowExecuteContext {
@Getter
private final ProcessDefinition workflowDefinition;
@Getter
private final ProcessInstance workflowInstance;
// This is the task definition graph
// todo: we need to add a task instance graph, then move the task instance from WorkflowExecuteRunnable to
// WorkflowExecuteContext
@Getter
private final IWorkflowGraph workflowGraph;
public WorkflowExecuteContext(ProcessDefinition workflowDefinition,
ProcessInstance workflowInstance,
IWorkflowGraph workflowGraph) {
checkNotNull(workflowDefinition, "workflowDefinition is null");
checkNotNull(workflowInstance, "workflowInstance is null");
checkNotNull(workflowGraph, "workflowGraph is null");
this.workflowDefinition = workflowDefinition;
this.workflowInstance = workflowInstance;
this.workflowGraph = workflowGraph;
}
}

99
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteContextFactory.java Normal file
View File

@ -0,0 +1,99 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dolphinscheduler.server.master.runner;
import org.apache.dolphinscheduler.common.enums.SlotCheckState;
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.server.master.config.MasterConfig;
import org.apache.dolphinscheduler.server.master.graph.IWorkflowGraph;
import org.apache.dolphinscheduler.server.master.graph.WorkflowGraphFactory;
import org.apache.dolphinscheduler.server.master.metrics.ProcessInstanceMetrics;
import org.apache.dolphinscheduler.server.master.registry.ServerNodeManager;
import org.apache.dolphinscheduler.service.exceptions.CronParseException;
import org.apache.dolphinscheduler.service.process.ProcessService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
@Slf4j
@Component
public class WorkflowExecuteContextFactory {
@Autowired
private ServerNodeManager serverNodeManager;
@Autowired
private ProcessService processService;
@Autowired
private WorkflowGraphFactory workflowGraphFactory;
@Autowired
private MasterConfig masterConfig;
public IWorkflowExecuteContext createWorkflowExecuteRunnableContext(Command command) throws Exception {
ProcessInstance workflowInstance = createWorkflowInstance(command);
ProcessDefinition workflowDefinition = processService.findProcessDefinition(
workflowInstance.getProcessDefinitionCode(), workflowInstance.getProcessDefinitionVersion());
workflowInstance.setProcessDefinition(workflowDefinition);
IWorkflowGraph workflowGraph = workflowGraphFactory.createWorkflowGraph(workflowInstance);
return new WorkflowExecuteContext(
workflowDefinition,
workflowInstance,
workflowGraph);
}
private ProcessInstance createWorkflowInstance(Command command) throws CronParseException {
long commandTransformStartTime = System.currentTimeMillis();
// Note: this check is not safe, the slot may change after command transform.
// We use the database transaction in `handleCommand` so that we can guarantee the command will
// always be executed
// by only one master
SlotCheckState slotCheckState = slotCheck(command);
if (slotCheckState.equals(SlotCheckState.CHANGE) || slotCheckState.equals(SlotCheckState.INJECT)) {
log.info("Master handle command {} skip, slot check state: {}", command.getId(), slotCheckState);
throw new RuntimeException("Slot check failed the current state: " + slotCheckState);
}
ProcessInstance processInstance = processService.handleCommand(masterConfig.getMasterAddress(), command);
log.info("Master handle command {} end, create process instance {}", command.getId(), processInstance.getId());
ProcessInstanceMetrics
.recordProcessInstanceGenerateTime(System.currentTimeMillis() - commandTransformStartTime);
return processInstance;
}
private SlotCheckState slotCheck(Command command) {
int slot = serverNodeManager.getSlot();
int masterSize = serverNodeManager.getMasterSize();
SlotCheckState state;
if (masterSize <= 0) {
state = SlotCheckState.CHANGE;
} else if (command.getId() % masterSize == slot) {
state = SlotCheckState.PASS;
} else {
state = SlotCheckState.INJECT;
}
return state;
}
}

555
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteRunnable.java
View File

File diff suppressed because it is too large Load Diff

49
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteRunnableFactory.java
View File

@ -17,16 +17,12 @@
package org.apache.dolphinscheduler.server.master.runner;
import org.apache.dolphinscheduler.common.enums.SlotCheckState;
import org.apache.dolphinscheduler.dao.entity.Command;
import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
import org.apache.dolphinscheduler.dao.repository.ProcessInstanceDao;
import org.apache.dolphinscheduler.dao.repository.TaskDefinitionLogDao;
import org.apache.dolphinscheduler.dao.repository.TaskInstanceDao;
import org.apache.dolphinscheduler.server.master.config.MasterConfig;
import org.apache.dolphinscheduler.server.master.exception.WorkflowCreateException;
import org.apache.dolphinscheduler.server.master.metrics.ProcessInstanceMetrics;
import org.apache.dolphinscheduler.server.master.registry.ServerNodeManager;
import org.apache.dolphinscheduler.server.master.rpc.MasterRpcClient;
import org.apache.dolphinscheduler.server.master.runner.execute.DefaultTaskExecuteRunnableFactory;
import org.apache.dolphinscheduler.service.alert.ProcessAlertManager;
@ -43,9 +39,6 @@ import org.springframework.stereotype.Component;
@Component
public class WorkflowExecuteRunnableFactory {
@Autowired
private ServerNodeManager serverNodeManager;
@Autowired
private CommandService commandService;
@ -79,10 +72,15 @@ public class WorkflowExecuteRunnableFactory {
@Autowired
private DefaultTaskExecuteRunnableFactory defaultTaskExecuteRunnableFactory;
@Autowired
private WorkflowExecuteContextFactory workflowExecuteContextFactory;
public WorkflowExecuteRunnable createWorkflowExecuteRunnable(Command command) throws WorkflowCreateException {
try {
ProcessInstance workflowInstance = createWorkflowInstance(command);
return new WorkflowExecuteRunnable(workflowInstance,
IWorkflowExecuteContext workflowExecuteRunnableContext =
workflowExecuteContextFactory.createWorkflowExecuteRunnableContext(command);
return new WorkflowExecuteRunnable(
workflowExecuteRunnableContext,
commandService,
processService,
processInstanceDao,
@ -92,43 +90,10 @@ public class WorkflowExecuteRunnableFactory {
stateWheelExecuteThread,
curingGlobalParamsService,
taskInstanceDao,
taskDefinitionLogDao,
defaultTaskExecuteRunnableFactory);
} catch (Exception ex) {
throw new WorkflowCreateException("Create workflow execute runnable failed", ex);
}
}
private ProcessInstance createWorkflowInstance(Command command) throws Exception {
long commandTransformStartTime = System.currentTimeMillis();
// Note: this check is not safe, the slot may change after command transform.
// We use the database transaction in `handleCommand` so that we can guarantee the command will
// always be executed
// by only one master
SlotCheckState slotCheckState = slotCheck(command);
if (slotCheckState.equals(SlotCheckState.CHANGE) || slotCheckState.equals(SlotCheckState.INJECT)) {
log.info("Master handle command {} skip, slot check state: {}", command.getId(), slotCheckState);
throw new RuntimeException("Slot check failed the current state: " + slotCheckState);
}
ProcessInstance processInstance = processService.handleCommand(masterConfig.getMasterAddress(), command);
log.info("Master handle command {} end, create process instance {}", command.getId(), processInstance.getId());
ProcessInstanceMetrics
.recordProcessInstanceGenerateTime(System.currentTimeMillis() - commandTransformStartTime);
return processInstance;
}
private SlotCheckState slotCheck(Command command) {
int slot = serverNodeManager.getSlot();
int masterSize = serverNodeManager.getMasterSize();
SlotCheckState state;
if (masterSize <= 0) {
state = SlotCheckState.CHANGE;
} else if (command.getId() % masterSize == slot) {
state = SlotCheckState.PASS;
} else {
state = SlotCheckState.INJECT;
}
return state;
}
}

31
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteThreadPool.java
View File

@ -74,7 +74,7 @@ public class WorkflowExecuteThreadPool extends ThreadPoolTaskExecutor {
/**
* multi-thread filter, avoid handling workflow at the same time
*/
private ConcurrentHashMap<String, WorkflowExecuteRunnable> multiThreadFilterMap = new ConcurrentHashMap<>();
private ConcurrentHashMap<Integer, WorkflowExecuteRunnable> multiThreadFilterMap = new ConcurrentHashMap<>();
@PostConstruct
private void init() {
@ -106,22 +106,26 @@ public class WorkflowExecuteThreadPool extends ThreadPoolTaskExecutor {
if (!workflowExecuteThread.isStart() || workflowExecuteThread.eventSize() == 0) {
return;
}
if (multiThreadFilterMap.containsKey(workflowExecuteThread.getKey())) {
IWorkflowExecuteContext workflowExecuteRunnableContext =
workflowExecuteThread.getWorkflowExecuteContext();
Integer workflowInstanceId = workflowExecuteRunnableContext.getWorkflowInstance().getId();
if (multiThreadFilterMap.containsKey(workflowInstanceId)) {
log.debug("The workflow has been executed by another thread");
return;
}
multiThreadFilterMap.put(workflowExecuteThread.getKey(), workflowExecuteThread);
int processInstanceId = workflowExecuteThread.getProcessInstance().getId();
multiThreadFilterMap.put(workflowInstanceId, workflowExecuteThread);
ListenableFuture<?> future = this.submitListenable(workflowExecuteThread::handleEvents);
future.addCallback(new ListenableFutureCallback() {
@Override
public void onFailure(Throwable ex) {
LogUtils.setWorkflowInstanceIdMDC(processInstanceId);
LogUtils.setWorkflowInstanceIdMDC(workflowInstanceId);
try {
log.error("Workflow instance events handle failed", ex);
notifyProcessChanged(workflowExecuteThread.getProcessInstance());
multiThreadFilterMap.remove(workflowExecuteThread.getKey());
notifyProcessChanged(
workflowExecuteThread.getWorkflowExecuteContext().getWorkflowInstance());
multiThreadFilterMap.remove(workflowInstanceId);
} finally {
LogUtils.removeWorkflowInstanceIdMDC();
}
@ -130,19 +134,22 @@ public class WorkflowExecuteThreadPool extends ThreadPoolTaskExecutor {
@Override
public void onSuccess(Object result) {
try {
LogUtils.setWorkflowInstanceIdMDC(workflowExecuteThread.getProcessInstance().getId());
LogUtils.setWorkflowInstanceIdMDC(
workflowExecuteThread.getWorkflowExecuteContext().getWorkflowInstance().getId());
if (workflowExecuteThread.workFlowFinish() && workflowExecuteThread.eventSize() == 0) {
stateWheelExecuteThread
.removeProcess4TimeoutCheck(workflowExecuteThread.getProcessInstance().getId());
processInstanceExecCacheManager.removeByProcessInstanceId(processInstanceId);
notifyProcessChanged(workflowExecuteThread.getProcessInstance());
.removeProcess4TimeoutCheck(workflowExecuteThread.getWorkflowExecuteContext()
.getWorkflowInstance().getId());
processInstanceExecCacheManager.removeByProcessInstanceId(workflowInstanceId);
notifyProcessChanged(
workflowExecuteThread.getWorkflowExecuteContext().getWorkflowInstance());
log.info("Workflow instance is finished.");
}
} catch (Exception e) {
log.error("Workflow instance is finished, but notify changed error", e);
} finally {
// make sure the process has been removed from multiThreadFilterMap
multiThreadFilterMap.remove(workflowExecuteThread.getKey());
multiThreadFilterMap.remove(workflowInstanceId);
LogUtils.removeWorkflowInstanceIdMDC();
}
}

2
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/WorkflowSubmitStatus.java → dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/WorkflowStartStatus.java
View File

@ -17,7 +17,7 @@
package org.apache.dolphinscheduler.server.master.runner;
public enum WorkflowSubmitStatus {
public enum WorkflowStartStatus {
/**
* Submit success
*/

2
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/execute/DefaultTaskExecuteRunnableFactory.java
View File

@ -48,7 +48,7 @@ public class DefaultTaskExecuteRunnableFactory implements TaskExecuteRunnableFac
processInstanceExecCacheManager.getByProcessInstanceId(taskInstance.getProcessInstanceId());
try {
return new DefaultTaskExecuteRunnable(
workflowExecuteRunnable.getProcessInstance(),
workflowExecuteRunnable.getWorkflowExecuteContext().getWorkflowInstance(),
taskInstance,
taskExecutionContextFactory.createTaskExecutionContext(taskInstance),
taskOperatorManager);

3
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/task/blocking/BlockingLogicTask.java
View File

@ -80,7 +80,8 @@ public class BlockingLogicTask extends BaseSyncLogicTask<BlockingParameters> {
boolean isBlocked = (expected == conditionResult);
log.info("blocking opportunity: expected-->{}, actual-->{}", expected, conditionResult);
ProcessInstance workflowInstance = processInstanceExecCacheManager
.getByProcessInstanceId(taskExecutionContext.getProcessInstanceId()).getProcessInstance();
.getByProcessInstanceId(taskExecutionContext.getProcessInstanceId()).getWorkflowExecuteContext()
.getWorkflowInstance();
workflowInstance.setBlocked(isBlocked);
if (isBlocked) {
workflowInstance.setStateWithDesc(WorkflowExecutionStatus.READY_BLOCK, "ready block");

6
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/task/switchtask/SwitchLogicTask.java
View File

@ -66,7 +66,7 @@ public class SwitchLogicTask extends BaseSyncLogicTask<SwitchParameters> {
.getSwitchDependency());
WorkflowExecuteRunnable workflowExecuteRunnable =
processInstanceExecCacheManager.getByProcessInstanceId(taskExecutionContext.getProcessInstanceId());
this.processInstance = workflowExecuteRunnable.getProcessInstance();
this.processInstance = workflowExecuteRunnable.getWorkflowExecuteContext().getWorkflowInstance();
this.taskInstance = workflowExecuteRunnable.getTaskInstance(taskExecutionContext.getTaskInstanceId())
.orElseThrow(() -> new LogicTaskInitializeException(
"Cannot find the task instance in workflow execute runnable"));
@ -168,8 +168,8 @@ public class SwitchLogicTask extends BaseSyncLogicTask<SwitchParameters> {
if (CollectionUtils.isEmpty(switchResult.getNextNode())) {
return false;
}
for (String nextNode : switchResult.getNextNode()) {
if (StringUtils.isEmpty(nextNode)) {
for (Long nextNode : switchResult.getNextNode()) {
if (nextNode == null) {
return false;
}
}

3
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/service/ExecutingService.java
View File

@ -55,7 +55,8 @@ public class ExecutingService {
}
try {
WorkflowExecuteDto workflowExecuteDto = new WorkflowExecuteDto();
BeanUtils.copyProperties(workflowExecuteDto, workflowExecuteRunnable.getProcessInstance());
BeanUtils.copyProperties(workflowExecuteDto,
workflowExecuteRunnable.getWorkflowExecuteContext().getWorkflowInstance());
List<TaskInstanceExecuteDto> taskInstanceList = Lists.newArrayList();
if (CollectionUtils.isNotEmpty(workflowExecuteRunnable.getAllTaskInstances())) {
for (TaskInstance taskInstance : workflowExecuteRunnable.getAllTaskInstances()) {

3
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/service/WorkerFailoverService.java
View File

@ -131,7 +131,8 @@ public class WorkerFailoverService {
if (workflowExecuteRunnable == null) {
return null;
}
return workflowExecuteRunnable.getProcessInstance();
return workflowExecuteRunnable.getWorkflowExecuteContext()
.getWorkflowInstance();
});
if (!checkTaskInstanceNeedFailover(needFailoverWorkerStartTime, processInstance, taskInstance)) {
log.info("Worker[{}] the current taskInstance doesn't need to failover", workerHost);

8
dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/cache/impl/ProcessInstanceExecCacheManagerImplTest.java vendored
View File

@ -27,7 +27,6 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.InjectMocks;
import org.mockito.Mock;
import org.mockito.Mockito;
import org.mockito.junit.jupiter.MockitoExtension;
import org.mockito.junit.jupiter.MockitoSettings;
import org.mockito.quality.Strictness;
@ -44,14 +43,13 @@ public class ProcessInstanceExecCacheManagerImplTest {
@BeforeEach
public void before() {
Mockito.when(workflowExecuteThread.getKey()).thenReturn("workflowExecuteThread1");
processInstanceExecCacheManager.cache(1, workflowExecuteThread);
}
@Test
public void testGetByProcessInstanceId() {
WorkflowExecuteRunnable workflowExecuteThread = processInstanceExecCacheManager.getByProcessInstanceId(1);
Assertions.assertEquals("workflowExecuteThread1", workflowExecuteThread.getKey());
Assertions.assertNotNull(workflowExecuteThread);
}
@Test
@ -61,9 +59,7 @@ public class ProcessInstanceExecCacheManagerImplTest {
@Test
public void testCacheNull() {
Assertions.assertThrows(NullPointerException.class, () -> {
processInstanceExecCacheManager.cache(2, null);
});
Assertions.assertThrows(NullPointerException.class, () -> processInstanceExecCacheManager.cache(2, null));
WorkflowExecuteRunnable workflowExecuteThread = processInstanceExecCacheManager.getByProcessInstanceId(2);
Assertions.assertNull(workflowExecuteThread);
}

58
dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteRunnableTest.java
View File

@ -36,6 +36,7 @@ import org.apache.dolphinscheduler.dao.repository.ProcessInstanceDao;
import org.apache.dolphinscheduler.dao.repository.TaskDefinitionLogDao;
import org.apache.dolphinscheduler.dao.repository.TaskInstanceDao;
import org.apache.dolphinscheduler.server.master.config.MasterConfig;
import org.apache.dolphinscheduler.server.master.graph.IWorkflowGraph;
import org.apache.dolphinscheduler.server.master.rpc.MasterRpcClient;
import org.apache.dolphinscheduler.server.master.runner.execute.DefaultTaskExecuteRunnableFactory;
import org.apache.dolphinscheduler.service.alert.ProcessAlertManager;
@ -98,6 +99,8 @@ public class WorkflowExecuteRunnableTest {
private DefaultTaskExecuteRunnableFactory defaultTaskExecuteRunnableFactory;
private WorkflowExecuteContextFactory workflowExecuteContextFactory;
@BeforeEach
public void init() throws Exception {
applicationContext = Mockito.mock(ApplicationContext.class);
@ -112,6 +115,8 @@ public class WorkflowExecuteRunnableTest {
taskInstanceDao = Mockito.mock(TaskInstanceDao.class);
taskDefinitionLogDao = Mockito.mock(TaskDefinitionLogDao.class);
defaultTaskExecuteRunnableFactory = Mockito.mock(DefaultTaskExecuteRunnableFactory.class);
workflowExecuteContextFactory = Mockito.mock(WorkflowExecuteContextFactory.class);
Map<String, String> cmdParam = new HashMap<>();
cmdParam.put(CMD_PARAM_COMPLEMENT_DATA_START_DATE, "2020-01-01 00:00:00");
cmdParam.put(CMD_PARAM_COMPLEMENT_DATA_END_DATE, "2020-01-20 23:00:00");
@ -124,14 +129,25 @@ public class WorkflowExecuteRunnableTest {
curingGlobalParamsService = Mockito.mock(CuringParamsService.class);
MasterRpcClient masterRpcClient = Mockito.mock(MasterRpcClient.class);
ProcessAlertManager processAlertManager = Mockito.mock(ProcessAlertManager.class);
WorkflowExecuteContext workflowExecuteContext = Mockito.mock(WorkflowExecuteContext.class);
Mockito.when(workflowExecuteContext.getWorkflowInstance()).thenReturn(processInstance);
IWorkflowGraph workflowGraph = Mockito.mock(IWorkflowGraph.class);
Mockito.when(workflowExecuteContext.getWorkflowGraph()).thenReturn(workflowGraph);
Mockito.when(workflowGraph.getDag()).thenReturn(new DAG<>());
workflowExecuteThread = Mockito.spy(
new WorkflowExecuteRunnable(processInstance, commandService, processService, processInstanceDao,
new WorkflowExecuteRunnable(
workflowExecuteContext,
commandService,
processService,
processInstanceDao,
masterRpcClient,
processAlertManager, config, stateWheelExecuteThread, curingGlobalParamsService,
taskInstanceDao, taskDefinitionLogDao, defaultTaskExecuteRunnableFactory));
Field dag = WorkflowExecuteRunnable.class.getDeclaredField("dag");
dag.setAccessible(true);
dag.set(workflowExecuteThread, new DAG());
processAlertManager,
config,
stateWheelExecuteThread,
curingGlobalParamsService,
taskInstanceDao,
defaultTaskExecuteRunnableFactory));
}
@Test
@ -187,9 +203,9 @@ public class WorkflowExecuteRunnableTest {
@Test
public void testGetPreVarPool() {
try {
Set<String> preTaskName = new HashSet<>();
preTaskName.add(Long.toString(1));
preTaskName.add(Long.toString(2));
Set<Long> preTaskName = new HashSet<>();
preTaskName.add(1L);
preTaskName.add(2L);
TaskInstance taskInstance = new TaskInstance();
@ -271,7 +287,7 @@ public class WorkflowExecuteRunnableTest {
Mockito.when(processService.findProcessInstanceById(222)).thenReturn(processInstance9);
workflowExecuteThread.checkSerialProcess(processDefinition1);
} catch (Exception e) {
Assertions.fail();
Assertions.fail(e);
}
}
@ -314,17 +330,23 @@ public class WorkflowExecuteRunnableTest {
Mockito.when(processInstance.getCommandType()).thenReturn(CommandType.EXECUTE_TASK);
Mockito.when(processInstance.getId()).thenReturn(123);
DAG<String, TaskNode, TaskNodeRelation> dag = Mockito.mock(DAG.class);
Set<String> taskCodesString = new HashSet<>();
taskCodesString.add("1");
taskCodesString.add("2");
DAG<Long, TaskNode, TaskNodeRelation> dag = Mockito.mock(DAG.class);
Set<Long> taskCodesString = new HashSet<>();
taskCodesString.add(1L);
taskCodesString.add(2L);
Mockito.when(dag.getAllNodesList()).thenReturn(taskCodesString);
Mockito.when(dag.containsNode("1")).thenReturn(true);
Mockito.when(dag.containsNode("2")).thenReturn(false);
Mockito.when(dag.containsNode(1L)).thenReturn(true);
Mockito.when(dag.containsNode(2L)).thenReturn(false);
Field dagField = masterExecThreadClass.getDeclaredField("dag");
WorkflowExecuteContext workflowExecuteContext = Mockito.mock(WorkflowExecuteContext.class);
Mockito.when(workflowExecuteContext.getWorkflowInstance()).thenReturn(processInstance);
IWorkflowGraph workflowGraph = Mockito.mock(IWorkflowGraph.class);
Mockito.when(workflowExecuteContext.getWorkflowGraph()).thenReturn(workflowGraph);
Mockito.when(workflowGraph.getDag()).thenReturn(dag);
Field dagField = masterExecThreadClass.getDeclaredField("workflowExecuteContext");
dagField.setAccessible(true);
dagField.set(workflowExecuteThread, dag);
dagField.set(workflowExecuteThread, workflowExecuteContext);
Mockito.when(taskInstanceDao.queryByWorkflowInstanceIdAndTaskCode(processInstance.getId(),
taskInstance1.getTaskCode()))

6
dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/service/FailoverServiceTest.java
View File

@ -36,6 +36,7 @@ import org.apache.dolphinscheduler.registry.api.enums.RegistryNodeType;
import org.apache.dolphinscheduler.server.master.cache.ProcessInstanceExecCacheManager;
import org.apache.dolphinscheduler.server.master.config.MasterConfig;
import org.apache.dolphinscheduler.server.master.event.StateEvent;
import org.apache.dolphinscheduler.server.master.runner.IWorkflowExecuteContext;
import org.apache.dolphinscheduler.server.master.runner.WorkflowExecuteRunnable;
import org.apache.dolphinscheduler.server.master.runner.WorkflowExecuteThreadPool;
import org.apache.dolphinscheduler.service.bean.SpringApplicationContext;
@ -199,7 +200,10 @@ public class FailoverServiceTest {
workerTaskInstance.setState(TaskExecutionStatus.RUNNING_EXECUTION);
WorkflowExecuteRunnable workflowExecuteRunnable = Mockito.mock(WorkflowExecuteRunnable.class);
Mockito.when(workflowExecuteRunnable.getAllTaskInstances()).thenReturn(Lists.newArrayList(workerTaskInstance));
Mockito.when(workflowExecuteRunnable.getProcessInstance()).thenReturn(processInstance);
IWorkflowExecuteContext workflowExecuteRunnableContext = Mockito.mock(IWorkflowExecuteContext.class);
Mockito.when(workflowExecuteRunnable.getWorkflowExecuteContext()).thenReturn(workflowExecuteRunnableContext);
Mockito.when(workflowExecuteRunnableContext.getWorkflowInstance()).thenReturn(processInstance);
Mockito.when(cacheManager.getAll()).thenReturn(Lists.newArrayList(workflowExecuteRunnable));
Mockito.when(cacheManager.getByProcessInstanceId(Mockito.anyInt())).thenReturn(workflowExecuteRunnable);

8
dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/model/TaskNode.java
View File

@ -133,7 +133,7 @@ public class TaskNode {
/**
* node dependency list
*/
private List<String> depList;
private List<Long> depList;
/**
* outer dependency information
@ -242,7 +242,7 @@ public class TaskNode {
public void setPreTasks(String preTasks) {
this.preTasks = preTasks;
this.depList = JSONUtils.toList(preTasks, String.class);
this.depList = JSONUtils.toList(preTasks, Long.class);
}
public String getExtras() {
@ -253,11 +253,11 @@ public class TaskNode {
this.extras = extras;
}
public List<String> getDepList() {
public List<Long> getDepList() {
return depList;
}
public void setDepList(List<String> depList) {
public void setDepList(List<Long> depList) {
if (depList != null) {
this.depList = depList;
this.preTasks = JSONUtils.toJsonString(depList);

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

@ -171,7 +171,7 @@ public interface ProcessService {
boolean isTaskOnline(long taskCode);
DAG<String, TaskNode, TaskNodeRelation> genDagGraph(ProcessDefinition processDefinition);
DAG<Long, TaskNode, TaskNodeRelation> genDagGraph(ProcessDefinition processDefinition);
DagData genDagData(ProcessDefinition processDefinition);

2
dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessServiceImpl.java
View File

@ -2125,7 +2125,7 @@ public class ProcessServiceImpl implements ProcessService {
* @return dag graph
*/
@Override
public DAG<String, TaskNode, TaskNodeRelation> genDagGraph(ProcessDefinition processDefinition) {
public DAG<Long, TaskNode, TaskNodeRelation> genDagGraph(ProcessDefinition processDefinition) {
List<ProcessTaskRelation> taskRelations =
this.findRelationByCode(processDefinition.getCode(), processDefinition.getVersion());
List<TaskNode> taskNodeList = transformTask(taskRelations, Lists.newArrayList());

170
dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/utils/DagHelper.java
View File

@ -61,11 +61,11 @@ public class DagHelper {
List<TaskNodeRelation> nodeRelationList = new ArrayList<>();
for (TaskNode taskNode : taskNodeList) {
String preTasks = taskNode.getPreTasks();
List<String> preTaskList = JSONUtils.toList(preTasks, String.class);
List<Long> preTaskList = JSONUtils.toList(preTasks, Long.class);
if (preTaskList != null) {
for (String depNodeCode : preTaskList) {
for (Long depNodeCode : preTaskList) {
if (null != findNodeByCode(taskNodeList, depNodeCode)) {
nodeRelationList.add(new TaskNodeRelation(depNodeCode, Long.toString(taskNode.getCode())));
nodeRelationList.add(new TaskNodeRelation(depNodeCode, taskNode.getCode()));
}
}
}
@ -83,11 +83,11 @@ public class DagHelper {
* @return task node list
*/
public static List<TaskNode> generateFlowNodeListByStartNode(List<TaskNode> taskNodeList,
List<String> startNodeNameList,
List<String> recoveryNodeCodeList,
List<Long> startNodeNameList,
List<Long> recoveryNodeCodeList,
TaskDependType taskDependType) {
List<TaskNode> destFlowNodeList = new ArrayList<>();
List<String> startNodeList = startNodeNameList;
List<Long> startNodeList = startNodeNameList;
if (taskDependType != TaskDependType.TASK_POST && CollectionUtils.isEmpty(startNodeList)) {
log.error("start node list is empty! cannot continue run the process ");
@ -106,7 +106,7 @@ public class DagHelper {
tmpTaskNodeList = taskNodeList;
} else {
// specified start nodes or resume execution
for (String startNodeCode : startNodeList) {
for (Long startNodeCode : startNodeList) {
TaskNode startNode = findNodeByCode(taskNodeList, startNodeCode);
List<TaskNode> childNodeList = new ArrayList<>();
if (startNode == null) {
@ -115,10 +115,10 @@ public class DagHelper {
taskNodeList);
continue;
} else if (TaskDependType.TASK_POST == taskDependType) {
List<String> visitedNodeCodeList = new ArrayList<>();
List<Long> visitedNodeCodeList = new ArrayList<>();
childNodeList = getFlowNodeListPost(startNode, taskNodeList, visitedNodeCodeList);
} else if (TaskDependType.TASK_PRE == taskDependType) {
List<String> visitedNodeCodeList = new ArrayList<>();
List<Long> visitedNodeCodeList = new ArrayList<>();
childNodeList =
getFlowNodeListPre(startNode, recoveryNodeCodeList, taskNodeList, visitedNodeCodeList);
} else {
@ -129,7 +129,7 @@ public class DagHelper {
}
for (TaskNode taskNode : tmpTaskNodeList) {
if (null == findNodeByCode(destTaskNodeList, Long.toString(taskNode.getCode()))) {
if (null == findNodeByCode(destTaskNodeList, taskNode.getCode())) {
destTaskNodeList.add(taskNode);
}
}
@ -143,19 +143,20 @@ public class DagHelper {
* @param taskNodeList taskNodeList
* @return task node list
*/
private static List<TaskNode> getFlowNodeListPost(TaskNode startNode, List<TaskNode> taskNodeList,
List<String> visitedNodeCodeList) {
private static List<TaskNode> getFlowNodeListPost(TaskNode startNode,
List<TaskNode> taskNodeList,
List<Long> visitedNodeCodeList) {
List<TaskNode> resultList = new ArrayList<>();
for (TaskNode taskNode : taskNodeList) {
List<String> depList = taskNode.getDepList();
if (null != depList && null != startNode && depList.contains(Long.toString(startNode.getCode()))
&& !visitedNodeCodeList.contains(Long.toString(taskNode.getCode()))) {
List<Long> depList = taskNode.getDepList();
if (null != depList && null != startNode && depList.contains(startNode.getCode())
&& !visitedNodeCodeList.contains(taskNode.getCode())) {
resultList.addAll(getFlowNodeListPost(taskNode, taskNodeList, visitedNodeCodeList));
}
}
// why add (startNode != null) condition? for SonarCloud Quality Gate passed
if (null != startNode) {
visitedNodeCodeList.add(Long.toString(startNode.getCode()));
visitedNodeCodeList.add(startNode.getCode());
}
resultList.add(startNode);
@ -170,12 +171,14 @@ public class DagHelper {
* @param taskNodeList taskNodeList
* @return task node list
*/
private static List<TaskNode> getFlowNodeListPre(TaskNode startNode, List<String> recoveryNodeCodeList,
List<TaskNode> taskNodeList, List<String> visitedNodeCodeList) {
private static List<TaskNode> getFlowNodeListPre(TaskNode startNode,
List<Long> recoveryNodeCodeList,
List<TaskNode> taskNodeList,
List<Long> visitedNodeCodeList) {
List<TaskNode> resultList = new ArrayList<>();
List<String> depList = new ArrayList<>();
List<Long> depList = new ArrayList<>();
if (null != startNode) {
depList = startNode.getDepList();
resultList.add(startNode);
@ -183,7 +186,7 @@ public class DagHelper {
if (CollectionUtils.isEmpty(depList)) {
return resultList;
}
for (String depNodeCode : depList) {
for (Long depNodeCode : depList) {
TaskNode start = findNodeByCode(taskNodeList, depNodeCode);
if (recoveryNodeCodeList.contains(depNodeCode)) {
resultList.add(start);
@ -193,7 +196,7 @@ public class DagHelper {
}
// why add (startNode != null) condition? for SonarCloud Quality Gate passed
if (null != startNode) {
visitedNodeCodeList.add(Long.toString(startNode.getCode()));
visitedNodeCodeList.add(startNode.getCode());
}
return resultList;
}
@ -209,8 +212,8 @@ public class DagHelper {
* @throws Exception if error throws Exception
*/
public static ProcessDag generateFlowDag(List<TaskNode> totalTaskNodeList,
List<String> startNodeNameList,
List<String> recoveryNodeCodeList,
List<Long> startNodeNameList,
List<Long> recoveryNodeCodeList,
TaskDependType depNodeType) throws Exception {
List<TaskNode> destTaskNodeList = generateFlowNodeListByStartNode(totalTaskNodeList, startNodeNameList,
@ -248,9 +251,9 @@ public class DagHelper {
* @param nodeCode nodeCode
* @return task node
*/
public static TaskNode findNodeByCode(List<TaskNode> nodeDetails, String nodeCode) {
public static TaskNode findNodeByCode(List<TaskNode> nodeDetails, Long nodeCode) {
for (TaskNode taskNode : nodeDetails) {
if (Long.toString(taskNode.getCode()).equals(nodeCode)) {
if (taskNode.getCode() == nodeCode) {
return taskNode;
}
}
@ -266,14 +269,14 @@ public class DagHelper {
* @return can submit
*/
public static boolean allDependsForbiddenOrEnd(TaskNode taskNode,
DAG<String, TaskNode, TaskNodeRelation> dag,
Map<String, TaskNode> skipTaskNodeList,
Map<String, TaskInstance> completeTaskList) {
List<String> dependList = taskNode.getDepList();
DAG<Long, TaskNode, TaskNodeRelation> dag,
Map<Long, TaskNode> skipTaskNodeList,
Map<Long, TaskInstance> completeTaskList) {
List<Long> dependList = taskNode.getDepList();
if (dependList == null) {
return true;
}
for (String dependNodeCode : dependList) {
for (Long dependNodeCode : dependList) {
TaskNode dependNode = dag.getNode(dependNodeCode);
if (dependNode == null || completeTaskList.containsKey(dependNodeCode)
|| dependNode.isForbidden()
@ -293,25 +296,25 @@ public class DagHelper {
*
* @return successor nodes
*/
public static Set<String> parsePostNodes(String preNodeCode,
Map<String, TaskNode> skipTaskNodeList,
DAG<String, TaskNode, TaskNodeRelation> dag,
Map<String, TaskInstance> completeTaskList) {
Set<String> postNodeList = new HashSet<>();
Collection<String> startVertexes = new ArrayList<>();
public static Set<Long> parsePostNodes(Long preNodeCode,
Map<Long, TaskNode> skipTaskNodeList,
DAG<Long, TaskNode, TaskNodeRelation> dag,
Map<Long, TaskInstance> completeTaskList) {
Set<Long> postNodeList = new HashSet<>();
Collection<Long> startVertexes = new ArrayList<>();
if (preNodeCode == null) {
startVertexes = dag.getBeginNode();
} else if (dag.getNode(preNodeCode).isConditionsTask()) {
List<String> conditionTaskList = parseConditionTask(preNodeCode, skipTaskNodeList, dag, completeTaskList);
List<Long> conditionTaskList = parseConditionTask(preNodeCode, skipTaskNodeList, dag, completeTaskList);
startVertexes.addAll(conditionTaskList);
} else if (dag.getNode(preNodeCode).isSwitchTask()) {
List<String> conditionTaskList = parseSwitchTask(preNodeCode, skipTaskNodeList, dag, completeTaskList);
List<Long> conditionTaskList = parseSwitchTask(preNodeCode, skipTaskNodeList, dag, completeTaskList);
startVertexes.addAll(conditionTaskList);
} else {
startVertexes = dag.getSubsequentNodes(preNodeCode);
}
for (String subsequent : startVertexes) {
for (Long subsequent : startVertexes) {
TaskNode taskNode = dag.getNode(subsequent);
if (taskNode == null) {
log.error("taskNode {} is null, please check dag", subsequent);
@ -337,11 +340,11 @@ public class DagHelper {
* if all of the task dependence are skipped, skip it too.
*/
private static boolean isTaskNodeNeedSkip(TaskNode taskNode,
Map<String, TaskNode> skipTaskNodeList) {
Map<Long, TaskNode> skipTaskNodeList) {
if (CollectionUtils.isEmpty(taskNode.getDepList())) {
return false;
}
for (String depNode : taskNode.getDepList()) {
for (Long depNode : taskNode.getDepList()) {
if (!skipTaskNodeList.containsKey(depNode)) {
return false;
}
@ -353,11 +356,11 @@ public class DagHelper {
* parse condition task find the branch process
* set skip flag for another one.
*/
public static List<String> parseConditionTask(String nodeCode,
Map<String, TaskNode> skipTaskNodeList,
DAG<String, TaskNode, TaskNodeRelation> dag,
Map<String, TaskInstance> completeTaskList) {
List<String> conditionTaskList = new ArrayList<>();
public static List<Long> parseConditionTask(Long nodeCode,
Map<Long, TaskNode> skipTaskNodeList,
DAG<Long, TaskNode, TaskNodeRelation> dag,
Map<Long, TaskInstance> completeTaskList) {
List<Long> conditionTaskList = new ArrayList<>();
TaskNode taskNode = dag.getNode(nodeCode);
if (!taskNode.isConditionsTask()) {
return conditionTaskList;
@ -368,7 +371,7 @@ public class DagHelper {
TaskInstance taskInstance = completeTaskList.get(nodeCode);
ConditionsParameters conditionsParameters =
JSONUtils.parseObject(taskNode.getConditionResult(), ConditionsParameters.class);
List<String> skipNodeList = new ArrayList<>();
List<Long> skipNodeList = new ArrayList<>();
if (taskInstance.getState().isSuccess()) {
conditionTaskList = conditionsParameters.getSuccessNode();
skipNodeList = conditionsParameters.getFailedNode();
@ -380,7 +383,7 @@ public class DagHelper {
}
// the skipNodeList maybe null if no next task
skipNodeList = Optional.ofNullable(skipNodeList).orElse(new ArrayList<>());
for (String failedNode : skipNodeList) {
for (Long failedNode : skipNodeList) {
setTaskNodeSkip(failedNode, dag, completeTaskList, skipTaskNodeList);
}
// the conditionTaskList maybe null if no next task
@ -395,11 +398,11 @@ public class DagHelper {
* @param nodeCode
* @return
*/
public static List<String> parseSwitchTask(String nodeCode,
Map<String, TaskNode> skipTaskNodeList,
DAG<String, TaskNode, TaskNodeRelation> dag,
Map<String, TaskInstance> completeTaskList) {
List<String> conditionTaskList = new ArrayList<>();
public static List<Long> parseSwitchTask(Long nodeCode,
Map<Long, TaskNode> skipTaskNodeList,
DAG<Long, TaskNode, TaskNodeRelation> dag,
Map<Long, TaskInstance> completeTaskList) {
List<Long> conditionTaskList = new ArrayList<>();
TaskNode taskNode = dag.getNode(nodeCode);
if (!taskNode.isSwitchTask()) {
return conditionTaskList;
@ -411,15 +414,16 @@ public class DagHelper {
return conditionTaskList;
}
private static List<String> skipTaskNode4Switch(TaskNode taskNode, Map<String, TaskNode> skipTaskNodeList,
Map<String, TaskInstance> completeTaskList,
DAG<String, TaskNode, TaskNodeRelation> dag) {
private static List<Long> skipTaskNode4Switch(TaskNode taskNode,
Map<Long, TaskNode> skipTaskNodeList,
Map<Long, TaskInstance> completeTaskList,
DAG<Long, TaskNode, TaskNodeRelation> dag) {
SwitchParameters switchParameters =
completeTaskList.get(Long.toString(taskNode.getCode())).getSwitchDependency();
completeTaskList.get(taskNode.getCode()).getSwitchDependency();
int resultConditionLocation = switchParameters.getResultConditionLocation();
List<SwitchResultVo> conditionResultVoList = switchParameters.getDependTaskList();
List<String> switchTaskList = conditionResultVoList.get(resultConditionLocation).getNextNode();
List<Long> switchTaskList = conditionResultVoList.get(resultConditionLocation).getNextNode();
if (CollectionUtils.isEmpty(switchTaskList)) {
switchTaskList = new ArrayList<>();
}
@ -436,16 +440,16 @@ public class DagHelper {
/**
* set task node and the post nodes skip flag
*/
private static void setTaskNodeSkip(String skipNodeCode,
DAG<String, TaskNode, TaskNodeRelation> dag,
Map<String, TaskInstance> completeTaskList,
Map<String, TaskNode> skipTaskNodeList) {
private static void setTaskNodeSkip(Long skipNodeCode,
DAG<Long, TaskNode, TaskNodeRelation> dag,
Map<Long, TaskInstance> completeTaskList,
Map<Long, TaskNode> skipTaskNodeList) {
if (!dag.containsNode(skipNodeCode)) {
return;
}
skipTaskNodeList.putIfAbsent(skipNodeCode, dag.getNode(skipNodeCode));
Collection<String> postNodeList = dag.getSubsequentNodes(skipNodeCode);
for (String post : postNodeList) {
Collection<Long> postNodeList = dag.getSubsequentNodes(skipNodeCode);
for (Long post : postNodeList) {
TaskNode postNode = dag.getNode(post);
if (isTaskNodeNeedSkip(postNode, skipTaskNodeList)) {
setTaskNodeSkip(post, dag, completeTaskList, skipTaskNodeList);
@ -458,14 +462,14 @@ public class DagHelper {
* @param processDag processDag
* @return dag
*/
public static DAG<String, TaskNode, TaskNodeRelation> buildDagGraph(ProcessDag processDag) {
public static DAG<Long, TaskNode, TaskNodeRelation> buildDagGraph(ProcessDag processDag) {
DAG<String, TaskNode, TaskNodeRelation> dag = new DAG<>();
DAG<Long, TaskNode, TaskNodeRelation> dag = new DAG<>();
// add vertex
if (CollectionUtils.isNotEmpty(processDag.getNodes())) {
for (TaskNode node : processDag.getNodes()) {
dag.addNode(Long.toString(node.getCode()), node);
dag.addNode(node.getCode(), node);
}
}
@ -490,12 +494,12 @@ public class DagHelper {
// Traverse node information and build relationships
for (TaskNode taskNode : taskNodeList) {
String preTasks = taskNode.getPreTasks();
List<String> preTasksList = JSONUtils.toList(preTasks, String.class);
List<Long> preTasksList = JSONUtils.toList(preTasks, Long.class);
// If the dependency is not empty
if (preTasksList != null) {
for (String depNode : preTasksList) {
taskNodeRelations.add(new TaskNodeRelation(depNode, Long.toString(taskNode.getCode())));
for (Long depNode : preTasksList) {
taskNodeRelations.add(new TaskNodeRelation(depNode, taskNode.getCode()));
}
}
}
@ -530,7 +534,7 @@ public class DagHelper {
TaskNode preNode = taskNodeMap.get(preTaskCode);
TaskNode postNode = taskNodeMap.get(postTaskCode);
taskNodeRelations
.add(new TaskNodeRelation(Long.toString(preNode.getCode()), Long.toString(postNode.getCode())));
.add(new TaskNodeRelation(preNode.getCode(), postNode.getCode()));
}
}
ProcessDag processDag = new ProcessDag();
@ -542,20 +546,20 @@ public class DagHelper {
/**
* is there have conditions after the parent node
*/
public static boolean haveConditionsAfterNode(String parentNodeCode,
DAG<String, TaskNode, TaskNodeRelation> dag) {
public static boolean haveConditionsAfterNode(Long parentNodeCode,
DAG<Long, TaskNode, TaskNodeRelation> dag) {
return haveSubAfterNode(parentNodeCode, dag, TaskConstants.TASK_TYPE_CONDITIONS);
}
/**
* is there have conditions after the parent node
*/
public static boolean haveConditionsAfterNode(String parentNodeCode, List<TaskNode> taskNodes) {
public static boolean haveConditionsAfterNode(Long parentNodeCode, List<TaskNode> taskNodes) {
if (CollectionUtils.isEmpty(taskNodes)) {
return false;
}
for (TaskNode taskNode : taskNodes) {
List<String> preTasksList = JSONUtils.toList(taskNode.getPreTasks(), String.class);
List<Long> preTasksList = JSONUtils.toList(taskNode.getPreTasks(), Long.class);
if (preTasksList.contains(parentNodeCode) && taskNode.isConditionsTask()) {
return true;
}
@ -566,32 +570,32 @@ public class DagHelper {
/**
* is there have blocking node after the parent node
*/
public static boolean haveBlockingAfterNode(String parentNodeCode,
DAG<String, TaskNode, TaskNodeRelation> dag) {
public static boolean haveBlockingAfterNode(Long parentNodeCode,
DAG<Long, TaskNode, TaskNodeRelation> dag) {
return haveSubAfterNode(parentNodeCode, dag, TaskConstants.TASK_TYPE_BLOCKING);
}
/**
* is there have all node after the parent node
*/
public static boolean haveAllNodeAfterNode(String parentNodeCode,
DAG<String, TaskNode, TaskNodeRelation> dag) {
public static boolean haveAllNodeAfterNode(Long parentNodeCode,
DAG<Long, TaskNode, TaskNodeRelation> dag) {
return haveSubAfterNode(parentNodeCode, dag, null);
}
/**
* Whether there is a specified type of child node after the parent node
*/
public static boolean haveSubAfterNode(String parentNodeCode,
DAG<String, TaskNode, TaskNodeRelation> dag, String filterNodeType) {
Set<String> subsequentNodes = dag.getSubsequentNodes(parentNodeCode);
public static boolean haveSubAfterNode(Long parentNodeCode,
DAG<Long, TaskNode, TaskNodeRelation> dag, String filterNodeType) {
Set<Long> subsequentNodes = dag.getSubsequentNodes(parentNodeCode);
if (CollectionUtils.isEmpty(subsequentNodes)) {
return false;
}
if (StringUtils.isBlank(filterNodeType)) {
return true;
}
for (String nodeName : subsequentNodes) {
for (Long nodeName : subsequentNodes) {
TaskNode taskNode = dag.getNode(nodeName);
if (taskNode.getType().equalsIgnoreCase(filterNodeType)) {
return true;

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

@ -704,7 +704,7 @@ public class ProcessServiceTest {
Mockito.when(processTaskRelationLogMapper.queryByProcessCodeAndVersion(Mockito.anyLong(), Mockito.anyInt()))
.thenReturn(list);
DAG<String, TaskNode, TaskNodeRelation> stringTaskNodeTaskNodeRelationDAG =
DAG<Long, TaskNode, TaskNodeRelation> stringTaskNodeTaskNodeRelationDAG =
processService.genDagGraph(processDefinition);
Assertions.assertEquals(1, stringTaskNodeTaskNodeRelationDAG.getNodesCount());
}

184
dolphinscheduler-service/src/test/java/org/apache/dolphinscheduler/service/utils/DagHelperTest.java
View File

@ -51,16 +51,16 @@ public class DagHelperTest {
@Test
public void testHaveSubAfterNode() {
String parentNodeCode = "5293789969856";
Long parentNodeCode = 5293789969856L;
List<TaskNodeRelation> taskNodeRelations = new ArrayList<>();
TaskNodeRelation relation = new TaskNodeRelation();
relation.setStartNode("5293789969856");
relation.setEndNode("5293789969857");
relation.setStartNode(5293789969856L);
relation.setEndNode(5293789969857L);
taskNodeRelations.add(relation);
TaskNodeRelation relationNext = new TaskNodeRelation();
relationNext.setStartNode("5293789969856");
relationNext.setEndNode("5293789969858");
relationNext.setStartNode(5293789969856L);
relationNext.setEndNode(5293789969858L);
taskNodeRelations.add(relationNext);
List<TaskNode> taskNodes = new ArrayList<>();
@ -85,7 +85,7 @@ public class DagHelperTest {
ProcessDag processDag = new ProcessDag();
processDag.setEdges(taskNodeRelations);
processDag.setNodes(taskNodes);
DAG<String, TaskNode, TaskNodeRelation> dag = DagHelper.buildDagGraph(processDag);
DAG<Long, TaskNode, TaskNodeRelation> dag = DagHelper.buildDagGraph(processDag);
boolean canSubmit = DagHelper.haveAllNodeAfterNode(parentNodeCode, dag);
Assertions.assertTrue(canSubmit);
@ -108,34 +108,34 @@ public class DagHelperTest {
public void testTaskNodeCanSubmit() throws IOException {
// 1->2->3->5->7
// 4->3->6
DAG<String, TaskNode, TaskNodeRelation> dag = generateDag();
TaskNode taskNode3 = dag.getNode("3");
Map<String, TaskInstance> completeTaskList = new HashMap<>();
Map<String, TaskNode> skipNodeList = new HashMap<>();
completeTaskList.putIfAbsent("1", new TaskInstance());
DAG<Long, TaskNode, TaskNodeRelation> dag = generateDag();
TaskNode taskNode3 = dag.getNode(3L);
Map<Long, TaskInstance> completeTaskList = new HashMap<>();
Map<Long, TaskNode> skipNodeList = new HashMap<>();
completeTaskList.putIfAbsent(1L, new TaskInstance());
Boolean canSubmit = false;
// 2/4 are forbidden submit 3
TaskNode node2 = dag.getNode("2");
TaskNode node2 = dag.getNode(2L);
node2.setRunFlag(Constants.FLOWNODE_RUN_FLAG_FORBIDDEN);
TaskNode nodex = dag.getNode("4");
TaskNode nodex = dag.getNode(4L);
nodex.setRunFlag(Constants.FLOWNODE_RUN_FLAG_FORBIDDEN);
canSubmit = DagHelper.allDependsForbiddenOrEnd(taskNode3, dag, skipNodeList, completeTaskList);
Assertions.assertEquals(canSubmit, true);
// 2forbidden, 3 cannot be submit
completeTaskList.putIfAbsent("2", new TaskInstance());
TaskNode nodey = dag.getNode("4");
completeTaskList.putIfAbsent(2L, new TaskInstance());
TaskNode nodey = dag.getNode(4L);
nodey.setRunFlag("");
canSubmit = DagHelper.allDependsForbiddenOrEnd(taskNode3, dag, skipNodeList, completeTaskList);
Assertions.assertEquals(canSubmit, false);
// 2/3 forbidden submit 5
TaskNode node3 = dag.getNode("3");
TaskNode node3 = dag.getNode(3L);
node3.setRunFlag(Constants.FLOWNODE_RUN_FLAG_FORBIDDEN);
TaskNode node8 = dag.getNode("8");
TaskNode node8 = dag.getNode(8L);
node8.setRunFlag(Constants.FLOWNODE_RUN_FLAG_FORBIDDEN);
TaskNode node5 = dag.getNode("5");
TaskNode node5 = dag.getNode(5L);
canSubmit = DagHelper.allDependsForbiddenOrEnd(node5, dag, skipNodeList, completeTaskList);
Assertions.assertEquals(canSubmit, true);
}
@ -145,64 +145,64 @@ public class DagHelperTest {
*/
@Test
public void testParsePostNodeList() throws IOException {
DAG<String, TaskNode, TaskNodeRelation> dag = generateDag();
Map<String, TaskInstance> completeTaskList = new HashMap<>();
Map<String, TaskNode> skipNodeList = new HashMap<>();
DAG<Long, TaskNode, TaskNodeRelation> dag = generateDag();
Map<Long, TaskInstance> completeTaskList = new HashMap<>();
Map<Long, TaskNode> skipNodeList = new HashMap<>();
Set<String> postNodes = null;
Set<Long> postNodes = null;
// complete : null
// expect post: 1/4
postNodes = DagHelper.parsePostNodes(null, skipNodeList, dag, completeTaskList);
Assertions.assertEquals(2, postNodes.size());
Assertions.assertTrue(postNodes.contains("1"));
Assertions.assertTrue(postNodes.contains("4"));
Assertions.assertTrue(postNodes.contains(1L));
Assertions.assertTrue(postNodes.contains(4L));
// complete : 1
// expect post: 2/4
completeTaskList.put("1", new TaskInstance());
completeTaskList.put(1L, new TaskInstance());
postNodes = DagHelper.parsePostNodes(null, skipNodeList, dag, completeTaskList);
Assertions.assertEquals(2, postNodes.size());
Assertions.assertTrue(postNodes.contains("2"));
Assertions.assertTrue(postNodes.contains("4"));
Assertions.assertTrue(postNodes.contains(2L));
Assertions.assertTrue(postNodes.contains(4L));
// complete : 1/2
// expect post: 4
completeTaskList.put("2", new TaskInstance());
completeTaskList.put(2L, new TaskInstance());
postNodes = DagHelper.parsePostNodes(null, skipNodeList, dag, completeTaskList);
Assertions.assertEquals(2, postNodes.size());
Assertions.assertTrue(postNodes.contains("4"));
Assertions.assertTrue(postNodes.contains("8"));
Assertions.assertTrue(postNodes.contains(4L));
Assertions.assertTrue(postNodes.contains(8L));
// complete : 1/2/4
// expect post: 3
completeTaskList.put("4", new TaskInstance());
completeTaskList.put(4L, new TaskInstance());
postNodes = DagHelper.parsePostNodes(null, skipNodeList, dag, completeTaskList);
Assertions.assertEquals(2, postNodes.size());
Assertions.assertTrue(postNodes.contains("3"));
Assertions.assertTrue(postNodes.contains("8"));
Assertions.assertTrue(postNodes.contains(3L));
Assertions.assertTrue(postNodes.contains(8L));
// complete : 1/2/4/3
// expect post: 8/6
completeTaskList.put("3", new TaskInstance());
completeTaskList.put(3L, new TaskInstance());
postNodes = DagHelper.parsePostNodes(null, skipNodeList, dag, completeTaskList);
Assertions.assertEquals(2, postNodes.size());
Assertions.assertTrue(postNodes.contains("8"));
Assertions.assertTrue(postNodes.contains("6"));
Assertions.assertTrue(postNodes.contains(8L));
Assertions.assertTrue(postNodes.contains(6L));
// complete : 1/2/4/3/8
// expect post: 6/5
completeTaskList.put("8", new TaskInstance());
completeTaskList.put(8L, new TaskInstance());
postNodes = DagHelper.parsePostNodes(null, skipNodeList, dag, completeTaskList);
Assertions.assertEquals(2, postNodes.size());
Assertions.assertTrue(postNodes.contains("5"));
Assertions.assertTrue(postNodes.contains("6"));
Assertions.assertTrue(postNodes.contains(5L));
Assertions.assertTrue(postNodes.contains(6L));
// complete : 1/2/4/3/5/6/8
// expect post: 7
completeTaskList.put("6", new TaskInstance());
completeTaskList.put("5", new TaskInstance());
completeTaskList.put(6L, new TaskInstance());
completeTaskList.put(5L, new TaskInstance());
postNodes = DagHelper.parsePostNodes(null, skipNodeList, dag, completeTaskList);
Assertions.assertEquals(1, postNodes.size());
Assertions.assertTrue(postNodes.contains("7"));
Assertions.assertTrue(postNodes.contains(7L));
}
/**
@ -212,35 +212,35 @@ public class DagHelperTest {
*/
@Test
public void testForbiddenPostNode() throws IOException {
DAG<String, TaskNode, TaskNodeRelation> dag = generateDag();
Map<String, TaskInstance> completeTaskList = new HashMap<>();
Map<String, TaskNode> skipNodeList = new HashMap<>();
Set<String> postNodes = null;
DAG<Long, TaskNode, TaskNodeRelation> dag = generateDag();
Map<Long, TaskInstance> completeTaskList = new HashMap<>();
Map<Long, TaskNode> skipNodeList = new HashMap<>();
Set<Long> postNodes = null;
// dag: 1-2-3-5-7 4-3-6 2-8-5-7
// forbid:2 complete:1 post:4/8
completeTaskList.put("1", new TaskInstance());
TaskNode node2 = dag.getNode("2");
completeTaskList.put(1L, new TaskInstance());
TaskNode node2 = dag.getNode(2L);
node2.setRunFlag(Constants.FLOWNODE_RUN_FLAG_FORBIDDEN);
postNodes = DagHelper.parsePostNodes(null, skipNodeList, dag, completeTaskList);
Assertions.assertEquals(2, postNodes.size());
Assertions.assertTrue(postNodes.contains("4"));
Assertions.assertTrue(postNodes.contains("8"));
Assertions.assertTrue(postNodes.contains(4L));
Assertions.assertTrue(postNodes.contains(8L));
// forbid:2/4 complete:1 post:3/8
TaskNode node4 = dag.getNode("4");
TaskNode node4 = dag.getNode(4L);
node4.setRunFlag(Constants.FLOWNODE_RUN_FLAG_FORBIDDEN);
postNodes = DagHelper.parsePostNodes(null, skipNodeList, dag, completeTaskList);
Assertions.assertEquals(2, postNodes.size());
Assertions.assertTrue(postNodes.contains("3"));
Assertions.assertTrue(postNodes.contains("8"));
Assertions.assertTrue(postNodes.contains(3L));
Assertions.assertTrue(postNodes.contains(8L));
// forbid:2/4/5 complete:1/8 post:3
completeTaskList.put("8", new TaskInstance());
TaskNode node5 = dag.getNode("5");
completeTaskList.put(8L, new TaskInstance());
TaskNode node5 = dag.getNode(5L);
node5.setRunFlag(Constants.FLOWNODE_RUN_FLAG_FORBIDDEN);
postNodes = DagHelper.parsePostNodes(null, skipNodeList, dag, completeTaskList);
Assertions.assertEquals(1, postNodes.size());
Assertions.assertTrue(postNodes.contains("3"));
Assertions.assertTrue(postNodes.contains(3L));
}
/**
@ -250,16 +250,16 @@ public class DagHelperTest {
*/
@Test
public void testConditionPostNode() throws IOException {
DAG<String, TaskNode, TaskNodeRelation> dag = generateDag();
Map<String, TaskInstance> completeTaskList = new HashMap<>();
Map<String, TaskNode> skipNodeList = new HashMap<>();
Set<String> postNodes = null;
DAG<Long, TaskNode, TaskNodeRelation> dag = generateDag();
Map<Long, TaskInstance> completeTaskList = new HashMap<>();
Map<Long, TaskNode> skipNodeList = new HashMap<>();
Set<Long> postNodes = null;
// dag: 1-2-3-5-7 4-3-6 2-8-5-7
// 3-if
completeTaskList.put("1", new TaskInstance());
completeTaskList.put("2", new TaskInstance());
completeTaskList.put("4", new TaskInstance());
TaskNode node3 = dag.getNode("3");
completeTaskList.put(1L, new TaskInstance());
completeTaskList.put(2L, new TaskInstance());
completeTaskList.put(4L, new TaskInstance());
TaskNode node3 = dag.getNode(3L);
node3.setType(TASK_TYPE_CONDITIONS);
node3.setConditionResult("{\n"
+
@ -272,51 +272,51 @@ public class DagHelperTest {
" ]\n"
+
" }");
completeTaskList.remove("3");
completeTaskList.remove(3L);
TaskInstance taskInstance = new TaskInstance();
taskInstance.setState(TaskExecutionStatus.SUCCESS);
// complete 1/2/3/4 expect:8
completeTaskList.put("3", taskInstance);
completeTaskList.put(3L, taskInstance);
postNodes = DagHelper.parsePostNodes(null, skipNodeList, dag, completeTaskList);
Assertions.assertEquals(1, postNodes.size());
Assertions.assertTrue(postNodes.contains("8"));
Assertions.assertTrue(postNodes.contains(8L));
// 2.complete 1/2/3/4/8 expect:5 skip:6
completeTaskList.put("8", new TaskInstance());
completeTaskList.put(8L, new TaskInstance());
postNodes = DagHelper.parsePostNodes(null, skipNodeList, dag, completeTaskList);
Assertions.assertTrue(postNodes.contains("5"));
Assertions.assertTrue(postNodes.contains(5L));
Assertions.assertEquals(1, skipNodeList.size());
Assertions.assertTrue(skipNodeList.containsKey("6"));
Assertions.assertTrue(skipNodeList.containsKey(6L));
// 3.complete 1/2/3/4/5/8 expect post:7 skip:6
skipNodeList.clear();
TaskInstance taskInstance1 = new TaskInstance();
taskInstance.setState(TaskExecutionStatus.SUCCESS);
completeTaskList.put("5", taskInstance1);
completeTaskList.put(5L, taskInstance1);
postNodes = DagHelper.parsePostNodes(null, skipNodeList, dag, completeTaskList);
Assertions.assertEquals(1, postNodes.size());
Assertions.assertTrue(postNodes.contains("7"));
Assertions.assertTrue(postNodes.contains(7L));
Assertions.assertEquals(1, skipNodeList.size());
Assertions.assertTrue(skipNodeList.containsKey("6"));
Assertions.assertTrue(skipNodeList.containsKey(6L));
// dag: 1-2-3-5-7 4-3-6
// 3-if , complete:1/2/3/4
// 1.failure:3 expect post:6 skip:5/7
skipNodeList.clear();
completeTaskList.remove("3");
completeTaskList.remove(3L);
taskInstance = new TaskInstance();
Map<String, Object> taskParamsMap = new HashMap<>();
taskParamsMap.put(Constants.SWITCH_RESULT, "");
taskInstance.setTaskParams(JSONUtils.toJsonString(taskParamsMap));
taskInstance.setState(TaskExecutionStatus.FAILURE);
completeTaskList.put("3", taskInstance);
completeTaskList.put(3L, taskInstance);
postNodes = DagHelper.parsePostNodes(null, skipNodeList, dag, completeTaskList);
Assertions.assertEquals(1, postNodes.size());
Assertions.assertTrue(postNodes.contains("6"));
Assertions.assertTrue(postNodes.contains(6L));
Assertions.assertEquals(2, skipNodeList.size());
Assertions.assertTrue(skipNodeList.containsKey("5"));
Assertions.assertTrue(skipNodeList.containsKey("7"));
Assertions.assertTrue(skipNodeList.containsKey(5L));
Assertions.assertTrue(skipNodeList.containsKey(7L));
// dag: 1-2-3-5-7 4-3-6
// 3-if , complete:1/2/3/4
@ -325,8 +325,8 @@ public class DagHelperTest {
skipNodeList.clear();
completeTaskList.clear();
taskInstance.setSwitchDependency(getSwitchNode());
completeTaskList.put("1", taskInstance);
postNodes = DagHelper.parsePostNodes("1", skipNodeList, dag, completeTaskList);
completeTaskList.put(1L, taskInstance);
postNodes = DagHelper.parsePostNodes(1L, skipNodeList, dag, completeTaskList);
Assertions.assertEquals(1, postNodes.size());
}
@ -345,7 +345,7 @@ public class DagHelperTest {
* @return dag
* @throws JsonProcessingException if error throws JsonProcessingException
*/
private DAG<String, TaskNode, TaskNodeRelation> generateDag() throws IOException {
private DAG<Long, TaskNode, TaskNodeRelation> generateDag() throws IOException {
List<TaskNode> taskNodeList = new ArrayList<>();
TaskNode node1 = new TaskNode();
node1.setId("1");
@ -423,8 +423,8 @@ public class DagHelperTest {
node8.setPreTasks(JSONUtils.toJsonString(dep8));
taskNodeList.add(node8);
List<String> startNodes = new ArrayList<>();
List<String> recoveryNodes = new ArrayList<>();
List<Long> startNodes = new ArrayList<>();
List<Long> recoveryNodes = new ArrayList<>();
List<TaskNode> destTaskNodeList = DagHelper.generateFlowNodeListByStartNode(taskNodeList,
startNodes, recoveryNodes, TaskDependType.TASK_POST);
List<TaskNodeRelation> taskNodeRelations = DagHelper.generateRelationListByFlowNodes(destTaskNodeList);
@ -445,7 +445,7 @@ public class DagHelperTest {
* @return dag
* @throws JsonProcessingException if error throws JsonProcessingException
*/
private DAG<String, TaskNode, TaskNodeRelation> generateDag2() throws IOException {
private DAG<Long, TaskNode, TaskNodeRelation> generateDag2() throws IOException {
List<TaskNode> taskNodeList = new ArrayList<>();
TaskNode node = new TaskNode();
@ -488,13 +488,13 @@ public class DagHelperTest {
node5.setName("4");
node5.setCode(4);
node5.setType("SHELL");
List<String> dep5 = new ArrayList<>();
dep5.add("1");
List<Long> dep5 = new ArrayList<>();
dep5.add(1L);
node5.setPreTasks(JSONUtils.toJsonString(dep5));
taskNodeList.add(node5);
List<String> startNodes = new ArrayList<>();
List<String> recoveryNodes = new ArrayList<>();
List<Long> startNodes = new ArrayList<>();
List<Long> recoveryNodes = new ArrayList<>();
List<TaskNode> destTaskNodeList = DagHelper.generateFlowNodeListByStartNode(taskNodeList,
startNodes, recoveryNodes, TaskDependType.TASK_POST);
List<TaskNodeRelation> taskNodeRelations = DagHelper.generateRelationListByFlowNodes(destTaskNodeList);
@ -508,15 +508,15 @@ public class DagHelperTest {
SwitchParameters conditionsParameters = new SwitchParameters();
SwitchResultVo switchResultVo1 = new SwitchResultVo();
switchResultVo1.setCondition(" 2 == 1");
switchResultVo1.setNextNode("2");
switchResultVo1.setNextNode(2L);
SwitchResultVo switchResultVo2 = new SwitchResultVo();
switchResultVo2.setCondition(" 2 == 2");
switchResultVo2.setNextNode("4");
switchResultVo2.setNextNode(4L);
List<SwitchResultVo> list = new ArrayList<>();
list.add(switchResultVo1);
list.add(switchResultVo2);
conditionsParameters.setDependTaskList(list);
conditionsParameters.setNextNode("5");
conditionsParameters.setNextNode(5L);
conditionsParameters.setRelation("AND");
// in: AND(AND(1 is SUCCESS))
@ -540,7 +540,7 @@ public class DagHelperTest {
assert processData != null;
List<TaskNode> taskNodeList = processData.getTasks();
ProcessDag processDag = DagHelper.getProcessDag(taskNodeList);
DAG<String, TaskNode, TaskNodeRelation> dag = DagHelper.buildDagGraph(processDag);
DAG<Long, TaskNode, TaskNodeRelation> dag = DagHelper.buildDagGraph(processDag);
Assertions.assertNotNull(dag);
}

12
dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/model/SwitchResultVo.java
View File

@ -30,16 +30,12 @@ import lombok.NoArgsConstructor;
public class SwitchResultVo {
private String condition;
private List<String> nextNode;
private List<Long> nextNode;
public void setNextNode(Object nextNode) {
if (nextNode instanceof String) {
List<String> nextNodeList = new ArrayList<>();
nextNodeList.add(String.valueOf(nextNode));
this.nextNode = nextNodeList;
} else if (nextNode instanceof Number) {
List<String> nextNodeList = new ArrayList<>();
nextNodeList.add(nextNode.toString());
if (nextNode instanceof Long) {
List<Long> nextNodeList = new ArrayList<>();
nextNodeList.add((Long) nextNode);
this.nextNode = nextNodeList;
} else {
this.nextNode = (ArrayList) nextNode;

39
dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/parameters/ConditionsParameters.java
View File

@ -24,6 +24,9 @@ import org.apache.dolphinscheduler.plugin.task.api.model.ResourceInfo;
import java.util.ArrayList;
import java.util.List;
import lombok.Data;
@Data
public class ConditionsParameters extends AbstractParameters {
// depend node list and state, only need task name
@ -31,10 +34,10 @@ public class ConditionsParameters extends AbstractParameters {
private DependentRelation dependRelation;
// node list to run when success
private List<String> successNode;
private List<Long> successNode;
// node list to run when failed
private List<String> failedNode;
private List<Long> failedNode;
@Override
public boolean checkParameters() {
@ -46,38 +49,6 @@ public class ConditionsParameters extends AbstractParameters {
return new ArrayList<>();
}
public List<DependentTaskModel> getDependTaskList() {
return dependTaskList;
}
public void setDependTaskList(List<DependentTaskModel> dependTaskList) {
this.dependTaskList = dependTaskList;
}
public DependentRelation getDependRelation() {
return dependRelation;
}
public void setDependRelation(DependentRelation dependRelation) {
this.dependRelation = dependRelation;
}
public List<String> getSuccessNode() {
return successNode;
}
public void setSuccessNode(List<String> successNode) {
this.successNode = successNode;
}
public List<String> getFailedNode() {
return failedNode;
}
public void setFailedNode(List<String> failedNode) {
this.failedNode = failedNode;
}
public String getConditionResult() {
return "{"
+ "\"successNode\": [\"" + successNode.get(0)

14
dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/parameters/SwitchParameters.java
View File

@ -27,7 +27,7 @@ public class SwitchParameters extends AbstractParameters {
private DependentRelation dependRelation;
private String relation;
private List<String> nextNode;
private List<Long> nextNode;
@Override
public boolean checkParameters() {
@ -69,18 +69,14 @@ public class SwitchParameters extends AbstractParameters {
this.dependTaskList = dependTaskList;
}
public List<String> getNextNode() {
public List<Long> getNextNode() {
return nextNode;
}
public void setNextNode(Object nextNode) {
if (nextNode instanceof String) {
List<String> nextNodeList = new ArrayList<>();
nextNodeList.add(String.valueOf(nextNode));
this.nextNode = nextNodeList;
} else if (nextNode instanceof Number) {
List<String> nextNodeList = new ArrayList<>();
nextNodeList.add(nextNode.toString());
if (nextNode instanceof Long) {
List<Long> nextNodeList = new ArrayList<>();
nextNodeList.add((Long) nextNode);
this.nextNode = nextNodeList;
} else {
this.nextNode = (ArrayList) nextNode;