diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/IChain.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/dag/BaseDAG.java similarity index 90% rename from dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/IChain.java rename to dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/dag/BaseDAG.java index 5c86c8b6f7..a63154c2c4 100644 --- a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/IChain.java +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/dag/BaseDAG.java @@ -15,5 +15,7 @@ * limitations under the License. */ -package org.apache.dolphinscheduler.workflow.engine.workflow;public interface IChain { +package org.apache.dolphinscheduler.workflow.engine.dag; + +public class BaseDAG { } diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/dag/DAG.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/dag/DAG.java index aed6193c83..80a5d89143 100644 --- a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/dag/DAG.java +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/dag/DAG.java @@ -18,7 +18,6 @@ package org.apache.dolphinscheduler.workflow.engine.dag; import java.util.List; -import java.util.stream.Collectors; /** * The Directed Acyclic Graph class. @@ -27,9 +26,8 @@ import java.util.stream.Collectors; * The nodes are the tasks, the edges are the dependencies between the tasks. * The DAG is acyclic, which means there is no cycle in the graph. * The DAG is a directed graph, which means the edges have direction. - * */ -public interface DAG { +public interface DAG { /** * Get the direct post node of given dagNode, if the dagNode is null, return the nodes which doesn't have inDegrees. @@ -50,26 +48,7 @@ public interface DAG { */ List getDirectPostNodes(Node node); - /** - * Same with {@link #getDirectPostNodes(Node)}. - *

- * If the dagNodeName is null, return the nodes which doesn't have inDegrees. Otherwise, return the post nodes of - * the given dagNodeName. If the dagNodeName is not null and cannot find the node in DAG, throw IllegalArgumentException. - * - * @param dagNodeName task name, can be null. - * @return post task name list, sort by priority. - * @throws IllegalArgumentException if the dagNodeName is not null and cannot find the node in DAG. - */ - List getDirectPostNodes(String dagNodeName); - - /** - * Same with {@link #getDirectPostNodes(String)}. Return the post node names. - * - * @param dagNodeName task name, can be null. - * @return post task name list, sort by priority. - * @throws IllegalArgumentException if the dagNodeName is not null and cannot find the node in DAG. - */ - List getDirectPostNodeNames(String dagNodeName); + List getDirectPostNodesByIdentify(NodeIdentify nodeIdentify); /** * Get the direct pre node of given dagNode, if the dagNode is null, return the nodes which doesn't have outDegrees. @@ -90,33 +69,14 @@ public interface DAG { */ List getDirectPreNodes(Node node); - /** - * Same with {@link #getDirectPreNodes(Node)}. - *

- * If the dagNodeName is null, return the nodes which doesn't have outDegrees. Otherwise, return the pre nodes of - * the given dagNodeName. If the dagNodeName is not null and cannot find the node in DAG, throw IllegalArgumentException. - * - * @param dagNodeName task name, can be null. - * @return pre task name list, sort by priority. - * @throws IllegalArgumentException if the dagNodeName is not null and cannot find the node in DAG. - */ - List getDirectPreNodes(String dagNodeName); - - /** - * Same with {@link #getDirectPreNodes(String)}. Return the pre node names. - * - * @param dagNodeName task name, can be null. - * @return pre task name list, sort by priority. - * @throws IllegalArgumentException if the dagNodeName is not null and cannot find the node in DAG. - */ - List getDirectPreNodeNames(String dagNodeName); + List getDirectPreNodesByIdentify(NodeIdentify nodeIdentify); /** * Get the node of the DAG by the node name. * - * @param nodeName the node name. + * @param nodeIdentify the node name. * @return the node of the DAG, return null if cannot find the node. */ - Node getDAGNode(String nodeName); + Node getDAGNode(NodeIdentify nodeIdentify); } diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/dag/ITask.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/dag/ITask.java new file mode 100644 index 0000000000..6d6c0c6f5f --- /dev/null +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/dag/ITask.java @@ -0,0 +1,26 @@ +/* + * 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.workflow.engine.dag; + +public interface ITask { + + ITaskIdentify getIdentify(); + + ITaskContext getContext(); + +} diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/dag/ITaskChain.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/dag/ITaskChain.java new file mode 100644 index 0000000000..ccefc085e5 --- /dev/null +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/dag/ITaskChain.java @@ -0,0 +1,26 @@ +/* + * 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.workflow.engine.dag; + +public interface ITaskChain { + + ITask getFrom(); + + ITask getTo(); + +} diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/dag/ITaskContext.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/dag/ITaskContext.java new file mode 100644 index 0000000000..69e4a00d42 --- /dev/null +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/dag/ITaskContext.java @@ -0,0 +1,22 @@ +/* + * 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.workflow.engine.dag; + +public interface ITaskContext { + +} diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/dag/ITaskIdentify.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/dag/ITaskIdentify.java new file mode 100644 index 0000000000..d0968582d4 --- /dev/null +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/dag/ITaskIdentify.java @@ -0,0 +1,26 @@ +/* + * 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.workflow.engine.dag; + +public interface ITaskIdentify { + + Long getId(); + + String getName(); + +} diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/dag/Node.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/dag/Node.java index 8e8719ac7d..7c04ccd97a 100644 --- a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/dag/Node.java +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/dag/Node.java @@ -19,11 +19,6 @@ package org.apache.dolphinscheduler.workflow.engine.dag; import lombok.AllArgsConstructor; import lombok.Builder; -import org.apache.commons.collections4.CollectionUtils; -import org.apache.commons.lang3.StringUtils; - -import java.util.List; - import lombok.Getter; import lombok.NoArgsConstructor; diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/dag/Task.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/dag/Task.java new file mode 100644 index 0000000000..bc0c80bf1d --- /dev/null +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/dag/Task.java @@ -0,0 +1,40 @@ +/* + * 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.workflow.engine.dag; + +public class Task implements ITask { + + private final TaskIdentify taskIdentify; + + private final TaskContext taskContext; + + public Task(TaskIdentify taskIdentify, TaskContext taskContext) { + this.taskIdentify = taskIdentify; + this.taskContext = taskContext; + } + + @Override + public TaskIdentify getIdentify() { + return taskIdentify; + } + + @Override + public TaskContext getContext() { + return taskContext; + } +} diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/dag/TaskChain.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/dag/TaskChain.java new file mode 100644 index 0000000000..b47d99059f --- /dev/null +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/dag/TaskChain.java @@ -0,0 +1,43 @@ +/* + * 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.workflow.engine.dag; + +public class TaskChain implements ITaskChain { + + private final Task from; + + private final Task to; + + public TaskChain(Task from, Task to) { + if (from == null && to == null) { + throw new IllegalArgumentException("from and to can not be null at the same time"); + } + this.from = from; + this.to = to; + } + + @Override + public Task getFrom() { + return from; + } + + @Override + public Task getTo() { + return to; + } +} diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/dag/TaskContext.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/dag/TaskContext.java new file mode 100644 index 0000000000..8f542549db --- /dev/null +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/dag/TaskContext.java @@ -0,0 +1,21 @@ +/* + * 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.workflow.engine.dag; + +public class TaskContext implements ITaskContext { +} diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/WorkflowInstance.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/dag/TaskIdentify.java similarity index 71% rename from dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/WorkflowInstance.java rename to dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/dag/TaskIdentify.java index 5280e86c08..2afc1a95a7 100644 --- a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/WorkflowInstance.java +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/dag/TaskIdentify.java @@ -15,24 +15,29 @@ * limitations under the License. */ -package org.apache.dolphinscheduler.workflow.engine.workflow; +package org.apache.dolphinscheduler.workflow.engine.dag; -import lombok.Getter; +import lombok.EqualsAndHashCode; -@Getter -public class WorkflowInstance implements IWorkflowInstance { - - private final int id; +@EqualsAndHashCode +public class TaskIdentify implements ITaskIdentify { + private final Long id; private final String name; - public WorkflowInstance(int id, String name) { + public TaskIdentify(Long id, String name) { this.id = id; this.name = name; } - public static WorkflowInstance of(int id, String name) { - return new WorkflowInstance(id, name); + @Override + public Long getId() { + return id; + } + + @Override + public String getName() { + return name; } } diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/dag/WorkflowDAG.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/dag/WorkflowDAG.java index 1b3cb35e40..65fed98a88 100644 --- a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/dag/WorkflowDAG.java +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/dag/WorkflowDAG.java @@ -17,105 +17,108 @@ package org.apache.dolphinscheduler.workflow.engine.dag; -import java.util.ArrayList; import java.util.Collections; import java.util.HashMap; +import java.util.HashSet; import java.util.List; import java.util.Map; -import java.util.function.Function; +import java.util.Set; import java.util.stream.Collectors; /** * The IWorkflowDAG represent the DAG of a workflow. */ -public class WorkflowDAG implements DAG { +public class WorkflowDAG implements DAG { - private final Map dagNodeMap; + private final Map dagNodeMap; - private final Map> outdegreeMap; + private final Map> outdegreeMap; - private final Map> inDegredMap; + private final Map> inDegredMap; - public WorkflowDAG(List nodes, List edges) { - this.dagNodeMap = nodes.stream().collect(Collectors.toMap(Node::getNodeIdentify, Function.identity())); + public WorkflowDAG(List tasks, + List taskChains) { + this.dagNodeMap = new HashMap<>(); this.outdegreeMap = new HashMap<>(); this.inDegredMap = new HashMap<>(); - // todo: - } - @Override - public List getDirectPostNodes(Node dagNode) { - NodeIdentify nodeIdentify = dagNode.getNodeIdentify(); - if (!dagNodeMap.containsKey(nodeIdentify)) { - return Collections.emptyList(); - } - Node node = dagNodeMap.get(nodeIdentify); - List nodes = new ArrayList<>(); - for (DAGEdge edge : node.getOutDegrees()) { - if (dagNodeMap.containsKey(edge.getToNodeName())) { - nodes.add(dagNodeMap.get(edge.getToNodeName())); + for (ITask task : tasks) { + ITaskIdentify identify = task.getIdentify(); + if (dagNodeMap.containsKey(identify)) { + throw new IllegalArgumentException("Duplicate task identify: " + identify); } + dagNodeMap.put(identify, task); + } + for (ITaskChain taskChain : taskChains) { + ITask from = taskChain.getFrom(); + ITask to = taskChain.getTo(); + if (from == null) { + continue; + } + if (to == null) { + continue; + } + ITaskIdentify fromIdentify = from.getIdentify(); + ITaskIdentify toIdentify = to.getIdentify(); + Set outDegrees = outdegreeMap.computeIfAbsent(fromIdentify, k -> new HashSet<>()); + if (outDegrees.contains(toIdentify)) { + throw new IllegalArgumentException("Duplicate task chain: " + fromIdentify + " -> " + toIdentify); + } + outDegrees.add(toIdentify); + Set inDegrees = inDegredMap.computeIfAbsent(toIdentify, k -> new HashSet<>()); + if (inDegrees.contains(fromIdentify)) { + throw new IllegalArgumentException("Duplicate task chain: " + fromIdentify + " -> " + toIdentify); + } + inDegrees.add(fromIdentify); } - return nodes; } @Override - public List getDirectPostNodes(String dagNodeName) { - Node node = getDAGNode(dagNodeName); - if (dagNodeName != null && node == null) { - throw new IllegalArgumentException("Cannot find the Node: " + dagNodeName + " in DAG"); + public List getDirectPostNodes(ITask iTask) { + if (iTask == null) { + return getDirectPostNodesByIdentify(null); } - return getDirectPostNodes(node); + return getDirectPostNodesByIdentify(iTask.getIdentify()); } @Override - public List getDirectPostNodeNames(String dagNodeName) { - Node node = getDAGNode(dagNodeName); - if (dagNodeName != null && node == null) { - throw new IllegalArgumentException("Cannot find the Node: " + dagNodeName + " in DAG"); + public List getDirectPostNodesByIdentify(ITaskIdentify taskIdentify) { + if (taskIdentify == null) { + return dagNodeMap.values() + .stream() + .filter(task -> !inDegredMap.containsKey(task.getIdentify())) + .collect(Collectors.toList()); } - return getDirectPostNodes(node).stream() - .map(Node::getNodeName) + return outdegreeMap.getOrDefault(taskIdentify, Collections.emptySet()) + .stream() + .map(dagNodeMap::get) .collect(Collectors.toList()); } @Override - public List getDirectPreNodes(Node dagNode) { - final String nodeName = dagNode.getNodeName(); - if (!dagNodeMap.containsKey(nodeName)) { - return Collections.emptyList(); + public List getDirectPreNodes(ITask iTask) { + if (iTask == null) { + return getDirectPreNodesByIdentify(null); } - Node node = dagNodeMap.get(nodeName); - List nodes = new ArrayList<>(); - for (DAGEdge edge : node.getInDegrees()) { - if (dagNodeMap.containsKey(edge.getFromNodeName())) { - nodes.add(dagNodeMap.get(edge.getFromNodeName())); - } - } - return nodes; + return getDirectPreNodesByIdentify(iTask.getIdentify()); } @Override - public List getDirectPreNodes(String dagNodeName) { - Node node = getDAGNode(dagNodeName); - if (dagNodeName != null && node == null) { - throw new IllegalArgumentException("Cannot find the Node: " + dagNodeName + " in DAG"); + public List getDirectPreNodesByIdentify(ITaskIdentify taskIdentify) { + if (taskIdentify == null) { + return dagNodeMap.values() + .stream() + .filter(task -> !outdegreeMap.containsKey(taskIdentify)) + .collect(Collectors.toList()); } - return getDirectPreNodes(node); + return inDegredMap.getOrDefault(taskIdentify, Collections.emptySet()) + .stream() + .map(dagNodeMap::get) + .collect(Collectors.toList()); } @Override - public List getDirectPreNodeNames(String dagNodeName) { - Node node = getDAGNode(dagNodeName); - if (dagNodeName != null && node == null) { - throw new IllegalArgumentException("Cannot find the Node: " + dagNodeName + " in DAG"); - } - return getDirectPreNodes(node).stream().map(Node::getNodeName).collect(Collectors.toList()); + public ITask getDAGNode(ITaskIdentify taskIdentify) { + return dagNodeMap.get(taskIdentify); } - - @Override - public Node getDAGNode(String nodeName) { - return dagNodeMap.get(nodeName); - } - } diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/engine/DAGEngine.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/engine/DAGEngine.java index 0a847583b4..f928322d87 100644 --- a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/engine/DAGEngine.java +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/engine/DAGEngine.java @@ -17,155 +17,167 @@ package org.apache.dolphinscheduler.workflow.engine.engine; -import org.apache.dolphinscheduler.workflow.engine.dag.Node; +import org.apache.dolphinscheduler.workflow.engine.dag.ITask; +import org.apache.dolphinscheduler.workflow.engine.dag.ITaskIdentify; +import org.apache.dolphinscheduler.workflow.engine.dag.WorkflowDAG; +import org.apache.dolphinscheduler.workflow.engine.event.IEventRepository; import org.apache.dolphinscheduler.workflow.engine.event.TaskOperationEvent; -import org.apache.dolphinscheduler.workflow.engine.event.TaskOperationType; +import org.apache.dolphinscheduler.workflow.engine.event.WorkflowFinishEvent; +import org.apache.dolphinscheduler.workflow.engine.workflow.ITaskExecutionPlan; import org.apache.dolphinscheduler.workflow.engine.workflow.ITaskExecutionRunnable; -import org.apache.dolphinscheduler.workflow.engine.workflow.ITaskExecutionRunnableFactory; +import org.apache.dolphinscheduler.workflow.engine.workflow.ITaskExecutionRunnableIdentify; import org.apache.dolphinscheduler.workflow.engine.workflow.IWorkflowExecutionContext; -import org.apache.dolphinscheduler.workflow.engine.workflow.IWorkflowExecutionDAG; +import org.apache.dolphinscheduler.workflow.engine.workflow.IWorkflowExecutionRunnableIdentify; +import org.apache.dolphinscheduler.workflow.engine.workflow.WorkflowExecutionDAG; + +import org.apache.commons.collections4.CollectionUtils; + +import java.util.List; +import java.util.stream.Collectors; import lombok.extern.slf4j.Slf4j; +@Deprecated @Slf4j public class DAGEngine implements IDAGEngine { private final IWorkflowExecutionContext workflowExecutionContext; - private final ITaskExecutionRunnableFactory taskExecutionRunnableFactory; + private final IEventRepository eventRepository; + private final WorkflowExecutionDAG workflowExecutionDAG; + private final WorkflowDAG workflowDAG; - public DAGEngine(IWorkflowExecutionContext workflowExecutionContext, - ITaskExecutionRunnableFactory taskExecutionRunnableFactory) { + private final IWorkflowExecutionRunnableIdentify workflowExecutionRunnableIdentify; + + public DAGEngine(IWorkflowExecutionContext workflowExecutionContext) { this.workflowExecutionContext = workflowExecutionContext; - this.taskExecutionRunnableFactory = taskExecutionRunnableFactory; + + this.workflowExecutionRunnableIdentify = workflowExecutionContext.getIdentify(); + this.eventRepository = workflowExecutionContext.getEventRepository(); + this.workflowDAG = workflowExecutionContext.getWorkflowDAG(); + this.workflowExecutionDAG = workflowExecutionContext.getWorkflowExecutionDAG(); } @Override - public void triggerNextTasks(String parentTaskNodeName) { - workflowExecutionContext.getWorkflowExecutionDAG() - .getDirectPostNodeNames(parentTaskNodeName) - .forEach(this::triggerTask); + public void start() { + List startTaskIdentifies = workflowExecutionContext.getStartTaskIdentifies(); + // If the start task is empty, trigger from the beginning + if (CollectionUtils.isEmpty(startTaskIdentifies)) { + startTaskIdentifies = workflowDAG.getDirectPostNodesByIdentify(null) + .stream() + .map(ITask::getIdentify) + .collect(Collectors.toList()); + } + + if (CollectionUtils.isEmpty(startTaskIdentifies)) { + workflowFinish(); + return; + } + startTaskIdentifies.forEach(this::triggerTask); } @Override - public void triggerTask(String taskName) { - IWorkflowExecutionDAG workflowExecutionDAG = workflowExecutionContext.getWorkflowExecutionDAG(); - Node node = workflowExecutionDAG.getDAGNode(taskName); - if (node == null) { - log.error("Cannot find the DAGNode for task: {}", taskName); - return; - } - - // todo: Use condition check? - // How to make sure the - if (!workflowExecutionDAG.isTaskAbleToBeTriggered(taskName)) { - log.info("The task: {} is not able to be triggered", taskName); - return; - } - - if (node.isSkip()) { - log.info("The task: {} is skipped", taskName); - triggerNextTasks(taskName); - return; - } - - ITaskExecutionRunnable taskExecutionRunnable = - taskExecutionRunnableFactory.createTaskExecutionRunnable(taskName, workflowExecutionContext); - workflowExecutionDAG.storeTaskExecutionRunnable(taskExecutionRunnable); - TaskOperationEvent taskOperationEvent = TaskOperationEvent.builder() - .taskExecutionRunnable(taskExecutionRunnable) - .taskOperationType(TaskOperationType.RUN) - .build(); - workflowExecutionContext.getEventRepository().storeEventToTail(taskOperationEvent); - } - - @Override - public void failoverTask(Integer taskInstanceId) { - IWorkflowExecutionDAG workflowExecutionDAG = workflowExecutionContext.getWorkflowExecutionDAG(); - ITaskExecutionRunnable taskExecutionRunnable = - workflowExecutionDAG.getTaskExecutionRunnableById(taskInstanceId); - if (taskExecutionRunnable == null) { - log.error("Cannot find the ITaskExecutionRunnable for taskInstance: {}", taskInstanceId); - return; - } - ITaskExecutionRunnable failoverTaskExecutionRunnable = taskExecutionRunnableFactory - .createFailoverTaskExecutionRunnable(taskExecutionRunnable, workflowExecutionContext); - workflowExecutionDAG.storeTaskExecutionRunnable(failoverTaskExecutionRunnable); - - TaskOperationEvent taskOperationEvent = TaskOperationEvent.builder() - .taskExecutionRunnable(taskExecutionRunnable) - .taskOperationType(TaskOperationType.FAILOVER) - .build(); - workflowExecutionContext.getEventRepository().storeEventToTail(taskOperationEvent); - } - - @Override - public void retryTask(Integer taskInstanceId) { - IWorkflowExecutionDAG workflowExecutionDAG = workflowExecutionContext.getWorkflowExecutionDAG(); - ITaskExecutionRunnable taskExecutionRunnable = - workflowExecutionDAG.getTaskExecutionRunnableById(taskInstanceId); - if (taskExecutionRunnable == null) { - log.error("Cannot find the ITaskExecutionRunnable for taskInstance: {}", taskInstanceId); - return; - } - ITaskExecutionRunnable retryTaskExecutionRunnable = taskExecutionRunnableFactory - .createRetryTaskExecutionRunnable(taskExecutionRunnable, workflowExecutionContext); - workflowExecutionDAG.storeTaskExecutionRunnable(retryTaskExecutionRunnable); - - TaskOperationEvent taskOperationEvent = TaskOperationEvent.builder() - .taskExecutionRunnable(taskExecutionRunnable) - .taskOperationType(TaskOperationType.RETRY) - .build(); - workflowExecutionContext.getEventRepository().storeEventToTail(taskOperationEvent); - } - - @Override - public void pauseAllTask() { - workflowExecutionContext.getWorkflowExecutionDAG() - .getActiveTaskExecutionRunnable() + public void triggerNextTasks(ITaskIdentify taskIdentify) { + List directPostNodeIdentifies = workflowDAG.getDirectPostNodesByIdentify(taskIdentify) .stream() - .map(taskExecutionRunnable -> taskExecutionRunnable.getTaskExecutionContext().getTaskInstance().getId()) - .forEach(this::pauseTask); - } - - @Override - public void pauseTask(Integer taskInstanceId) { - ITaskExecutionRunnable taskExecutionRunnable = - workflowExecutionContext.getWorkflowExecutionDAG().getTaskExecutionRunnableById(taskInstanceId); - if (taskExecutionRunnable == null) { - log.error("Cannot find the ITaskExecutionRunnable for taskInstance: {}", taskInstanceId); + .map(ITask::getIdentify) + .collect(Collectors.toList()); + if (CollectionUtils.isNotEmpty(directPostNodeIdentifies)) { + directPostNodeIdentifies.forEach(this::triggerTask); return; } - TaskOperationEvent taskOperationEvent = TaskOperationEvent.builder() - .taskExecutionRunnable(taskExecutionRunnable) - .taskOperationType(TaskOperationType.PAUSE) - .build(); - workflowExecutionContext.getEventRepository().storeEventToTail(taskOperationEvent); + List activeTaskExecutionIdentify = getActiveTaskExecutionIdentify(); + if (CollectionUtils.isEmpty(activeTaskExecutionIdentify)) { + workflowFinish(); + return; + } + // The task chain is finished, but there are still active tasks, wait for the active tasks to finish } @Override - public void killAllTask() { - workflowExecutionContext.getWorkflowExecutionDAG() - .getActiveTaskExecutionRunnable() + public void triggerTask(ITaskIdentify taskIdentify) { + ITaskExecutionPlan taskExecutionPlan = workflowExecutionDAG.getDAGNode(taskIdentify); + if (taskExecutionPlan == null) { + throw new IllegalArgumentException("Cannot find the ITaskExecutionPlan for taskIdentify: " + taskIdentify); + } + eventRepository.storeEventToTail(TaskOperationEvent.startEvent(taskExecutionPlan)); + } + + @Override + public void failoverTask(ITaskExecutionRunnableIdentify taskExecutionRunnableIdentify) { + ITaskExecutionPlan taskExecutionPlan = + workflowExecutionDAG.getDAGNode(taskExecutionRunnableIdentify.getTaskIdentify()); + if (taskExecutionPlan == null) { + throw new IllegalArgumentException("Cannot find the ITaskExecutionPlan for taskIdentify: " + + taskExecutionRunnableIdentify.getTaskIdentify()); + } + eventRepository.storeEventToTail(TaskOperationEvent.failoverEvent(taskExecutionPlan)); + } + + @Override + public void retryTask(ITaskExecutionRunnableIdentify taskExecutionRunnableIdentify) { + ITaskExecutionPlan taskExecutionPlan = + workflowExecutionDAG.getDAGNode(taskExecutionRunnableIdentify.getTaskIdentify()); + if (taskExecutionPlan == null) { + throw new IllegalArgumentException("Cannot find the ITaskExecutionPlan for taskIdentify: " + + taskExecutionRunnableIdentify.getTaskIdentify()); + } + eventRepository.storeEventToTail(TaskOperationEvent.retryEvent(taskExecutionPlan)); + } + + @Override + public void pause() { + List activeTaskExecutionIdentify = getActiveTaskExecutionIdentify(); + if (CollectionUtils.isEmpty(activeTaskExecutionIdentify)) { + workflowFinish(); + return; + } + activeTaskExecutionIdentify.forEach(this::pauseTask); + } + + @Override + public void pauseTask(ITaskExecutionRunnableIdentify taskExecutionIdentify) { + ITaskExecutionPlan taskExecutionPlan = workflowExecutionDAG.getDAGNode(taskExecutionIdentify.getTaskIdentify()); + if (taskExecutionPlan == null) { + throw new IllegalArgumentException( + "Cannot find the ITaskExecutionPlan for taskIdentify: " + taskExecutionIdentify.getTaskIdentify()); + } + eventRepository.storeEventToTail(TaskOperationEvent.pauseEvent(taskExecutionPlan)); + } + + @Override + public void kill() { + List activeTaskExecutionIdentify = getActiveTaskExecutionIdentify(); + if (CollectionUtils.isEmpty(activeTaskExecutionIdentify)) { + workflowFinish(); + return; + } + activeTaskExecutionIdentify.forEach(this::killTask); + } + + @Override + public void killTask(ITaskExecutionRunnableIdentify taskExecutionIdentify) { + ITaskExecutionPlan taskExecutionPlan = workflowExecutionDAG.getDAGNode(taskExecutionIdentify.getTaskIdentify()); + if (taskExecutionPlan == null) { + throw new IllegalArgumentException( + "Cannot find the ITaskExecutionPlan for taskIdentify: " + taskExecutionIdentify.getTaskIdentify()); + } + eventRepository.storeEventToTail(TaskOperationEvent.killEvent(taskExecutionPlan)); + } + + private void workflowFinish() { + if (workflowExecutionDAG.isFailed()) { + + } + eventRepository.storeEventToTail(WorkflowFinishEvent.of(workflowExecutionRunnableIdentify)); + } + + private List getActiveTaskExecutionIdentify() { + return workflowExecutionDAG.getActiveTaskExecutionPlan() .stream() - .map(taskExecutionRunnable -> taskExecutionRunnable.getTaskExecutionContext().getTaskInstance().getId()) - .forEach(this::killTask); - } - - @Override - public void killTask(Integer taskInstanceId) { - ITaskExecutionRunnable taskExecutionRunnable = - workflowExecutionContext.getWorkflowExecutionDAG().getTaskExecutionRunnableById(taskInstanceId); - if (taskExecutionRunnable == null) { - log.error("Cannot find the ITaskExecutionRunnable for taskInstance: {}", taskInstanceId); - return; - } - - TaskOperationEvent taskOperationEvent = TaskOperationEvent.builder() - .taskExecutionRunnable(taskExecutionRunnable) - .taskOperationType(TaskOperationType.KILL) - .build(); - workflowExecutionContext.getEventRepository().storeEventToTail(taskOperationEvent); + .map(ITaskExecutionPlan::getActiveTaskExecutionRunnable) + .map(ITaskExecutionRunnable::getIdentify) + .collect(Collectors.toList()); } } diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/engine/DAGEngineFactory.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/engine/DAGEngineFactory.java index c07ca6ffbb..78cbdabcb6 100644 --- a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/engine/DAGEngineFactory.java +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/engine/DAGEngineFactory.java @@ -33,6 +33,6 @@ public class DAGEngineFactory implements IDAGEngineFactory { @Override public IDAGEngine createDAGEngine(IWorkflowExecutionContext workflowExecutionContext) { - return new DAGEngine(workflowExecutionContext, taskExecutionRunnableFactory); + return new DAGEngine(workflowExecutionContext); } } diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/engine/EventEngine.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/engine/EventEngine.java index 10e483f0ad..70e83820fa 100644 --- a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/engine/EventEngine.java +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/engine/EventEngine.java @@ -22,6 +22,7 @@ import org.apache.dolphinscheduler.common.thread.BaseDaemonThread; import org.apache.dolphinscheduler.common.thread.ThreadUtils; import org.apache.dolphinscheduler.workflow.engine.workflow.IWorkflowExecutionContext; import org.apache.dolphinscheduler.workflow.engine.workflow.IWorkflowExecutionRunnable; +import org.apache.dolphinscheduler.workflow.engine.workflow.IWorkflowExecutionRunnableIdentify; import org.apache.dolphinscheduler.workflow.engine.workflow.IWorkflowExecutionRunnableRepository; import org.apache.commons.collections4.CollectionUtils; @@ -83,24 +84,22 @@ public class EventEngine extends BaseDaemonThread implements IEventEngine { for (IWorkflowExecutionRunnable workflowExecutionRunnable : workflowExecutionRunnableList) { IWorkflowExecutionContext workflowExecutionContext = workflowExecutionRunnable.getWorkflowExecutionContext(); - final Integer workflowInstanceId = workflowExecutionContext.getWorkflowInstanceId(); - final String workflowInstanceName = workflowExecutionContext.getWorkflowInstanceName(); + IWorkflowExecutionRunnableIdentify identify = workflowExecutionContext.getIdentify(); try { - MDC.put(Constants.WORKFLOW_INSTANCE_ID_MDC_KEY, String.valueOf(workflowInstanceId)); - if (workflowExecutionRunnable.isEventFiring()) { - log.debug("WorkflowExecutionRunnable: {} is already in firing", workflowInstanceName); - continue; - } + MDC.put(Constants.WORKFLOW_INSTANCE_ID_MDC_KEY, String.valueOf(identify.getId())); + // if (workflowExecutionRunnable.isEventFiring()) { + // log.debug("WorkflowExecutionRunnable: {} is already in firing", identify); + // continue; + // } eventFirer.fireActiveEvents(workflowExecutionRunnable) .whenComplete((fireCount, ex) -> { - workflowExecutionRunnable.setEventFiring(false); + // workflowExecutionRunnable.setEventFiring(false); if (ex != null) { - log.error("Fire event for WorkflowExecutionRunnable: {} error", workflowInstanceName, - ex); + log.error("Fire event for WorkflowExecutionRunnable: {} error", identify, ex); } else { if (fireCount > 0) { log.info("Fire {} events for WorkflowExecutionRunnable: {} success", fireCount, - workflowInstanceName); + identify); } } }); diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/engine/EventEngineFactory.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/engine/EventEngineFactory.java index 6438990825..102ba0f937 100644 --- a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/engine/EventEngineFactory.java +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/engine/EventEngineFactory.java @@ -17,9 +17,6 @@ package org.apache.dolphinscheduler.workflow.engine.engine; -import org.apache.dolphinscheduler.workflow.engine.event.EventOperatorManager; -import org.apache.dolphinscheduler.workflow.engine.event.IEvent; -import org.apache.dolphinscheduler.workflow.engine.event.IEventOperatorManager; import org.apache.dolphinscheduler.workflow.engine.workflow.IWorkflowExecutionRunnableRepository; import org.apache.dolphinscheduler.workflow.engine.workflow.SingletonWorkflowExecutionRunnableRepository; @@ -33,11 +30,6 @@ public class EventEngineFactory implements IEventEngineFactory { private IWorkflowExecutionRunnableRepository workflowExecuteRunnableRepository; - private static final IEventOperatorManager DEFAULT_EVENT_OPERATOR_MANAGER = - EventOperatorManager.getInstance(); - - private IEventOperatorManager eventOperatorManager = DEFAULT_EVENT_OPERATOR_MANAGER; - private static final int DEFAULT_EVENT_FIRE_THREAD_POOL_SIZE = Runtime.getRuntime().availableProcessors() * 2; private int eventFireThreadPoolSize = DEFAULT_EVENT_FIRE_THREAD_POOL_SIZE; @@ -54,11 +46,6 @@ public class EventEngineFactory implements IEventEngineFactory { return this; } - public EventEngineFactory withEventOperatorManager(IEventOperatorManager eventOperatorManager) { - this.eventOperatorManager = eventOperatorManager; - return this; - } - public int withEventFireThreadPoolSize(int eventFireThreadPoolSize) { this.eventFireThreadPoolSize = eventFireThreadPoolSize; return this.eventFireThreadPoolSize; @@ -66,7 +53,7 @@ public class EventEngineFactory implements IEventEngineFactory { @Override public IEventEngine createEventEngine() { - EventFirer eventFirer = new EventFirer(eventOperatorManager, eventFireThreadPoolSize); + EventFirer eventFirer = new EventFirer(eventFireThreadPoolSize); return new EventEngine(workflowExecuteRunnableRepository, eventFirer); } } diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/engine/EventFirer.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/engine/EventFirer.java index ab9fecf8c3..02d4193fa9 100644 --- a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/engine/EventFirer.java +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/engine/EventFirer.java @@ -20,7 +20,6 @@ package org.apache.dolphinscheduler.workflow.engine.engine; import org.apache.dolphinscheduler.common.thread.ThreadUtils; import org.apache.dolphinscheduler.workflow.engine.event.IAsyncEvent; import org.apache.dolphinscheduler.workflow.engine.event.IEvent; -import org.apache.dolphinscheduler.workflow.engine.event.IEventOperatorManager; import org.apache.dolphinscheduler.workflow.engine.event.IEventRepository; import org.apache.dolphinscheduler.workflow.engine.utils.ExceptionUtils; import org.apache.dolphinscheduler.workflow.engine.workflow.IEventfulExecutionRunnable; @@ -33,12 +32,9 @@ import lombok.extern.slf4j.Slf4j; @Slf4j public class EventFirer implements IEventFirer { - private final IEventOperatorManager eventOperatorManager; - private final ThreadPoolExecutor eventFireThreadPool; - public EventFirer(IEventOperatorManager eventOperatorManager, int eventFireThreadPoolSize) { - this.eventOperatorManager = eventOperatorManager; + public EventFirer(int eventFireThreadPoolSize) { this.eventFireThreadPool = ThreadUtils.newDaemonFixedThreadExecutor("EventFireThreadPool", eventFireThreadPoolSize); } @@ -86,7 +82,7 @@ public class EventFirer implements IEventFirer { private void fireAsyncEvent(IEvent event) { CompletableFuture.runAsync(() -> { log.info("Begin fire IAsyncEvent: {}", event); - eventOperatorManager.getEventOperator(event).handleEvent(event); + event.getEventOperation().operate(); log.info("Success fire IAsyncEvent: {}", event); }, eventFireThreadPool).exceptionally(ex -> { log.error("Failed to fire IAsyncEvent: {}", event, ex); @@ -96,7 +92,7 @@ public class EventFirer implements IEventFirer { private void fireSyncEvent(IEvent event) { log.info("Begin fire SyncEvent: {}", event); - eventOperatorManager.getEventOperator(event).handleEvent(event); + event.getEventOperation().operate(); log.info("Success fire SyncEvent: {}", event); } diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/engine/IDAGEngine.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/engine/IDAGEngine.java index 0971e2a179..bea93975e0 100644 --- a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/engine/IDAGEngine.java +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/engine/IDAGEngine.java @@ -17,6 +17,8 @@ package org.apache.dolphinscheduler.workflow.engine.engine; +import org.apache.dolphinscheduler.workflow.engine.dag.ITaskIdentify; +import org.apache.dolphinscheduler.workflow.engine.workflow.ITaskExecutionRunnableIdentify; import org.apache.dolphinscheduler.workflow.engine.workflow.IWorkflowExecutionDAG; /** @@ -26,47 +28,51 @@ import org.apache.dolphinscheduler.workflow.engine.workflow.IWorkflowExecutionDA public interface IDAGEngine { /** - * Trigger the tasks which are post of the given task. - *

If there are no task after the given taskNode, will try to finish the WorkflowExecutionRunnable. - *

If the - * - * @param parentTaskNodeName the parent task name + * Start the DAGEngine, will trigger the start tasks. */ - void triggerNextTasks(String parentTaskNodeName); + void start(); /** - * Trigger the given task + * Trigger the tasks which are post of the given task. + *

If there are no task after the given taskNode, will try to finish the WorkflowExecutionRunnable(Send a task chain end event). * - * @param taskName task name + * @param taskIdentify the parent task identify */ - void triggerTask(String taskName); + void triggerNextTasks(ITaskIdentify taskIdentify); + + /** + * Trigger the given task. + * + * @param taskIdentify task name + */ + void triggerTask(ITaskIdentify taskIdentify); /** * Failover the given task. * * @param taskInstanceId taskInstanceId */ - void failoverTask(Integer taskInstanceId); + void failoverTask(ITaskExecutionRunnableIdentify taskInstanceId); /** * Retry the given task. * * @param taskInstanceId taskInstanceId */ - void retryTask(Integer taskInstanceId); + void retryTask(ITaskExecutionRunnableIdentify taskInstanceId); - void pauseAllTask(); + void pause(); /** * Pause the given task. */ - void pauseTask(Integer taskInstanceId); + void pauseTask(ITaskExecutionRunnableIdentify taskExecutionIdentify); - void killAllTask(); + void kill(); /** * Kill the given task. */ - void killTask(Integer taskId); + void killTask(ITaskExecutionRunnableIdentify taskExecutionIdentify); } diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/engine/IWorkflowEngine.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/engine/IWorkflowEngine.java index 50ccccae67..f173cae643 100644 --- a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/engine/IWorkflowEngine.java +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/engine/IWorkflowEngine.java @@ -19,6 +19,7 @@ package org.apache.dolphinscheduler.workflow.engine.engine; import org.apache.dolphinscheduler.workflow.engine.exception.WorkflowExecuteRunnableNotFoundException; import org.apache.dolphinscheduler.workflow.engine.workflow.IWorkflowExecutionRunnable; +import org.apache.dolphinscheduler.workflow.engine.workflow.IWorkflowExecutionRunnableIdentify; /** * The WorkflowEngine is responsible for starting, stopping, pausing, and finalizing {@link IWorkflowExecutionRunnable}. @@ -41,25 +42,25 @@ public interface IWorkflowEngine { /** * Pause a workflow instance. * - * @param workflowInstanceId the ID of the workflow to pause + * @param workflowExecutionRunnableIdentify the ID of the workflow to pause * @throws WorkflowExecuteRunnableNotFoundException if the workflow is not found */ - void pauseWorkflow(Integer workflowInstanceId); + void pauseWorkflow(IWorkflowExecutionRunnableIdentify workflowExecutionRunnableIdentify); /** * Kill a workflow instance. * - * @param workflowInstanceId the ID of the workflow to stop + * @param workflowExecutionRunnableIdentify the ID of the workflow to stop * @throws WorkflowExecuteRunnableNotFoundException if the workflow is not found */ - void killWorkflow(Integer workflowInstanceId); + void killWorkflow(IWorkflowExecutionRunnableIdentify workflowExecutionRunnableIdentify); /** * Finalize a workflow instance. Once a workflow has been finalized, then it cannot receive new operation, and will be removed from memory. * - * @param workflowInstanceId the ID of the workflow to finalize + * @param workflowExecutionRunnableIdentify the ID of the workflow to finalize */ - void finalizeWorkflow(Integer workflowInstanceId); + void finalizeWorkflow(IWorkflowExecutionRunnableIdentify workflowExecutionRunnableIdentify); /** * Shutdown the workflow engine. The workflow engine cannot be restarted after shutdown. This method will block until the workflow engine is completely shutdown. diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/engine/WorkflowEngine.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/engine/WorkflowEngine.java index b9c2b05391..26498e0e82 100644 --- a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/engine/WorkflowEngine.java +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/engine/WorkflowEngine.java @@ -21,6 +21,7 @@ import org.apache.dolphinscheduler.workflow.engine.event.WorkflowOperationEvent; import org.apache.dolphinscheduler.workflow.engine.exception.WorkflowExecuteRunnableNotFoundException; import org.apache.dolphinscheduler.workflow.engine.workflow.IWorkflowExecutionContext; import org.apache.dolphinscheduler.workflow.engine.workflow.IWorkflowExecutionRunnable; +import org.apache.dolphinscheduler.workflow.engine.workflow.IWorkflowExecutionRunnableIdentify; import org.apache.dolphinscheduler.workflow.engine.workflow.IWorkflowExecutionRunnableRepository; import lombok.extern.slf4j.Slf4j; @@ -46,47 +47,49 @@ public class WorkflowEngine implements IWorkflowEngine { @Override public void triggerWorkflow(IWorkflowExecutionRunnable workflowExecuteRunnable) { IWorkflowExecutionContext workflowExecutionContext = workflowExecuteRunnable.getWorkflowExecutionContext(); - Integer workflowInstanceId = workflowExecutionContext.getWorkflowInstanceId(); - log.info("Triggering WorkflowExecutionRunnable: {}", workflowExecutionContext.getWorkflowInstanceName()); + IWorkflowExecutionRunnableIdentify workflowExecutionIdentify = workflowExecutionContext.getIdentify(); + log.info("Triggering WorkflowExecutionRunnable: {}", workflowExecutionIdentify); workflowExecuteRunnableRepository.storeWorkflowExecutionRunnable(workflowExecuteRunnable); - workflowExecuteRunnable.storeEventToTail(WorkflowOperationEvent.triggerEvent(workflowInstanceId)); + workflowExecuteRunnable + .storeEventToTail(WorkflowOperationEvent.triggerEvent(workflowExecuteRunnable)); } @Override - public void pauseWorkflow(Integer workflowInstanceId) { + public void pauseWorkflow(IWorkflowExecutionRunnableIdentify workflowExecutionRunnableIdentify) { IWorkflowExecutionRunnable workflowExecuteRunnable = - workflowExecuteRunnableRepository.getWorkflowExecutionRunnableById(workflowInstanceId); + workflowExecuteRunnableRepository.getWorkflowExecutionRunnable(workflowExecutionRunnableIdentify); if (workflowExecuteRunnable == null) { - throw new WorkflowExecuteRunnableNotFoundException(workflowInstanceId); + throw new WorkflowExecuteRunnableNotFoundException(workflowExecutionRunnableIdentify); } log.info("Pausing WorkflowExecutionRunnable: {}", - workflowExecuteRunnable.getWorkflowExecutionContext().getWorkflowInstanceName()); - workflowExecuteRunnable.storeEventToTail(WorkflowOperationEvent.pauseEvent(workflowInstanceId)); + workflowExecuteRunnable.getWorkflowExecutionContext().getIdentify()); + workflowExecuteRunnable + .storeEventToTail(WorkflowOperationEvent.pauseEvent(workflowExecuteRunnable)); } @Override - public void killWorkflow(Integer workflowInstanceId) { + public void killWorkflow(IWorkflowExecutionRunnableIdentify workflowExecutionRunnableIdentify) { IWorkflowExecutionRunnable workflowExecuteRunnable = - workflowExecuteRunnableRepository.getWorkflowExecutionRunnableById(workflowInstanceId); + workflowExecuteRunnableRepository.getWorkflowExecutionRunnable(workflowExecutionRunnableIdentify); if (workflowExecuteRunnable == null) { - throw new WorkflowExecuteRunnableNotFoundException(workflowInstanceId); + throw new WorkflowExecuteRunnableNotFoundException(workflowExecutionRunnableIdentify); } log.info("Killing WorkflowExecutionRunnable: {}", - workflowExecuteRunnable.getWorkflowExecutionContext().getWorkflowInstanceName()); - workflowExecuteRunnable.storeEventToTail(WorkflowOperationEvent.killEvent(workflowInstanceId)); + workflowExecuteRunnable.getWorkflowExecutionContext().getIdentify()); + workflowExecuteRunnable + .storeEventToTail(WorkflowOperationEvent.killEvent(workflowExecuteRunnable)); } @Override - public void finalizeWorkflow(Integer workflowInstanceId) { + public void finalizeWorkflow(IWorkflowExecutionRunnableIdentify workflowExecutionRunnableIdentify) { IWorkflowExecutionRunnable workflowExecutionRunnable = - workflowExecuteRunnableRepository.getWorkflowExecutionRunnableById(workflowInstanceId); + workflowExecuteRunnableRepository.getWorkflowExecutionRunnable(workflowExecutionRunnableIdentify); if (workflowExecutionRunnable == null) { return; } // todo: If the workflowExecutionRunnable is not finished, we cannot finalize it. - log.info("Finalizing WorkflowExecutionRunnable: {}", - workflowExecutionRunnable.getWorkflowExecutionContext().getWorkflowInstanceName()); - workflowExecuteRunnableRepository.removeWorkflowExecutionRunnable(workflowInstanceId); + log.info("Finalizing WorkflowExecutionRunnable: {}", workflowExecutionRunnable.getIdentity()); + workflowExecuteRunnableRepository.removeWorkflowExecutionRunnable(workflowExecutionRunnableIdentify); } @Override diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/EventAction.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/EventAction.java new file mode 100644 index 0000000000..fac73cd49c --- /dev/null +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/EventAction.java @@ -0,0 +1,21 @@ +/* + * 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.workflow.engine.event; + +public interface EventAction { +} diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/EventDispatcher.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/EventDispatcher.java new file mode 100644 index 0000000000..7edfa32d71 --- /dev/null +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/EventDispatcher.java @@ -0,0 +1,44 @@ +/* + * 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.workflow.engine.event; + +import org.apache.dolphinscheduler.workflow.engine.workflow.IWorkflowExecutionRunnableIdentify; +import org.apache.dolphinscheduler.workflow.engine.workflow.IWorkflowExecutionRunnableRepository; + +public class EventDispatcher implements IEventDispatcher { + + private final IWorkflowExecutionRunnableRepository workflowExecutionRunnableRepository; + + public EventDispatcher(IWorkflowExecutionRunnableRepository workflowExecutionRunnableRepository) { + this.workflowExecutionRunnableRepository = workflowExecutionRunnableRepository; + } + + // todo: Do we need to split the EventRepository from WorkflowExecutionRunnable? + @Override + public void dispatch(IEvent event) { + IWorkflowExecutionRunnableIdentify workflowExecutionRunnableIdentify; + if (event instanceof IWorkflowEvent) { + workflowExecutionRunnableIdentify = ((IWorkflowEvent) event).getWorkflowExecutionRunnableIdentify(); + } else { + throw new UnsupportedOperationException("Unsupported event: " + event); + } + // todo: + workflowExecutionRunnableRepository.getWorkflowExecutionRunnable(workflowExecutionRunnableIdentify) + .getEventRepository().storeEventToTail(event); + } +} diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/EventOperatorManager.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/EventOperatorManager.java index f2ed5fdcd3..373f51ee46 100644 --- a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/EventOperatorManager.java +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/EventOperatorManager.java @@ -28,7 +28,7 @@ import lombok.extern.slf4j.Slf4j; @Slf4j public class EventOperatorManager implements IEventOperatorManager { - private static final Map> EVENT_OPERATOR_MAP = new HashMap<>(); + private static final Map> EVENT_OPERATOR_MAP = new HashMap<>(); private static final EventOperatorManager INSTANCE = new EventOperatorManager(); @@ -39,8 +39,8 @@ public class EventOperatorManager implements IEventOperatorManager { return INSTANCE; } - public void registerEventOperator(IEventOperator eventOperator) { - EVENT_OPERATOR_MAP.put(eventOperator.getClass().getSimpleName(), eventOperator); + public void registerEventOperator(IEventType eventType, IEventOperator eventOperator) { + EVENT_OPERATOR_MAP.put(eventType, eventOperator); } @Override @@ -48,10 +48,10 @@ public class EventOperatorManager implements IEventOperatorManager { if (event == null) { throw new IllegalArgumentException("event cannot be null"); } - if (event.getEventOperatorClass() == null) { + if (event.getEventType() == null) { throw new IllegalArgumentException("event operator class cannot be null"); } - return EVENT_OPERATOR_MAP.get(event.getEventOperatorClass().getSimpleName()); + return EVENT_OPERATOR_MAP.get(event.getEventType()); } } diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/IEvent.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/IEvent.java index bb270dc323..9630d19b19 100644 --- a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/IEvent.java +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/IEvent.java @@ -17,8 +17,10 @@ package org.apache.dolphinscheduler.workflow.engine.event; +import org.apache.dolphinscheduler.workflow.engine.workflow.IWorkflowExecutionRunnableIdentify; + public interface IEvent { - Class getEventOperatorClass(); + IWorkflowExecutionRunnableIdentify getWorkflowExecutionRunnableIdentify(); } diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/IEventDispatcher.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/IEventDispatcher.java new file mode 100644 index 0000000000..ab01b7e54f --- /dev/null +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/IEventDispatcher.java @@ -0,0 +1,24 @@ +/* + * 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.workflow.engine.event; + +public interface IEventDispatcher { + + void dispatch(IEvent event); + +} diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/IEventOperation.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/IEventOperation.java new file mode 100644 index 0000000000..450495f321 --- /dev/null +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/IEventOperation.java @@ -0,0 +1,23 @@ +/* + * 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.workflow.engine.event; + +public interface IEventOperation { + + void operate(T t); +} diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/IEventOperatorManager.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/IEventOperatorManager.java index 406e5619d1..e0189ad409 100644 --- a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/IEventOperatorManager.java +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/IEventOperatorManager.java @@ -22,7 +22,7 @@ package org.apache.dolphinscheduler.workflow.engine.event; */ public interface IEventOperatorManager { - void registerEventOperator(IEventOperator eventOperator); + void registerEventOperator(IEventType eventType, IEventOperator eventOperator); /** * Get the {@link IEventOperator} for the given event. diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/IEventType.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/IEventType.java new file mode 100644 index 0000000000..d64beef2dd --- /dev/null +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/IEventType.java @@ -0,0 +1,21 @@ +/* + * 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.workflow.engine.event; + +public interface IEventType { +} diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/ISyncEvent.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/ISyncEvent.java index 5b4ecb6162..45fb02c8e9 100644 --- a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/ISyncEvent.java +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/ISyncEvent.java @@ -18,4 +18,5 @@ package org.apache.dolphinscheduler.workflow.engine.event; public interface ISyncEvent { + } diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/ITaskEvent.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/ITaskEvent.java index dd1f315aed..445fe53b14 100644 --- a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/ITaskEvent.java +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/ITaskEvent.java @@ -17,10 +17,6 @@ package org.apache.dolphinscheduler.workflow.engine.event; -import org.apache.dolphinscheduler.workflow.engine.workflow.ITaskExecutionRunnable; - public interface ITaskEvent extends IEvent { - ITaskExecutionRunnable getTaskExecutionRunnable(); - } diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/IWorkflowEvent.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/IWorkflowEvent.java index 304776b582..7c688027a5 100644 --- a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/IWorkflowEvent.java +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/IWorkflowEvent.java @@ -17,13 +17,8 @@ package org.apache.dolphinscheduler.workflow.engine.event; -public interface IWorkflowEvent extends IEvent { +import org.apache.dolphinscheduler.workflow.engine.workflow.IWorkflowExecutionRunnableIdentify; - /** - * The id of WorkflowInstance which the event is related to - * - * @return workflowInstanceId, shouldn't be null - */ - Integer getWorkflowInstanceId(); +public interface IWorkflowEvent extends IEvent { } diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/IWorkflowExecutionRunnableEventBuilder.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/IWorkflowExecutionRunnableEventBuilder.java new file mode 100644 index 0000000000..bf298d622f --- /dev/null +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/IWorkflowExecutionRunnableEventBuilder.java @@ -0,0 +1,22 @@ +/* + * 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.workflow.engine.event; + +public interface IWorkflowExecutionRunnableEventBuilder { + +} diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/IWorkflowExecutionRunnableEventOperation.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/IWorkflowExecutionRunnableEventOperation.java new file mode 100644 index 0000000000..195f140059 --- /dev/null +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/IWorkflowExecutionRunnableEventOperation.java @@ -0,0 +1,22 @@ +/* + * 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.workflow.engine.event; + +public interface IWorkflowExecutionRunnableEventOperation extends IEventOperation { + +} diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/TaskOperationEvent.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/TaskOperationEvent.java index cdeeb30d2b..9b77805dd2 100644 --- a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/TaskOperationEvent.java +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/TaskOperationEvent.java @@ -17,25 +17,39 @@ package org.apache.dolphinscheduler.workflow.engine.event; -import org.apache.dolphinscheduler.workflow.engine.workflow.ITaskExecutionRunnable; +import org.apache.dolphinscheduler.workflow.engine.workflow.ITaskExecutionPlan; -import lombok.AllArgsConstructor; import lombok.Builder; import lombok.Data; -import lombok.NoArgsConstructor; @Data @Builder -@NoArgsConstructor -@AllArgsConstructor public class TaskOperationEvent implements ITaskEvent, ISyncEvent { - private ITaskExecutionRunnable taskExecutionRunnable; + private IEventOperation eventOperation; - private TaskOperationType taskOperationType; - - @Override - public Class getEventOperatorClass() { - return TaskOperationEventOperator.class; + public TaskOperationEvent(IEventOperation eventOperation) { + this.eventOperation = eventOperation; } + + public static TaskOperationEvent startEvent(ITaskExecutionPlan taskExecutionPlan) { + return new TaskOperationEvent(taskExecutionPlan::start); + } + + public static TaskOperationEvent failoverEvent(ITaskExecutionPlan taskExecutionPlan) { + return new TaskOperationEvent(taskExecutionPlan::failoverTask); + } + + public static TaskOperationEvent retryEvent(ITaskExecutionPlan taskExecutionPlan) { + return new TaskOperationEvent(taskExecutionPlan::retryTask); + } + + public static TaskOperationEvent pauseEvent(ITaskExecutionPlan taskExecutionPlan) { + return new TaskOperationEvent(taskExecutionPlan::pauseTask); + } + + public static TaskOperationEvent killEvent(ITaskExecutionPlan taskExecutionPlan) { + return new TaskOperationEvent(taskExecutionPlan::killTask); + } + } diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/TaskOperationEventOperator.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/TaskOperationEventOperator.java index e17868822c..72bcf4843c 100644 --- a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/TaskOperationEventOperator.java +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/TaskOperationEventOperator.java @@ -17,8 +17,6 @@ package org.apache.dolphinscheduler.workflow.engine.event; -import org.apache.dolphinscheduler.workflow.engine.workflow.ITaskExecutionRunnable; - import lombok.extern.slf4j.Slf4j; @Slf4j @@ -26,19 +24,6 @@ public class TaskOperationEventOperator implements ITaskEventOperator getTaskExecutionRunnableList(); + + ITaskExecutionRunnable getTaskExecutionRunnable(ITaskExecutionRunnableIdentify taskExecutionRunnableIdentify); + + ITaskExecutionRunnable storeTaskExecutionRunnable(ITaskExecutionRunnable taskExecutionRunnable); +} diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/ITaskExecutionPlanChain.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/ITaskExecutionPlanChain.java new file mode 100644 index 0000000000..b97dab6390 --- /dev/null +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/ITaskExecutionPlanChain.java @@ -0,0 +1,26 @@ +/* + * 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.workflow.engine.workflow; + +public interface ITaskExecutionPlanChain { + + ITaskExecutionPlan getFrom(); + + ITaskExecutionPlan getTo(); + +} diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/ITaskExecutionRunnable.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/ITaskExecutionRunnable.java index a3ab6cc6a6..268b7d8aa1 100644 --- a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/ITaskExecutionRunnable.java +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/ITaskExecutionRunnable.java @@ -20,17 +20,12 @@ package org.apache.dolphinscheduler.workflow.engine.workflow; /** * The TaskExecutionRunnable represent the running task, it is responsible for operate the task instance. e.g. dispatch, kill, pause. */ -public interface ITaskExecutionRunnable { +public interface ITaskExecutionRunnable extends IEventfulExecutionRunnable { /** - * Dispatch the task instance. + * Start the task instance. */ - void dispatch(); - - /** - * Run the task instance. - */ - void run(); + void start(); /** * Kill the task instance. @@ -42,6 +37,13 @@ public interface ITaskExecutionRunnable { */ void pause(); + /** + * Get the task execution identify. + * + * @return the task execution identify + */ + ITaskExecutionRunnableIdentify getIdentify(); + /** * Get the task execution context. * @@ -56,4 +58,5 @@ public interface ITaskExecutionRunnable { * @return true if the current task can be accessed to the post task. */ boolean isReadyToTrigger(String taskNodeName); + } diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/ITaskExecutionRunnableChain.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/ITaskExecutionRunnableChain.java new file mode 100644 index 0000000000..ce47fcac73 --- /dev/null +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/ITaskExecutionRunnableChain.java @@ -0,0 +1,26 @@ +/* + * 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.workflow.engine.workflow; + +public interface ITaskExecutionRunnableChain { + + ITaskExecutionRunnable getFrom(); + + ITaskExecutionRunnable getTo(); + +} diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/ITaskExecutionRunnableDelegate.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/ITaskExecutionRunnableDelegate.java new file mode 100644 index 0000000000..09e6408353 --- /dev/null +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/ITaskExecutionRunnableDelegate.java @@ -0,0 +1,34 @@ +/* + * 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.workflow.engine.workflow; + +public interface ITaskExecutionRunnableDelegate { + + void beforeStart(); + + void afterStart(); + + void beforePause(); + + void afterPause(); + + void beforeKill(); + + void afterKill(); + +} diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/ITaskExecutionRunnableDelegateFactory.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/ITaskExecutionRunnableDelegateFactory.java new file mode 100644 index 0000000000..4502b079a2 --- /dev/null +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/ITaskExecutionRunnableDelegateFactory.java @@ -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.workflow.engine.workflow; + +import org.apache.dolphinscheduler.workflow.engine.dag.ITaskIdentify; + +public interface ITaskExecutionRunnableDelegateFactory { + + ITaskExecutionRunnableDelegate createTaskExecutionRunnable(ITaskIdentify taskIdentify, + IWorkflowExecutionContext workflowExecutionContext); + + ITaskExecutionRunnableDelegate createFailoverTaskExecutionRunnable(ITaskExecutionRunnable taskExecutionRunnable, + IWorkflowExecutionContext workflowExecutionContext); + + ITaskExecutionRunnableDelegate createRetryTaskExecutionRunnable(ITaskExecutionRunnable taskExecutionRunnable, + IWorkflowExecutionContext workflowExecutionContext); + +} diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/ITaskExecutionRunnableFactory.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/ITaskExecutionRunnableFactory.java index b23364597b..8a4630e156 100644 --- a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/ITaskExecutionRunnableFactory.java +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/ITaskExecutionRunnableFactory.java @@ -17,9 +17,11 @@ package org.apache.dolphinscheduler.workflow.engine.workflow; +import org.apache.dolphinscheduler.workflow.engine.dag.ITaskIdentify; + public interface ITaskExecutionRunnableFactory { - ITaskExecutionRunnable createTaskExecutionRunnable(String taskName, + ITaskExecutionRunnable createTaskExecutionRunnable(ITaskIdentify taskIdentify, IWorkflowExecutionContext workflowExecutionContext); ITaskExecutionRunnable createFailoverTaskExecutionRunnable(ITaskExecutionRunnable taskExecutionRunnable, diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/ITaskExecutionRunnableIdentify.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/ITaskExecutionRunnableIdentify.java new file mode 100644 index 0000000000..37c70d8036 --- /dev/null +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/ITaskExecutionRunnableIdentify.java @@ -0,0 +1,30 @@ +/* + * 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.workflow.engine.workflow; + +import org.apache.dolphinscheduler.workflow.engine.dag.TaskIdentify; + +public interface ITaskExecutionRunnableIdentify { + + Long getId(); + + String getName(); + + TaskIdentify getTaskIdentify(); + +} diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/IWorkflowExecutionContext.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/IWorkflowExecutionContext.java index 265076b4f1..3ecb75fb92 100644 --- a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/IWorkflowExecutionContext.java +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/IWorkflowExecutionContext.java @@ -17,22 +17,22 @@ package org.apache.dolphinscheduler.workflow.engine.workflow; +import org.apache.dolphinscheduler.workflow.engine.dag.ITaskIdentify; +import org.apache.dolphinscheduler.workflow.engine.dag.WorkflowDAG; import org.apache.dolphinscheduler.workflow.engine.event.IEventRepository; +import java.util.List; + public interface IWorkflowExecutionContext { - IWorkflowInstance getWorkflowInstance(); + IWorkflowExecutionRunnableIdentify getIdentify(); - IWorkflowExecutionDAG getWorkflowExecutionDAG(); + List getStartTaskIdentifies(); + + WorkflowDAG getWorkflowDAG(); + + WorkflowExecutionDAG getWorkflowExecutionDAG(); IEventRepository getEventRepository(); - default int getWorkflowInstanceId() { - return getWorkflowInstance().getId(); - } - - default String getWorkflowInstanceName() { - return getWorkflowInstance().getName(); - } - } diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/IWorkflowExecutionDAG.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/IWorkflowExecutionDAG.java index 81732181ae..f27327314c 100644 --- a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/IWorkflowExecutionDAG.java +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/IWorkflowExecutionDAG.java @@ -24,25 +24,7 @@ import java.util.List; /** * The WorkflowExecutionDAG represent the running workflow DAG. */ -public interface IWorkflowExecutionDAG { - - List getStartNodeNames(); - - /** - * Get TaskExecutionRunnable by given TaskInstanceId. - * - * @param taskInstanceId taskInstanceId. - * @return TaskExecutionRunnable - */ - ITaskExecutionRunnable getTaskExecutionRunnableById(Integer taskInstanceId); - - /** - * Get TaskExecutionRunnable by given taskName. - * - * @param taskName task name. - * @return TaskExecutionRunnable - */ - ITaskExecutionRunnable getTaskExecutionRunnableByName(String taskName); +public interface IWorkflowExecutionDAG extends DAG { /** * Get TaskExecutionRunnable which is not finished. @@ -51,21 +33,4 @@ public interface IWorkflowExecutionDAG { */ List getActiveTaskExecutionRunnable(); - /** - * Get the direct pre TaskExecutionRunnable of the given taskName. - * - * @param taskName task name. - * @return TaskExecutionRunnable - */ - List getDirectPreTaskExecutionRunnable(String taskName); - - /** - * Check whether the taskNode is ready to run. - * - * @param taskName taskNodeName - * @return true if the taskNode is ready to run. - */ - boolean isTaskAbleToBeTriggered(String taskName); - - void storeTaskExecutionRunnable(ITaskExecutionRunnable taskExecutionRunnable); } diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/IWorkflowExecutionRunnable.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/IWorkflowExecutionRunnable.java index 1c6e75ba1e..2d218730d1 100644 --- a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/IWorkflowExecutionRunnable.java +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/IWorkflowExecutionRunnable.java @@ -17,27 +17,21 @@ package org.apache.dolphinscheduler.workflow.engine.workflow; -import org.apache.dolphinscheduler.workflow.engine.engine.IDAGEngine; - /** * The IWorkflowExecuteRunnable represent a running workflow instance, it is responsible for operate the workflow instance. e.g. start, kill, pause. */ -public interface IWorkflowExecutionRunnable extends IEventfulExecutionRunnable { +public interface IWorkflowExecutionRunnable + extends + IWorkflowOuterAction, + IWorkflowInnerAction, + IEventfulExecutionRunnable { /** - * Start the workflow instance. + * Get the identity of the workflow execution runnable. + * + * @return the identity of the workflow execution runnable */ - void start(); - - /** - * Kill the workflow instance. - */ - void kill(); - - /** - * Pause the workflow instance. - */ - void pause(); + IWorkflowExecutionRunnableIdentify getIdentity(); /** * Get the workflow execution context. @@ -46,11 +40,4 @@ public interface IWorkflowExecutionRunnable extends IEventfulExecutionRunnable { */ IWorkflowExecutionContext getWorkflowExecutionContext(); - /** - * Get the {@link IDAGEngine} which used to execute the dag of the workflow instance. - * - * @return dag engine. - */ - IDAGEngine getDagEngine(); - } diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/IWorkflowExecutionRunnableDelegate.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/IWorkflowExecutionRunnableDelegate.java index 5eae17192e..84ebb64c05 100644 --- a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/IWorkflowExecutionRunnableDelegate.java +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/IWorkflowExecutionRunnableDelegate.java @@ -15,5 +15,14 @@ * limitations under the License. */ -package org.apache.dolphinscheduler.workflow.engine.workflow;public interface IWorkflowExecutionRunnableDelegate { +package org.apache.dolphinscheduler.workflow.engine.workflow; + +public interface IWorkflowExecutionRunnableDelegate { + + void start(); + + void pause(); + + void kill(); + } diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/IWorkflowExecutionRunnableDelegateFactory.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/IWorkflowExecutionRunnableDelegateFactory.java new file mode 100644 index 0000000000..14c17ad8b2 --- /dev/null +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/IWorkflowExecutionRunnableDelegateFactory.java @@ -0,0 +1,24 @@ +/* + * 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.workflow.engine.workflow; + +public interface IWorkflowExecutionRunnableDelegateFactory { + + IWorkflowExecutionRunnableDelegate createWorkflowExecutionRunnableDelegate(IWorkflowExecutionContext workflowExecutionContext); + +} diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/IWorkflowInstance.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/IWorkflowExecutionRunnableIdentify.java similarity index 92% rename from dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/IWorkflowInstance.java rename to dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/IWorkflowExecutionRunnableIdentify.java index 49944d248a..494de82abc 100644 --- a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/IWorkflowInstance.java +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/IWorkflowExecutionRunnableIdentify.java @@ -17,9 +17,9 @@ package org.apache.dolphinscheduler.workflow.engine.workflow; -public interface IWorkflowInstance { +public interface IWorkflowExecutionRunnableIdentify { - int getId(); + Long getId(); String getName(); diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/IWorkflowExecutionRunnableOperationCheck.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/IWorkflowExecutionRunnableOperationCheck.java new file mode 100644 index 0000000000..5e54bce1ee --- /dev/null +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/IWorkflowExecutionRunnableOperationCheck.java @@ -0,0 +1,28 @@ +/* + * 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.workflow.engine.workflow; + +public interface IWorkflowExecutionRunnableOperationCheck { + + boolean canStart(); + + boolean canPause(); + + boolean canKill(); + +} diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/IWorkflowExecutionRunnableRepository.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/IWorkflowExecutionRunnableRepository.java index 3996b02405..b2320765b6 100644 --- a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/IWorkflowExecutionRunnableRepository.java +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/IWorkflowExecutionRunnableRepository.java @@ -23,11 +23,11 @@ public interface IWorkflowExecutionRunnableRepository { void storeWorkflowExecutionRunnable(IWorkflowExecutionRunnable workflowExecutionRunnable); - IWorkflowExecutionRunnable getWorkflowExecutionRunnableById(Integer workflowInstanceId); + IWorkflowExecutionRunnable getWorkflowExecutionRunnable(IWorkflowExecutionRunnableIdentify workflowExecutionRunnableIdentify); Collection getActiveWorkflowExecutionRunnable(); - void removeWorkflowExecutionRunnable(Integer workflowInstanceId); + void removeWorkflowExecutionRunnable(IWorkflowExecutionRunnableIdentify workflowExecutionRunnableIdentify); void clear(); } diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/ITaskInstance.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/IWorkflowExecutionRunnableStateEventListener.java similarity index 92% rename from dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/ITaskInstance.java rename to dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/IWorkflowExecutionRunnableStateEventListener.java index 2e86662267..f1bd3e052b 100644 --- a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/ITaskInstance.java +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/IWorkflowExecutionRunnableStateEventListener.java @@ -17,10 +17,6 @@ package org.apache.dolphinscheduler.workflow.engine.workflow; -public interface ITaskInstance { - - int getId(); - - String getName(); +public interface IWorkflowExecutionRunnableStateEventListener { } diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/IWorkflowInnerAction.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/IWorkflowInnerAction.java new file mode 100644 index 0000000000..5c8705ff00 --- /dev/null +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/IWorkflowInnerAction.java @@ -0,0 +1,35 @@ +/* + * 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.workflow.engine.workflow; + +import org.apache.dolphinscheduler.workflow.engine.dag.ITaskIdentify; + +public interface IWorkflowInnerAction { + + void triggerNextTasks(ITaskIdentify taskIdentify); + + void triggerTask(ITaskIdentify taskIdentify); + + void failoverTask(ITaskExecutionRunnableIdentify taskExecutionRunnableIdentify); + + void retryTask(ITaskExecutionRunnableIdentify taskExecutionRunnableIdentify); + + void pauseTask(ITaskExecutionRunnableIdentify taskExecutionIdentify); + + void killTask(ITaskExecutionRunnableIdentify taskExecutionIdentify); +} diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/IWorkflowOuterAction.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/IWorkflowOuterAction.java new file mode 100644 index 0000000000..076c24df84 --- /dev/null +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/IWorkflowOuterAction.java @@ -0,0 +1,37 @@ +/* + * 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.workflow.engine.workflow; + +public interface IWorkflowOuterAction { + + /** + * Start the workflow instance, this method will trigger the start task. + */ + void start(); + + /** + * Kill the workflow instance. + */ + void kill(); + + /** + * Pause the workflow instance. + */ + void pause(); + +} diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/SingletonWorkflowExecutionRunnableRepository.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/SingletonWorkflowExecutionRunnableRepository.java index d28f9a4fb1..4b308ca6cb 100644 --- a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/SingletonWorkflowExecutionRunnableRepository.java +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/SingletonWorkflowExecutionRunnableRepository.java @@ -26,7 +26,7 @@ public class SingletonWorkflowExecutionRunnableRepository implements IWorkflowEx private static final IWorkflowExecutionRunnableRepository INSTANCE = new SingletonWorkflowExecutionRunnableRepository(); - private final Map workflowExecutionRunnableMap; + private final Map workflowExecutionRunnableMap; private SingletonWorkflowExecutionRunnableRepository() { this.workflowExecutionRunnableMap = new ConcurrentHashMap<>(); @@ -36,22 +36,23 @@ public class SingletonWorkflowExecutionRunnableRepository implements IWorkflowEx return INSTANCE; } + @Override public void storeWorkflowExecutionRunnable(IWorkflowExecutionRunnable workflowExecutionRunnable) { - workflowExecutionRunnableMap.put( - workflowExecutionRunnable.getWorkflowExecutionContext().getWorkflowInstanceId(), - workflowExecutionRunnable); + workflowExecutionRunnableMap.put(workflowExecutionRunnable.getIdentity(), workflowExecutionRunnable); } - public IWorkflowExecutionRunnable getWorkflowExecutionRunnableById(Integer workflowInstanceId) { - return workflowExecutionRunnableMap.get(workflowInstanceId); + @Override + public IWorkflowExecutionRunnable getWorkflowExecutionRunnable(IWorkflowExecutionRunnableIdentify workflowExecutionRunnableIdentify) { + return workflowExecutionRunnableMap.get(workflowExecutionRunnableIdentify); } public Collection getActiveWorkflowExecutionRunnable() { return workflowExecutionRunnableMap.values(); } - public void removeWorkflowExecutionRunnable(Integer workflowInstanceId) { - workflowExecutionRunnableMap.remove(workflowInstanceId); + @Override + public void removeWorkflowExecutionRunnable(IWorkflowExecutionRunnableIdentify workflowExecutionRunnableIdentify) { + workflowExecutionRunnableMap.remove(workflowExecutionRunnableIdentify); } @Override diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/TaskExecutionContext.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/TaskExecutionContext.java index 380ed8b1ee..2f8b74a3f7 100644 --- a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/TaskExecutionContext.java +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/TaskExecutionContext.java @@ -19,14 +19,4 @@ package org.apache.dolphinscheduler.workflow.engine.workflow; public class TaskExecutionContext implements ITaskExecutionContext { - private final ITaskInstance taskInstance; - - public TaskExecutionContext(ITaskInstance taskInstance) { - this.taskInstance = taskInstance; - } - - @Override - public ITaskInstance getTaskInstance() { - return taskInstance; - } } diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/TaskExecutionPlan.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/TaskExecutionPlan.java new file mode 100644 index 0000000000..f176d70ea3 --- /dev/null +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/TaskExecutionPlan.java @@ -0,0 +1,113 @@ +/* + * 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.workflow.engine.workflow; + +import org.apache.dolphinscheduler.workflow.engine.dag.ITask; +import org.apache.dolphinscheduler.workflow.engine.dag.ITaskIdentify; + +import java.util.List; +import java.util.Map; + +public class TaskExecutionPlan implements ITaskExecutionPlan { + + private ITaskIdentify taskIdentify; + + private ITask task; + + private IWorkflowExecutionContext workflowExecutionContext; + + private ITaskExecutionRunnableFactory taskExecutionRunnableFactory; + + private Map taskExecutionRunnableMap; + + private ITaskExecutionRunnable activeTaskExecutionRunnable; + + @Override + public void start() { + ITaskExecutionRunnable taskExecutionRunnable = + taskExecutionRunnableFactory.createTaskExecutionRunnable(taskIdentify, workflowExecutionContext); + taskExecutionRunnableMap.put(taskExecutionRunnable.getIdentify(), taskExecutionRunnable); + activeTaskExecutionRunnable = taskExecutionRunnable; + taskExecutionRunnable.start(); + } + + @Override + public void failoverTask() { + // todo: check if the task can takeover + ITaskExecutionRunnable failoverTaskExecutionRunnable = taskExecutionRunnableFactory + .createFailoverTaskExecutionRunnable(activeTaskExecutionRunnable, workflowExecutionContext); + taskExecutionRunnableMap.put(failoverTaskExecutionRunnable.getIdentify(), failoverTaskExecutionRunnable); + activeTaskExecutionRunnable = failoverTaskExecutionRunnable; + failoverTaskExecutionRunnable.start(); + } + + @Override + public void retryTask() { + // check the retry times + ITaskExecutionRunnable retryTaskExecutionRunnable = taskExecutionRunnableFactory + .createRetryTaskExecutionRunnable(activeTaskExecutionRunnable, workflowExecutionContext); + taskExecutionRunnableMap.put(retryTaskExecutionRunnable.getIdentify(), retryTaskExecutionRunnable); + activeTaskExecutionRunnable = retryTaskExecutionRunnable; + + retryTaskExecutionRunnable.start(); + } + + @Override + public void pauseTask() { + activeTaskExecutionRunnable.pause(); + } + + @Override + public void killTask() { + activeTaskExecutionRunnable.kill(); + } + + @Override + public ITaskIdentify getTaskIdentify() { + return taskIdentify; + } + + @Override + public ITaskExecutionRunnable getActiveTaskExecutionRunnable() { + return activeTaskExecutionRunnable; + } + + @Override + public ITask getTask() { + return task; + } + + @Override + public List getTaskExecutionRunnableList() { + return null; + } + + @Override + public ITaskExecutionRunnable getTaskExecutionRunnable(ITaskExecutionRunnableIdentify taskExecutionRunnableIdentify) { + return taskExecutionRunnableMap.get(taskExecutionRunnableIdentify); + } + + @Override + public ITaskExecutionRunnable storeTaskExecutionRunnable(ITaskExecutionRunnable taskExecutionRunnable) { + if (taskExecutionRunnable == null) { + throw new IllegalArgumentException("taskExecutionRunnable cannot be null"); + } + return taskExecutionRunnableMap.put(taskExecutionRunnable.getIdentify(), taskExecutionRunnable); + } + +} diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/TaskExecutionRunnable.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/TaskExecutionRunnable.java index 322cd8ee8d..2f10ad759d 100644 --- a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/TaskExecutionRunnable.java +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/TaskExecutionRunnable.java @@ -17,41 +17,59 @@ package org.apache.dolphinscheduler.workflow.engine.workflow; +import org.apache.dolphinscheduler.workflow.engine.event.IEventRepository; + public class TaskExecutionRunnable implements ITaskExecutionRunnable { - public final ITaskExecutionContext taskExecutionContext; + private final ITaskExecutionContext taskExecutionContext; - public TaskExecutionRunnable(ITaskExecutionContext taskExecutionContext) { + private final ITaskExecutionRunnableDelegate taskExecutionRunnableDelegate; + + public TaskExecutionRunnable(ITaskExecutionContext taskExecutionContext, + ITaskExecutionRunnableDelegate taskExecutionRunnableDelegate) { this.taskExecutionContext = taskExecutionContext; + this.taskExecutionRunnableDelegate = taskExecutionRunnableDelegate; } @Override - public void dispatch() { + public void start() { + taskExecutionRunnableDelegate.beforeStart(); - } - - @Override - public void run() { - - } - - @Override - public void kill() { + taskExecutionRunnableDelegate.afterStart(); } @Override public void pause() { + taskExecutionRunnableDelegate.beforePause(); + taskExecutionRunnableDelegate.afterPause(); } + @Override + public void kill() { + taskExecutionRunnableDelegate.beforeKill(); + taskExecutionRunnableDelegate.afterKill(); + + } + + @Override + public boolean isReadyToTrigger(String taskNodeName) { + return false; + } + + @Override + public ITaskExecutionRunnableIdentify getIdentify() { + return taskExecutionContext.getIdentify(); + } + @Override public ITaskExecutionContext getTaskExecutionContext() { return taskExecutionContext; } @Override - public boolean isReadyToTrigger(String taskNodeName) { - return false; + public IEventRepository getEventRepository() { + return taskExecutionContext.getEventRepository(); } } diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/TaskExecutionRunnableFactory.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/TaskExecutionRunnableFactory.java index 9b55c91639..f9c7028103 100644 --- a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/TaskExecutionRunnableFactory.java +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/TaskExecutionRunnableFactory.java @@ -17,20 +17,32 @@ package org.apache.dolphinscheduler.workflow.engine.workflow; +import org.apache.dolphinscheduler.workflow.engine.dag.ITaskIdentify; + public class TaskExecutionRunnableFactory implements ITaskExecutionRunnableFactory { private final ITaskExecutionContextFactory taskExecutionContextFactory; + private ITaskExecutionRunnableDelegateFactory taskExecutionRunnableDelegateFactory = + DefaultTaskExecutionRunnableDelegateFactory.getInstance(); + public TaskExecutionRunnableFactory(ITaskExecutionContextFactory taskExecutionContextFactory) { this.taskExecutionContextFactory = taskExecutionContextFactory; } + public TaskExecutionRunnableFactory withTaskExecutionRunnableDelegateFactory(ITaskExecutionRunnableDelegateFactory taskExecutionRunnableDelegateFactory) { + this.taskExecutionRunnableDelegateFactory = taskExecutionRunnableDelegateFactory; + return this; + } + @Override - public ITaskExecutionRunnable createTaskExecutionRunnable(String taskName, + public ITaskExecutionRunnable createTaskExecutionRunnable(ITaskIdentify taskIdentify, IWorkflowExecutionContext workflowExecutionContext) { ITaskExecutionContext taskExecutionContext = - taskExecutionContextFactory.createTaskExecutionContext(taskName, workflowExecutionContext); - return new TaskExecutionRunnable(taskExecutionContext); + taskExecutionContextFactory.createTaskExecutionContext(taskIdentify, workflowExecutionContext); + ITaskExecutionRunnableDelegate taskExecutionRunnableDelegate = taskExecutionRunnableDelegateFactory + .createTaskExecutionRunnable(taskIdentify, workflowExecutionContext); + return new TaskExecutionRunnable(taskExecutionContext, taskExecutionRunnableDelegate); } @Override diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/TaskExecutionRunnableIdentify.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/TaskExecutionRunnableIdentify.java new file mode 100644 index 0000000000..8a5fbf44b2 --- /dev/null +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/TaskExecutionRunnableIdentify.java @@ -0,0 +1,53 @@ +/* + * 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.workflow.engine.workflow; + +import org.apache.dolphinscheduler.workflow.engine.dag.TaskIdentify; + +import lombok.EqualsAndHashCode; + +@EqualsAndHashCode +public class TaskExecutionRunnableIdentify implements ITaskExecutionRunnableIdentify { + + private final Long id; + + private final String name; + + private final TaskIdentify taskIdentify; + + public TaskExecutionRunnableIdentify(Long id, String name, TaskIdentify taskIdentify) { + this.id = id; + this.name = name; + this.taskIdentify = taskIdentify; + } + + @Override + public Long getId() { + return id; + } + + @Override + public String getName() { + return name; + } + + @Override + public TaskIdentify getTaskIdentify() { + return taskIdentify; + } +} diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/WorkflowExecutionContext.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/WorkflowExecutionContext.java index 6cca1eced1..129cd3d624 100644 --- a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/WorkflowExecutionContext.java +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/WorkflowExecutionContext.java @@ -17,6 +17,7 @@ package org.apache.dolphinscheduler.workflow.engine.workflow; +import org.apache.dolphinscheduler.workflow.engine.dag.WorkflowDAG; import org.apache.dolphinscheduler.workflow.engine.event.IEventRepository; import lombok.AllArgsConstructor; @@ -30,10 +31,23 @@ import lombok.NoArgsConstructor; @NoArgsConstructor public class WorkflowExecutionContext implements IWorkflowExecutionContext { - private IWorkflowInstance workflowInstance; + @Override + public IWorkflowExecutionRunnableIdentify getIdentify() { + return null; + } - private IWorkflowExecutionDAG workflowExecutionDAG; + @Override + public WorkflowDAG getWorkflowDAG() { + return null; + } - private IEventRepository eventRepository; + @Override + public WorkflowExecutionDAG getWorkflowExecutionDAG() { + return null; + } + @Override + public IEventRepository getEventRepository() { + return null; + } } diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/WorkflowExecutionDAG.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/WorkflowExecutionDAG.java index 6bcd123b2a..daea7c6beb 100644 --- a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/WorkflowExecutionDAG.java +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/WorkflowExecutionDAG.java @@ -17,93 +17,134 @@ package org.apache.dolphinscheduler.workflow.engine.workflow; -import org.apache.dolphinscheduler.workflow.engine.dag.Node; -import org.apache.dolphinscheduler.workflow.engine.dag.WorkflowDAG; +import org.apache.dolphinscheduler.workflow.engine.dag.DAG; +import org.apache.dolphinscheduler.workflow.engine.dag.ITaskIdentify; +import org.apache.dolphinscheduler.workflow.engine.utils.IWorkflowExecutionDAGStatusCheck; import java.util.ArrayList; import java.util.Collections; +import java.util.HashMap; import java.util.List; +import java.util.Map; import java.util.stream.Collectors; -import lombok.Getter; -import lombok.extern.slf4j.Slf4j; +public class WorkflowExecutionDAG implements IWorkflowExecutionDAGStatusCheck, DAG { -/** - * The WorkflowExecutionDAG represent a running workflow instance DAG. - */ -@Slf4j -public class WorkflowExecutionDAG implements IWorkflowExecutionDAG { + private final Map taskExecutionPlanMap; - private final ITaskExecutionRunnableRepository taskExecutionRunnableRepository; + private final Map> outdegreeMap; - private final WorkflowDAG workflowDAG; + private final Map> inDegredMap; - @Getter - private final List startNodeNames; + public WorkflowExecutionDAG(List tasks, + List taskChains) { + this.taskExecutionPlanMap = new HashMap<>(); + this.outdegreeMap = new HashMap<>(); + this.inDegredMap = new HashMap<>(); - public WorkflowExecutionDAG(ITaskExecutionRunnableRepository taskExecutionRunnableRepository, - WorkflowDAG workflowDAG) { - this(taskExecutionRunnableRepository, workflowDAG, Collections.emptyList()); - } - - public WorkflowExecutionDAG(ITaskExecutionRunnableRepository taskExecutionRunnableRepository, - WorkflowDAG workflowDAG, - List startNodeNames) { - this.taskExecutionRunnableRepository = taskExecutionRunnableRepository; - this.workflowDAG = workflowDAG; - this.startNodeNames = startNodeNames; + for (ITaskExecutionPlan task : tasks) { + ITaskIdentify identify = task.getTaskIdentify(); + if (taskExecutionPlanMap.containsKey(identify)) { + throw new IllegalArgumentException("Duplicate task identify: " + identify); + } + taskExecutionPlanMap.put(identify, task); + } + for (ITaskExecutionPlanChain taskChain : taskChains) { + ITaskExecutionPlan from = taskChain.getFrom(); + ITaskExecutionPlan to = taskChain.getTo(); + if (from == null) { + continue; + } + if (to == null) { + continue; + } + ITaskIdentify fromIdentify = from.getTaskIdentify(); + ITaskIdentify toIdentify = to.getTaskIdentify(); + List outDegrees = + outdegreeMap.computeIfAbsent(fromIdentify, k -> new ArrayList<>()); + if (outDegrees.contains(toIdentify)) { + throw new IllegalArgumentException("Duplicate task chain: " + fromIdentify + " -> " + toIdentify); + } + outDegrees.add(toIdentify); + List inDegrees = + inDegredMap.computeIfAbsent(toIdentify, k -> new ArrayList<>()); + if (inDegrees.contains(fromIdentify)) { + throw new IllegalArgumentException("Duplicate task chain: " + fromIdentify + " -> " + toIdentify); + } + inDegrees.add(fromIdentify); + } } @Override - public ITaskExecutionRunnable getTaskExecutionRunnableById(Integer taskInstanceId) { - return taskExecutionRunnableRepository.getTaskExecutionRunnableById(taskInstanceId); + public List getDirectPostNodes(ITaskExecutionPlan taskExecutionPlan) { + if (taskExecutionPlan == null) { + return getDirectPostNodesByIdentify(null); + } + return getDirectPostNodesByIdentify(taskExecutionPlan.getTaskIdentify()); } @Override - public ITaskExecutionRunnable getTaskExecutionRunnableByName(String taskName) { - return taskExecutionRunnableRepository.getTaskExecutionRunnableByName(taskName); - } - - @Override - public List getActiveTaskExecutionRunnable() { - return new ArrayList<>(taskExecutionRunnableRepository.getActiveTaskExecutionRunnable()); - } - - @Override - public List getDirectPreTaskExecutionRunnable(String taskName) { - return getDirectPreNodeNames(taskName) + public List getDirectPostNodesByIdentify(ITaskIdentify taskIdentify) { + if (taskIdentify == null) { + return taskExecutionPlanMap.values() + .stream() + .filter(task -> !inDegredMap.containsKey(task.getTaskIdentify())) + .collect(Collectors.toList()); + } + return inDegredMap.getOrDefault(taskIdentify, Collections.emptyList()) .stream() - .map(taskExecutionRunnableRepository::getTaskExecutionRunnableByName) + .map(taskExecutionPlanMap::get) .collect(Collectors.toList()); } @Override - public boolean isTaskAbleToBeTriggered(String taskNodeName) { - // todo: Check whether the workflow instance is finished or ready to finish. - List directPreNodes = getDirectPreNodes(taskNodeName); - if (log.isDebugEnabled()) { - log.debug("Begin to check whether the task {} is able to be triggered.", taskNodeName); - log.debug("Task {} directly dependent on the task: {}.", taskNodeName, - directPreNodes.stream().map(Node::getNodeName).collect(Collectors.toList())); + public List getDirectPreNodes(ITaskExecutionPlan iTaskExecutionPlan) { + if (iTaskExecutionPlan == null) { + return getDirectPreNodesByIdentify(null); } - for (Node directPreNode : directPreNodes) { - if (directPreNode.isSkip()) { - log.debug("The task {} is skipped.", directPreNode.getNodeName()); - continue; - } - ITaskExecutionRunnable taskExecutionRunnable = getTaskExecutionRunnableByName(directPreNode.getNodeName()); - if (taskExecutionRunnable == null || taskExecutionRunnable.isReadyToTrigger(taskNodeName)) { - log.debug("The task {} is not finished or not able to access to the task {}.", - directPreNode.getNodeName(), taskNodeName); - } - } - return true; + return getDirectPreNodesByIdentify(iTaskExecutionPlan.getTaskIdentify()); } @Override - public void storeTaskExecutionRunnable(ITaskExecutionRunnable taskExecutionRunnable) { - taskExecutionRunnableRepository.storeTaskExecutionRunnable(taskExecutionRunnable); + public List getDirectPreNodesByIdentify(ITaskIdentify taskIdentify) { + if (taskIdentify == null) { + return taskExecutionPlanMap.values() + .stream() + .filter(task -> !outdegreeMap.containsKey(task.getTaskIdentify())) + .collect(Collectors.toList()); + } + return outdegreeMap.getOrDefault(taskIdentify, Collections.emptyList()) + .stream() + .map(taskExecutionPlanMap::get) + .collect(Collectors.toList()); } + public List getActiveTaskExecutionPlan() { + return null; + } + @Override + public ITaskExecutionPlan getDAGNode(ITaskIdentify taskIdentify) { + return taskExecutionPlanMap.get(taskIdentify); + } + + @Override + public boolean isSuccess() { + return false; + } + + @Override + public boolean isFailed() { + return false; + } + + @Override + public boolean isKilled() { + return false; + } + + @Override + public boolean isPaused() { + return false; + } } diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/WorkflowExecutionDAGFactory.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/WorkflowExecutionDAGFactory.java new file mode 100644 index 0000000000..9768ba8607 --- /dev/null +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/WorkflowExecutionDAGFactory.java @@ -0,0 +1,26 @@ +/* + * 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.workflow.engine.workflow; + +public class WorkflowExecutionDAGFactory { + + public WorkflowExecutionDAG createDAG() { + return new WorkflowExecutionDAG(null, null); + } + +} diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/WorkflowExecutionRunnable.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/WorkflowExecutionRunnable.java index cf609b1f15..ac8ce451c9 100644 --- a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/WorkflowExecutionRunnable.java +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/WorkflowExecutionRunnable.java @@ -17,60 +17,169 @@ package org.apache.dolphinscheduler.workflow.engine.workflow; -import org.apache.dolphinscheduler.workflow.engine.engine.IDAGEngine; -import org.apache.dolphinscheduler.workflow.engine.event.IEventRepository; +import org.apache.dolphinscheduler.workflow.engine.dag.ITask; +import org.apache.dolphinscheduler.workflow.engine.dag.ITaskIdentify; +import org.apache.dolphinscheduler.workflow.engine.dag.WorkflowDAG; +import org.apache.dolphinscheduler.workflow.engine.event.TaskOperationEvent; +import org.apache.dolphinscheduler.workflow.engine.event.WorkflowFinishEvent; import org.apache.commons.collections4.CollectionUtils; import java.util.List; +import java.util.stream.Collectors; import lombok.Getter; @Getter -public class WorkflowExecutionRunnable implements IWorkflowExecutionRunnable { +public class WorkflowExecutionRunnable extends BaseWorkflowExecutionRunnable { - private final IWorkflowExecutionContext workflowExecutionContext; + private final WorkflowExecutionDAG workflowExecutionDAG; + private final WorkflowDAG workflowDAG; - private final IDAGEngine dagEngine; + private final IWorkflowExecutionRunnableDelegate workflowExecutionRunnableDelegate; - private volatile boolean eventFiring = false; - - public WorkflowExecutionRunnable(IWorkflowExecutionContext workflowExecutionContext, IDAGEngine dagEngine) { - this.workflowExecutionContext = workflowExecutionContext; - this.dagEngine = dagEngine; + public WorkflowExecutionRunnable(IWorkflowExecutionContext workflowExecutionContext, + IWorkflowExecutionRunnableDelegate workflowExecutionRunnableDelegate) { + super(workflowExecutionContext, WorkflowExecutionRunnableStatus.CREATED); + this.workflowExecutionDAG = workflowExecutionContext.getWorkflowExecutionDAG(); + this.workflowDAG = workflowExecutionContext.getWorkflowDAG(); + this.workflowExecutionRunnableDelegate = workflowExecutionRunnableDelegate; } + @Override public void start() { - List workflowStartNodeNames = workflowExecutionContext.getWorkflowExecutionDAG().getStartNodeNames(); - if (CollectionUtils.isEmpty(workflowStartNodeNames)) { - dagEngine.triggerNextTasks(null); - } else { - workflowStartNodeNames.forEach(dagEngine::triggerTask); + if (!workflowExecutionRunnableStatus.canStart()) { + throw new UnsupportedOperationException( + "The current status: " + workflowExecutionRunnableStatus + " cannot start."); } + + statusTransform(WorkflowExecutionRunnableStatus.RUNNING, () -> { + workflowExecutionRunnableDelegate.start(); + List startTaskIdentifies = workflowExecutionContext.getStartTaskIdentifies(); + // If the start task is empty, trigger from the beginning + if (CollectionUtils.isEmpty(startTaskIdentifies)) { + workflowFinish(); + return; + } + startTaskIdentifies.forEach(this::triggerTask); + }); + } + + @Override + public void triggerNextTasks(ITaskIdentify taskIdentify) { + List directPostNodeIdentifies = workflowDAG.getDirectPostNodesByIdentify(taskIdentify) + .stream() + .map(ITask::getIdentify) + .collect(Collectors.toList()); + if (CollectionUtils.isNotEmpty(directPostNodeIdentifies)) { + directPostNodeIdentifies.forEach(this::triggerTask); + return; + } + List activeTaskExecutionIdentify = getActiveTaskExecutionIdentify(); + if (CollectionUtils.isEmpty(activeTaskExecutionIdentify)) { + workflowFinish(); + return; + } + // The task chain is finished, but there are still active tasks, wait for the active tasks to finish + } + + @Override + public void triggerTask(ITaskIdentify taskIdentify) { + ITaskExecutionPlan taskExecutionPlan = workflowExecutionDAG.getDAGNode(taskIdentify); + if (taskExecutionPlan == null) { + throw new IllegalArgumentException("Cannot find the ITaskExecutionPlan for taskIdentify: " + taskIdentify); + } + getEventRepository().storeEventToTail(TaskOperationEvent.startEvent(taskExecutionPlan)); + } + + @Override + public void failoverTask(ITaskExecutionRunnableIdentify taskExecutionRunnableIdentify) { + ITaskExecutionPlan taskExecutionPlan = + workflowExecutionDAG.getDAGNode(taskExecutionRunnableIdentify.getTaskIdentify()); + if (taskExecutionPlan == null) { + throw new IllegalArgumentException("Cannot find the ITaskExecutionPlan for taskIdentify: " + + taskExecutionRunnableIdentify.getTaskIdentify()); + } + getEventRepository().storeEventToTail(TaskOperationEvent.failoverEvent(taskExecutionPlan)); + } + + @Override + public void retryTask(ITaskExecutionRunnableIdentify taskExecutionRunnableIdentify) { + ITaskExecutionPlan taskExecutionPlan = + workflowExecutionDAG.getDAGNode(taskExecutionRunnableIdentify.getTaskIdentify()); + if (taskExecutionPlan == null) { + throw new IllegalArgumentException("Cannot find the ITaskExecutionPlan for taskIdentify: " + + taskExecutionRunnableIdentify.getTaskIdentify()); + } + getEventRepository().storeEventToTail(TaskOperationEvent.retryEvent(taskExecutionPlan)); } @Override public void pause() { - dagEngine.pauseAllTask(); + if (workflowExecutionRunnableStatus.canPause()) { + throw new UnsupportedOperationException( + "The current status: " + workflowExecutionRunnableStatus + " cannot pause."); + } + statusTransform(WorkflowExecutionRunnableStatus.PAUSING, () -> { + workflowExecutionRunnableDelegate.pause(); + List activeTaskExecutionIdentify = getActiveTaskExecutionIdentify(); + if (CollectionUtils.isEmpty(activeTaskExecutionIdentify)) { + workflowFinish(); + return; + } + activeTaskExecutionIdentify.forEach(this::pauseTask); + }); + } + + @Override + public void pauseTask(ITaskExecutionRunnableIdentify taskExecutionIdentify) { + ITaskExecutionPlan taskExecutionPlan = workflowExecutionDAG.getDAGNode(taskExecutionIdentify.getTaskIdentify()); + if (taskExecutionPlan == null) { + throw new IllegalArgumentException( + "Cannot find the ITaskExecutionPlan for taskIdentify: " + taskExecutionIdentify.getTaskIdentify()); + } + getEventRepository().storeEventToTail(TaskOperationEvent.pauseEvent(taskExecutionPlan)); } @Override public void kill() { - dagEngine.killAllTask(); + if (workflowExecutionRunnableStatus.canKill()) { + throw new UnsupportedOperationException( + "The current status: " + workflowExecutionRunnableStatus + " cannot kill."); + } + statusTransform(WorkflowExecutionRunnableStatus.KILLING, () -> { + workflowExecutionRunnableDelegate.kill(); + List activeTaskExecutionIdentify = getActiveTaskExecutionIdentify(); + if (CollectionUtils.isEmpty(activeTaskExecutionIdentify)) { + workflowFinish(); + return; + } + activeTaskExecutionIdentify.forEach(this::killTask); + }); } @Override - public IEventRepository getEventRepository() { - return workflowExecutionContext.getEventRepository(); + public void killTask(ITaskExecutionRunnableIdentify taskExecutionIdentify) { + ITaskExecutionPlan taskExecutionPlan = workflowExecutionDAG.getDAGNode(taskExecutionIdentify.getTaskIdentify()); + if (taskExecutionPlan == null) { + throw new IllegalArgumentException( + "Cannot find the ITaskExecutionPlan for taskIdentify: " + taskExecutionIdentify.getTaskIdentify()); + } + getEventRepository().storeEventToTail(TaskOperationEvent.killEvent(taskExecutionPlan)); } - @Override - public boolean isEventFiring() { - return eventFiring; + private void workflowFinish() { + if (workflowExecutionDAG.isFailed()) { + + } + getEventRepository().storeEventToTail(WorkflowFinishEvent.of(workflowExecutionRunnableIdentify)); } - @Override - public void setEventFiring(boolean eventFiring) { - this.eventFiring = eventFiring; + private List getActiveTaskExecutionIdentify() { + return workflowExecutionDAG.getActiveTaskExecutionPlan() + .stream() + .map(ITaskExecutionPlan::getActiveTaskExecutionRunnable) + .map(ITaskExecutionRunnable::getIdentify) + .collect(Collectors.toList()); } } diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/WorkflowExecutionRunnableFactory.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/WorkflowExecutionRunnableFactory.java index ebfb3510db..d24c0308c8 100644 --- a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/WorkflowExecutionRunnableFactory.java +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/WorkflowExecutionRunnableFactory.java @@ -24,13 +24,23 @@ public class WorkflowExecutionRunnableFactory implements IWorkflowExecutionRunna private final IDAGEngineFactory dagEngineFactory; + private IWorkflowExecutionRunnableDelegateFactory workflowExecutionRunnableDelegateFactory = + DefaultWorkflowExecutionRunnableDelegateFactory.getInstance(); + public WorkflowExecutionRunnableFactory(IDAGEngineFactory dagEngineFactory) { this.dagEngineFactory = dagEngineFactory; } + public WorkflowExecutionRunnableFactory withWorkflowExecutionRunnableDelegateFactory(IWorkflowExecutionRunnableDelegateFactory workflowExecutionRunnableDelegateFactory) { + this.workflowExecutionRunnableDelegateFactory = workflowExecutionRunnableDelegateFactory; + return this; + } + @Override public WorkflowExecutionRunnable createWorkflowExecutionRunnable(IWorkflowExecutionContext workflowExecutionContext) { IDAGEngine dagEngine = dagEngineFactory.createDAGEngine(workflowExecutionContext); - return new WorkflowExecutionRunnable(workflowExecutionContext, dagEngine); + IWorkflowExecutionRunnableDelegate workflowExecutionRunnableDelegate = workflowExecutionRunnableDelegateFactory + .createWorkflowExecutionRunnableDelegate(workflowExecutionContext); + return new WorkflowExecutionRunnable(workflowExecutionContext, dagEngine, workflowExecutionRunnableDelegate); } } diff --git a/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/WorkflowExecutionRunnableStatus.java b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/WorkflowExecutionRunnableStatus.java new file mode 100644 index 0000000000..f03f73f3df --- /dev/null +++ b/dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/WorkflowExecutionRunnableStatus.java @@ -0,0 +1,175 @@ +/* + * 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.workflow.engine.workflow; + +public enum WorkflowExecutionRunnableStatus implements IWorkflowExecutionRunnableOperationCheck { + + CREATED { + + @Override + public boolean canStart() { + return true; + } + + @Override + public boolean canPause() { + return true; + } + + @Override + public boolean canKill() { + return true; + } + + }, + RUNNING { + + @Override + public boolean canStart() { + throw new UnsupportedOperationException("canStart is not supported for " + this); + } + + @Override + public boolean canPause() { + throw new UnsupportedOperationException("canStart is not supported for " + this); + } + + @Override + public boolean canKill() { + throw new UnsupportedOperationException("canStart is not supported for " + this); + } + }, + PAUSING { + + @Override + public boolean canStart() { + throw new UnsupportedOperationException("canStart is not supported for " + this); + } + + @Override + public boolean canPause() { + throw new UnsupportedOperationException("canStart is not supported for " + this); + } + + @Override + public boolean canKill() { + throw new UnsupportedOperationException("canStart is not supported for " + this); + } + }, + PAUSED { + + @Override + public boolean canStart() { + throw new UnsupportedOperationException("canStart is not supported for " + this); + } + + @Override + public boolean canPause() { + throw new UnsupportedOperationException("canStart is not supported for " + this); + } + + @Override + public boolean canKill() { + throw new UnsupportedOperationException("canStart is not supported for " + this); + } + }, + KILLING { + + @Override + public boolean canStart() { + throw new UnsupportedOperationException("canStart is not supported for " + this); + } + + @Override + public boolean canPause() { + throw new UnsupportedOperationException("canStart is not supported for " + this); + } + + @Override + public boolean canKill() { + throw new UnsupportedOperationException("canStart is not supported for " + this); + } + }, + KILLED { + + @Override + public boolean canStart() { + throw new UnsupportedOperationException("canStart is not supported for " + this); + } + + @Override + public boolean canPause() { + throw new UnsupportedOperationException("canStart is not supported for " + this); + } + + @Override + public boolean canKill() { + throw new UnsupportedOperationException("canStart is not supported for " + this); + } + }, + FAILED { + + @Override + public boolean canStart() { + throw new UnsupportedOperationException("canStart is not supported for " + this); + } + + @Override + public boolean canPause() { + throw new UnsupportedOperationException("canStart is not supported for " + this); + } + + @Override + public boolean canKill() { + throw new UnsupportedOperationException("canStart is not supported for " + this); + } + }, + SUCCEEDED { + + @Override + public boolean canStart() { + throw new UnsupportedOperationException("canStart is not supported for " + this); + } + + @Override + public boolean canPause() { + throw new UnsupportedOperationException("canStart is not supported for " + this); + } + + @Override + public boolean canKill() { + throw new UnsupportedOperationException("canStart is not supported for " + this); + } + }, + ; + + @Override + public boolean canStart() { + throw new UnsupportedOperationException("canStart is not supported for " + this); + } + + @Override + public boolean canPause() { + throw new UnsupportedOperationException("canStart is not supported for " + this); + } + + @Override + public boolean canKill() { + throw new UnsupportedOperationException("canStart is not supported for " + this); + } +} diff --git a/dolphinscheduler-workflow-engine/src/test/java/org/apache/dolphinscheduler/workflow/engine/assertions/SingletonWorkflowExecuteRunnableRepositoryAssertions.java b/dolphinscheduler-workflow-engine/src/test/java/org/apache/dolphinscheduler/workflow/engine/assertions/SingletonWorkflowExecuteRunnableRepositoryAssertions.java index 40ac09007c..3658a174ea 100644 --- a/dolphinscheduler-workflow-engine/src/test/java/org/apache/dolphinscheduler/workflow/engine/assertions/SingletonWorkflowExecuteRunnableRepositoryAssertions.java +++ b/dolphinscheduler-workflow-engine/src/test/java/org/apache/dolphinscheduler/workflow/engine/assertions/SingletonWorkflowExecuteRunnableRepositoryAssertions.java @@ -26,12 +26,12 @@ public class SingletonWorkflowExecuteRunnableRepositoryAssertions { public static void existWorkflowExecutionRunnable(Integer workflowInstanceId) { assertNotNull(SingletonWorkflowExecutionRunnableRepository.getInstance() - .getWorkflowExecutionRunnableById(workflowInstanceId)); + .getWorkflowExecutionRunnable(workflowInstanceId)); } public static void notExistWorkflowExecutionRunnable(Integer workflowInstanceId) { assertNull(SingletonWorkflowExecutionRunnableRepository.getInstance() - .getWorkflowExecutionRunnableById(workflowInstanceId)); + .getWorkflowExecutionRunnable(workflowInstanceId)); } } diff --git a/dolphinscheduler-workflow-engine/src/test/java/org/apache/dolphinscheduler/workflow/engine/engine/WorkflowEngineTest.java b/dolphinscheduler-workflow-engine/src/test/java/org/apache/dolphinscheduler/workflow/engine/engine/WorkflowEngineTest.java index 482299227f..70f1aadf27 100644 --- a/dolphinscheduler-workflow-engine/src/test/java/org/apache/dolphinscheduler/workflow/engine/engine/WorkflowEngineTest.java +++ b/dolphinscheduler-workflow-engine/src/test/java/org/apache/dolphinscheduler/workflow/engine/engine/WorkflowEngineTest.java @@ -70,7 +70,7 @@ class WorkflowEngineTest { void killWorkflow_WorkflowNotExist() { WorkflowExecuteRunnableNotFoundException exception = assertThrows(WorkflowExecuteRunnableNotFoundException.class, - () -> workflowEngine.killWorkflow(1)); + () -> workflowEngine.killWorkflow(1L)); assertEquals("WorkflowExecuteRunnable not found: [id=1]", exception.getMessage()); } @@ -78,8 +78,8 @@ class WorkflowEngineTest { void killWorkflow_WorkflowExist() { IWorkflowExecutionRunnable mockWorkflowExecuteRunnable = MockWorkflowExecutionRunnableFactory.createWorkflowExecutionRunnable(); - Integer workflowInstanceId = - mockWorkflowExecuteRunnable.getWorkflowExecutionContext().getWorkflowInstanceId(); + Long workflowInstanceId = + mockWorkflowExecuteRunnable.getWorkflowExecutionContext().getIdentify().getId(); SingletonWorkflowExecutionRunnableRepository.getInstance() .storeWorkflowExecutionRunnable(mockWorkflowExecuteRunnable); @@ -90,15 +90,15 @@ class WorkflowEngineTest { @Test void finalizeWorkflow_WorkflowNotExist() { - workflowEngine.finalizeWorkflow(-1); + workflowEngine.finalizeWorkflow(-1L); } @Test void finalizeWorkflow_WorkflowExist() { IWorkflowExecutionRunnable emptyWorkflowExecuteRunnable = MockWorkflowExecutionRunnableFactory.createWorkflowExecutionRunnable(); - Integer workflowInstanceId = - emptyWorkflowExecuteRunnable.getWorkflowExecutionContext().getWorkflowInstanceId(); + Long workflowInstanceId = + emptyWorkflowExecuteRunnable.getWorkflowExecutionContext().getIdentify().getId(); SingletonWorkflowExecutionRunnableRepository.getInstance() .storeWorkflowExecutionRunnable(emptyWorkflowExecuteRunnable); SingletonWorkflowExecuteRunnableRepositoryAssertions.existWorkflowExecutionRunnable(workflowInstanceId); diff --git a/dolphinscheduler-workflow-engine/src/test/java/org/apache/dolphinscheduler/workflow/engine/event/WorkflowOperationEventTest.java b/dolphinscheduler-workflow-engine/src/test/java/org/apache/dolphinscheduler/workflow/engine/event/WorkflowOperationEventTest.java index 8356fbb1ac..d5952d4a09 100644 --- a/dolphinscheduler-workflow-engine/src/test/java/org/apache/dolphinscheduler/workflow/engine/event/WorkflowOperationEventTest.java +++ b/dolphinscheduler-workflow-engine/src/test/java/org/apache/dolphinscheduler/workflow/engine/event/WorkflowOperationEventTest.java @@ -10,21 +10,21 @@ class WorkflowOperationEventTest { void triggerEvent() { WorkflowOperationEvent workflowOperationEvent = WorkflowOperationEvent.triggerEvent(1); assertEquals(1, workflowOperationEvent.getWorkflowInstanceId()); - assertEquals(WorkflowOperationType.TRIGGER, workflowOperationEvent.getWorkflowOperationType()); + assertEquals(WorkflowOperationEventType.TRIGGER, workflowOperationEvent.getWorkflowOperationType()); } @Test void pauseEvent() { WorkflowOperationEvent workflowOperationEvent = WorkflowOperationEvent.pauseEvent(1); assertEquals(1, workflowOperationEvent.getWorkflowInstanceId()); - assertEquals(WorkflowOperationType.PAUSE, workflowOperationEvent.getWorkflowOperationType()); + assertEquals(WorkflowOperationEventType.PAUSE, workflowOperationEvent.getWorkflowOperationType()); } @Test void killEvent() { WorkflowOperationEvent workflowOperationEvent = WorkflowOperationEvent.killEvent(1); assertEquals(1, workflowOperationEvent.getWorkflowInstanceId()); - assertEquals(WorkflowOperationType.KILL, workflowOperationEvent.getWorkflowOperationType()); + assertEquals(WorkflowOperationEventType.KILL, workflowOperationEvent.getWorkflowOperationType()); } }