Add TaskExecutionRunnable in WorkflowExecutionDAG

dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/ProcessInstance.java

@@ -34,6 +34,7 @@ import java.util.Date;
 import java.util.List;
 
 import lombok.AllArgsConstructor;
+import lombok.Builder;
 import lombok.Data;
 import lombok.NoArgsConstructor;
 
@@ -44,11 +45,10 @@ import com.baomidou.mybatisplus.annotation.TableId;
 import com.baomidou.mybatisplus.annotation.TableName;
 import com.google.common.base.Strings;
 
-/**
- * process instance
- */
-@NoArgsConstructor
 @Data
+@Builder
+@AllArgsConstructor
+@NoArgsConstructor
 @TableName("t_ds_process_instance")
 public class ProcessInstance {
 

dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/ProcessTaskRelationLogDao.java

@@ -27,6 +27,9 @@ public interface ProcessTaskRelationLogDao extends IDao<ProcessTaskRelationLog>
 
     List<ProcessTaskRelationLog> queryByWorkflowDefinitionCode(long workflowDefinitionCode);
 
+    List<ProcessTaskRelationLog> queryByWorkflowDefinitionCodeAndVersion(long workflowDefinitionCode,
+                                                                         int workflowDefinitionVerison);
+
     void deleteByWorkflowDefinitionCode(long workflowDefinitionCode);
 
     /**

dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/impl/ProcessTaskRelationLogDaoImpl.java

@@ -44,6 +44,12 @@ public class ProcessTaskRelationLogDaoImpl extends BaseDao<ProcessTaskRelationLo
         return mybatisMapper.queryByProcessCode(workflowDefinitionCode);
     }
 
+    @Override
+    public List<ProcessTaskRelationLog> queryByWorkflowDefinitionCodeAndVersion(long workflowDefinitionCode,
+                                                                                int workflowDefinitionVerison) {
+        return mybatisMapper.queryByProcessCodeAndVersion(workflowDefinitionCode, workflowDefinitionVerison);
+    }
+
     @Override
     public void deleteByWorkflowDefinitionCode(long workflowDefinitionCode) {
         mybatisMapper.deleteByWorkflowDefinitionCode(workflowDefinitionCode);

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/exception/TaskExecuteRunnableNotFoundException.java

@@ -0,0 +1,29 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.dolphinscheduler.server.master.exception;
+
+public class TaskExecuteRunnableNotFoundException extends RuntimeException {
+
+    public TaskExecuteRunnableNotFoundException(Integer workflowInstanceId) {
+        super("WorkflowExecuteRunnable not found: [id=" + workflowInstanceId + "]");
+    }
+
+    public TaskExecuteRunnableNotFoundException(String workflowInstanceName) {
+        super("WorkflowExecuteRunnable not found: [name=" + workflowInstanceName + "]");
+    }
+}

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/exception/WorkflowExecuteRunnableNotFoundException.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.server.master.exception;
+
+public class WorkflowExecuteRunnableNotFoundException extends RuntimeException {
+
+    public WorkflowExecuteRunnableNotFoundException(Integer workflowInstanceId) {
+        super("WorkflowExecuteRunnable not found: [id=" + workflowInstanceId + "]");
+    }
+
+    public WorkflowExecuteRunnableNotFoundException(String workflowInstanceName) {
+        super("WorkflowExecuteRunnable not found: [name=" + workflowInstanceName + "]");
+    }
+
+}

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

@@ -28,8 +28,7 @@ public class DefaultTaskExecuteRunnable extends PriorityDelayTaskExecuteRunnable
 
     private final TaskExecuteRunnableOperatorManager taskExecuteRunnableOperatorManager;
 
-    public DefaultTaskExecuteRunnable(ProcessInstance workflowInstance,
-                                      TaskInstance taskInstance,
+    public DefaultTaskExecuteRunnable(ProcessInstance workflowInstance, TaskInstance taskInstance,
                                       TaskExecutionContext taskExecutionContext,
                                       TaskExecuteRunnableOperatorManager taskExecuteRunnableOperatorManager) {
         super(workflowInstance, taskInstance, taskExecutionContext);

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

@@ -60,7 +60,6 @@ import org.apache.dolphinscheduler.plugin.task.api.utils.MapUtils;
 import org.apache.dolphinscheduler.plugin.task.spark.SparkParameters;
 import org.apache.dolphinscheduler.server.master.builder.TaskExecutionContextBuilder;
 import org.apache.dolphinscheduler.server.master.config.MasterConfig;
-import org.apache.dolphinscheduler.server.master.exception.TaskExecutionContextCreateException;
 import org.apache.dolphinscheduler.service.expand.CuringParamsService;
 import org.apache.dolphinscheduler.service.process.ProcessService;
 import org.apache.dolphinscheduler.spi.datasource.BaseConnectionParam;
@@ -102,7 +101,7 @@ public class TaskExecutionContextFactory {
     @Autowired
     private HikariDataSource hikariDataSource;
 
-    public TaskExecutionContext createTaskExecutionContext(TaskInstance taskInstance) throws TaskExecutionContextCreateException {
+    public TaskExecutionContext createTaskExecutionContext(TaskInstance taskInstance) {
         ProcessInstance workflowInstance = taskInstance.getProcessInstance();
 
         ResourceParametersHelper resources =

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

@@ -19,8 +19,6 @@ package org.apache.dolphinscheduler.server.master.runner.execute;
 
 import org.apache.dolphinscheduler.dao.entity.TaskInstance;
 import org.apache.dolphinscheduler.server.master.cache.ProcessInstanceExecCacheManager;
-import org.apache.dolphinscheduler.server.master.exception.TaskExecuteRunnableCreateException;
-import org.apache.dolphinscheduler.server.master.exception.TaskExecutionContextCreateException;
 import org.apache.dolphinscheduler.server.master.runner.DefaultTaskExecuteRunnable;
 import org.apache.dolphinscheduler.server.master.runner.TaskExecuteRunnableFactory;
 import org.apache.dolphinscheduler.server.master.runner.TaskExecutionContextFactory;
@@ -46,17 +44,13 @@ public class DefaultTaskExecuteRunnableFactory implements TaskExecuteRunnableFac
     private TaskExecuteRunnableOperatorManager taskExecuteRunnableOperatorManager;
 
     @Override
-    public DefaultTaskExecuteRunnable createTaskExecuteRunnable(TaskInstance taskInstance) throws TaskExecuteRunnableCreateException {
+    public DefaultTaskExecuteRunnable createTaskExecuteRunnable(TaskInstance taskInstance) {
         WorkflowExecuteRunnable workflowExecuteRunnable =
                 processInstanceExecCacheManager.getByProcessInstanceId(taskInstance.getProcessInstanceId());
-        try {
-            return new DefaultTaskExecuteRunnable(
-                    workflowExecuteRunnable.getWorkflowExecuteContext().getWorkflowInstance(),
-                    taskInstance,
-                    taskExecutionContextFactory.createTaskExecutionContext(taskInstance),
-                    taskExecuteRunnableOperatorManager);
-        } catch (TaskExecutionContextCreateException ex) {
-            throw new TaskExecuteRunnableCreateException("Create DefaultTaskExecuteRunnable failed", ex);
-        }
+        return new DefaultTaskExecuteRunnable(
+                workflowExecuteRunnable.getWorkflowExecuteContext().getWorkflowInstance(),
+                taskInstance,
+                taskExecutionContextFactory.createTaskExecutionContext(taskInstance),
+                taskExecuteRunnableOperatorManager);
     }
 }

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/utils/ExceptionUtils.java

@@ -0,0 +1,31 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.dolphinscheduler.server.master.utils;
+
+import lombok.experimental.UtilityClass;
+
+import org.springframework.dao.DataAccessResourceFailureException;
+
+@UtilityClass
+public class ExceptionUtils {
+
+    public boolean isDatabaseConnectedFailedException(Throwable e) {
+        return e instanceof DataAccessResourceFailureException;
+    }
+
+}

dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/utils/ExceptionUtilsTest.java

@@ -0,0 +1,18 @@
+package org.apache.dolphinscheduler.server.master.utils;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.dao.DataAccessResourceFailureException;
+
+class ExceptionUtilsTest {
+
+    @Test
+    void isDatabaseConnectedFailedException() {
+        // todo: Directly connect to database
+        assertTrue(ExceptionUtils.isDatabaseConnectedFailedException(
+                new DataAccessResourceFailureException("Database connection failed")));
+        assertFalse(ExceptionUtils.isDatabaseConnectedFailedException(new RuntimeException("runtime exception")));
+    }
+}

dolphinscheduler-workflow-engine/README.md

@@ -0,0 +1,2 @@
+# Introduction
+ 

dolphinscheduler-workflow-engine/pom.xml

@@ -0,0 +1,36 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>org.apache.dolphinscheduler</groupId>
+        <artifactId>dolphinscheduler</artifactId>
+        <version>dev-SNAPSHOT</version>
+    </parent>
+
+    <artifactId>dolphinscheduler-workflow-engine</artifactId>
+
+    <dependencyManagement>
+        <dependencies>
+            <dependency>
+                <groupId>org.apache.dolphinscheduler</groupId>
+                <artifactId>dolphinscheduler-bom</artifactId>
+                <version>${project.version}</version>
+                <type>pom</type>
+                <scope>import</scope>
+            </dependency>
+        </dependencies>
+    </dependencyManagement>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.dolphinscheduler</groupId>
+            <artifactId>dolphinscheduler-common</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework</groupId>
+            <artifactId>spring-tx</artifactId>
+        </dependency>
+    </dependencies>
+</project>

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/dag/BaseDAG.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 BaseDAG {
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/dag/DAG.java

@@ -0,0 +1,82 @@
+/*
+ * 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;
+
+import java.util.List;
+
+/**
+ * The Directed Acyclic Graph class.
+ * <p>
+ * The DAG is a directed graph, which contains the nodes and the edges, the nodeName is the unique identifier of the node.
+ * 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<Node, NodeIdentify> {
+
+    /**
+     * Get the direct post node of given dagNode, if the dagNode is null, return the nodes which doesn't have inDegrees.
+     * e.g. The DAG is:
+     * <pre>
+     *      {@code
+     *          1 -> 2 -> 3
+     *          4 -> 5
+     *          6
+     *      }
+     * </pre>
+     * <li> The post node of 1 is 2.
+     * <li> The post node of 3 is empty.
+     * <li> The post node of null is 1,4,6.
+     *
+     * @param node the node of the DAG, can be null.
+     * @return post node list, sort by priority.
+     */
+    List<Node> getDirectPostNodes(Node node);
+
+    List<Node> getDirectPostNodesByIdentify(NodeIdentify nodeIdentify);
+
+    /**
+     * Get the direct pre node of given dagNode, if the dagNode is null, return the nodes which doesn't have outDegrees.
+     * e.g. The DAG is:
+     * <pre>
+     *      {@code
+     *          1 -> 2 -> 3
+     *          4 -> 5
+     *          6
+     *      }
+     * </pre>
+     * <li> The pre node of 1 is empty.
+     * <li> The pre node of 3 is 2.
+     * <li> The pre node of null is 3,5,6.
+     *
+     * @param node the node of the DAG, can be null.
+     * @return pre node list, sort by priority.
+     */
+    List<Node> getDirectPreNodes(Node node);
+
+    List<Node> getDirectPreNodesByIdentify(NodeIdentify nodeIdentify);
+
+    /**
+     * Get the node of the DAG by the node name.
+     *
+     * @param nodeIdentify the node name.
+     * @return the node of the DAG, return null if cannot find the node.
+     */
+    Node getDAGNode(NodeIdentify nodeIdentify);
+
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/dag/DAGEdge.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.dag;
+
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/**
+ * The edge of the DAG.
+ * <p>
+ * The edge contains the fromNodeName and the toNodeName, the fromNodeName is the node name of the from node, the toNodeName is the node name of the to node.
+ * <p>
+ * The formNodeName can be null, which means the edge is from the start node of the DAG.
+ * The toNodeName can be null, which means the edge is to the end node of the DAG.
+ * The fromNodeName and the toNodeName cannot be null at the same time.
+ */
+@Data
+@Builder
+@NoArgsConstructor
+public class DAGEdge {
+
+    private String fromNodeName;
+    private String toNodeName;
+
+    public DAGEdge(String fromNodeName, String toNodeName) {
+        if (fromNodeName == null && toNodeName == null) {
+            throw new IllegalArgumentException("fromNodeName and toNodeName cannot be null at the same time"
+                    + "fromNodeName: " + fromNodeName + ", toNodeName: " + toNodeName);
+        }
+        if (fromNodeName != null && fromNodeName.equals(toNodeName)) {
+            throw new IllegalArgumentException("fromNodeName and toNodeName cannot be the same"
+                    + "fromNodeName: " + fromNodeName + ", toNodeName: " + toNodeName);
+        }
+        this.fromNodeName = fromNodeName;
+        this.toNodeName = toNodeName;
+    }
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/dag/DAGNodeDefinition.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.dag;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class DAGNodeDefinition {
+
+    private String nodeName;
+
+    private boolean skip;
+
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/dag/Edge.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.dag;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+
+@Getter
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class Edge {
+
+    private NodeIdentify from;
+
+    private NodeIdentify to;
+
+}

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();
+
+}

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();
+
+}

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 {
+
+}

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();
+
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/dag/Node.java

@@ -0,0 +1,42 @@
+/*
+ * 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;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+
+/**
+ * The node of the DAG.
+ * <p>
+ * The node contains the node name, the content of the node, the inDegrees and the outDegrees.
+ * The inDegrees is the edge from other nodes to the current node, the outDegrees is the edge from the current
+ * node to other nodes.
+ */
+@Getter
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class Node {
+
+    private NodeIdentify nodeIdentify;
+
+    private NodeContext nodeContext;
+
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/dag/NodeContext.java

@@ -0,0 +1,36 @@
+/*
+ * 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;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@Builder
+@AllArgsConstructor
+@NoArgsConstructor
+public class NodeContext {
+
+    /**
+     * whether the node is skipped, default is false, which means the node is not skipped.
+     * If the node is skipped, the node will not be executed.
+     */
+    private boolean skip = false;
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/dag/NodeIdentify.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.dag;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@Builder
+@AllArgsConstructor
+@NoArgsConstructor
+public class NodeIdentify {
+
+    private Long id;
+
+    private String name;
+
+}

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;
+    }
+}

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;
+    }
+}

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 {
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/dag/TaskIdentify.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;
+
+import lombok.EqualsAndHashCode;
+
+@EqualsAndHashCode
+public class TaskIdentify implements ITaskIdentify {
+
+    private final Long id;
+    private final String name;
+
+    public TaskIdentify(Long id, String name) {
+        this.id = id;
+        this.name = name;
+    }
+
+    @Override
+    public Long getId() {
+        return id;
+    }
+
+    @Override
+    public String getName() {
+        return name;
+    }
+
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/dag/WorkflowDAG.java

@@ -0,0 +1,124 @@
+/*
+ * 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;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * The IWorkflowDAG represent the DAG of a workflow.
+ */
+public class WorkflowDAG implements DAG<ITask, ITaskIdentify> {
+
+    private final Map<ITaskIdentify, ITask> dagNodeMap;
+
+    private final Map<ITaskIdentify, Set<ITaskIdentify>> outdegreeMap;
+
+    private final Map<ITaskIdentify, Set<ITaskIdentify>> inDegredMap;
+
+    public WorkflowDAG(List<ITask> tasks,
+                       List<ITaskChain> taskChains) {
+        this.dagNodeMap = new HashMap<>();
+        this.outdegreeMap = new HashMap<>();
+        this.inDegredMap = new HashMap<>();
+
+        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<ITaskIdentify> outDegrees = outdegreeMap.computeIfAbsent(fromIdentify, k -> new HashSet<>());
+            if (outDegrees.contains(toIdentify)) {
+                throw new IllegalArgumentException("Duplicate task chain: " + fromIdentify + " -> " + toIdentify);
+            }
+            outDegrees.add(toIdentify);
+            Set<ITaskIdentify> inDegrees = inDegredMap.computeIfAbsent(toIdentify, k -> new HashSet<>());
+            if (inDegrees.contains(fromIdentify)) {
+                throw new IllegalArgumentException("Duplicate task chain: " + fromIdentify + " -> " + toIdentify);
+            }
+            inDegrees.add(fromIdentify);
+        }
+    }
+
+    @Override
+    public List<ITask> getDirectPostNodes(ITask iTask) {
+        if (iTask == null) {
+            return getDirectPostNodesByIdentify(null);
+        }
+        return getDirectPostNodesByIdentify(iTask.getIdentify());
+    }
+
+    @Override
+    public List<ITask> getDirectPostNodesByIdentify(ITaskIdentify taskIdentify) {
+        if (taskIdentify == null) {
+            return dagNodeMap.values()
+                    .stream()
+                    .filter(task -> !inDegredMap.containsKey(task.getIdentify()))
+                    .collect(Collectors.toList());
+        }
+        return outdegreeMap.getOrDefault(taskIdentify, Collections.emptySet())
+                .stream()
+                .map(dagNodeMap::get)
+                .collect(Collectors.toList());
+    }
+
+    @Override
+    public List<ITask> getDirectPreNodes(ITask iTask) {
+        if (iTask == null) {
+            return getDirectPreNodesByIdentify(null);
+        }
+        return getDirectPreNodesByIdentify(iTask.getIdentify());
+    }
+
+    @Override
+    public List<ITask> getDirectPreNodesByIdentify(ITaskIdentify taskIdentify) {
+        if (taskIdentify == null) {
+            return dagNodeMap.values()
+                    .stream()
+                    .filter(task -> !outdegreeMap.containsKey(taskIdentify))
+                    .collect(Collectors.toList());
+        }
+        return inDegredMap.getOrDefault(taskIdentify, Collections.emptySet())
+                .stream()
+                .map(dagNodeMap::get)
+                .collect(Collectors.toList());
+    }
+
+    @Override
+    public ITask getDAGNode(ITaskIdentify taskIdentify) {
+        return dagNodeMap.get(taskIdentify);
+    }
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/dag/WorkflowDAGBuilder.java

@@ -0,0 +1,107 @@
+/*
+ * 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;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Used to build WorkflowDAG, you need to add TaskNode first, then add TaskEdge.
+ * After adding all the TaskNodes and TaskEdges, you can call the build method to get the WorkflowDAG.
+ * <p>
+ * Example:
+ * <pre>
+ *     {@code
+ *          WorkflowDAG workflowDAG = WorkflowDAGBuilder.newBuilder()
+ *                 .addTaskNode(taskNodeA)
+ *                 .addTaskNode(taskNodeB)
+ *                 .addTaskNode(taskNodeC)
+ *                 .addTaskEdge(edgeAB)
+ *                 .addTaskEdge(edgeBC)
+ *                 .build();
+ *     }
+ * </pre>
+ */
+public class WorkflowDAGBuilder {
+
+    private final Map<String, Node> taskNameMap;
+
+    private WorkflowDAGBuilder() {
+        this.taskNameMap = new HashMap<>();
+    }
+
+    public static WorkflowDAGBuilder newBuilder() {
+        return new WorkflowDAGBuilder();
+    }
+
+    public WorkflowDAGBuilder addTaskNodes(List<DAGNodeDefinition> dagNodes) {
+        dagNodes.forEach(this::addTaskNode);
+        return this;
+    }
+
+    public WorkflowDAGBuilder addTaskNode(DAGNodeDefinition dagNodeDefinition) {
+        String nodeName = dagNodeDefinition.getNodeName();
+        if (taskNameMap.containsKey(nodeName)) {
+            throw new IllegalArgumentException("TaskNode with name " + nodeName + " already exists");
+        }
+
+        Node taskNode = Node.builder()
+                .nodeName(nodeName)
+                .inDegrees(new ArrayList<>())
+                .outDegrees(new ArrayList<>())
+                .skip(dagNodeDefinition.isSkip())
+                .build();
+        taskNameMap.put(nodeName, taskNode);
+        return this;
+    }
+
+    public WorkflowDAGBuilder addTaskEdges(List<DAGEdge> processTaskRelations) {
+        processTaskRelations.forEach(this::addTaskEdge);
+        return this;
+    }
+
+    public WorkflowDAGBuilder addTaskEdge(DAGEdge dagEdge) {
+        String fromNodeName = dagEdge.getFromNodeName();
+        String toNodeName = dagEdge.getToNodeName();
+
+        if (taskNameMap.containsKey(fromNodeName)) {
+            Node fromTask = taskNameMap.get(fromNodeName);
+            if (fromTask.getOutDegrees().contains(dagEdge)) {
+                throw new IllegalArgumentException(
+                        "Edge from " + fromNodeName + " to " + toNodeName + " already exists");
+            }
+            fromTask.getOutDegrees().add(dagEdge);
+        }
+        if (taskNameMap.containsKey(toNodeName)) {
+            Node toTask = taskNameMap.get(toNodeName);
+            if (toTask.getInDegrees().contains(dagEdge)) {
+                throw new IllegalArgumentException(
+                        "Edge from " + fromNodeName + " to " + toNodeName + " already exists");
+            }
+            toTask.getInDegrees().add(dagEdge);
+        }
+        return this;
+    }
+
+    public WorkflowDAG build() {
+        return new WorkflowDAG(new ArrayList<>(taskNameMap.values()));
+    }
+
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/engine/DAGEngine.java

@@ -0,0 +1,183 @@
+/*
+ * 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.engine;
+
+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.WorkflowFinishEvent;
+import org.apache.dolphinscheduler.workflow.engine.workflow.ITaskExecutionPlan;
+import org.apache.dolphinscheduler.workflow.engine.workflow.ITaskExecutionRunnable;
+import org.apache.dolphinscheduler.workflow.engine.workflow.ITaskExecutionRunnableIdentify;
+import org.apache.dolphinscheduler.workflow.engine.workflow.IWorkflowExecutionContext;
+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 IEventRepository eventRepository;
+    private final WorkflowExecutionDAG workflowExecutionDAG;
+    private final WorkflowDAG workflowDAG;
+
+    private final IWorkflowExecutionRunnableIdentify workflowExecutionRunnableIdentify;
+
+    public DAGEngine(IWorkflowExecutionContext workflowExecutionContext) {
+        this.workflowExecutionContext = workflowExecutionContext;
+
+        this.workflowExecutionRunnableIdentify = workflowExecutionContext.getIdentify();
+        this.eventRepository = workflowExecutionContext.getEventRepository();
+        this.workflowDAG = workflowExecutionContext.getWorkflowDAG();
+        this.workflowExecutionDAG = workflowExecutionContext.getWorkflowExecutionDAG();
+    }
+
+    @Override
+    public void start() {
+        List<ITaskIdentify> 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 triggerNextTasks(ITaskIdentify taskIdentify) {
+        List<ITaskIdentify> directPostNodeIdentifies = workflowDAG.getDirectPostNodesByIdentify(taskIdentify)
+                .stream()
+                .map(ITask::getIdentify)
+                .collect(Collectors.toList());
+        if (CollectionUtils.isNotEmpty(directPostNodeIdentifies)) {
+            directPostNodeIdentifies.forEach(this::triggerTask);
+            return;
+        }
+        List<ITaskExecutionRunnableIdentify> 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);
+        }
+        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<ITaskExecutionRunnableIdentify> 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<ITaskExecutionRunnableIdentify> 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<ITaskExecutionRunnableIdentify> getActiveTaskExecutionIdentify() {
+        return workflowExecutionDAG.getActiveTaskExecutionPlan()
+                .stream()
+                .map(ITaskExecutionPlan::getActiveTaskExecutionRunnable)
+                .map(ITaskExecutionRunnable::getIdentify)
+                .collect(Collectors.toList());
+    }
+
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/engine/DAGEngineFactory.java

@@ -0,0 +1,38 @@
+/*
+ * 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.engine;
+
+import org.apache.dolphinscheduler.workflow.engine.workflow.ITaskExecutionRunnableFactory;
+import org.apache.dolphinscheduler.workflow.engine.workflow.IWorkflowExecutionContext;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class DAGEngineFactory implements IDAGEngineFactory {
+
+    private final ITaskExecutionRunnableFactory taskExecutionRunnableFactory;
+
+    public DAGEngineFactory(ITaskExecutionRunnableFactory taskExecutionRunnableFactory) {
+        this.taskExecutionRunnableFactory = taskExecutionRunnableFactory;
+    }
+
+    @Override
+    public IDAGEngine createDAGEngine(IWorkflowExecutionContext workflowExecutionContext) {
+        return new DAGEngine(workflowExecutionContext);
+    }
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/engine/EventEngine.java

@@ -0,0 +1,122 @@
+/*
+ * 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.engine;
+
+import org.apache.dolphinscheduler.common.constants.Constants;
+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;
+import org.apache.commons.lang3.time.StopWatch;
+
+import java.util.Collection;
+
+import lombok.extern.slf4j.Slf4j;
+
+import org.slf4j.MDC;
+
+@Slf4j
+public class EventEngine extends BaseDaemonThread implements IEventEngine {
+
+    private final IWorkflowExecutionRunnableRepository workflowExecuteRunnableRepository;
+
+    private final IEventFirer eventFirer;
+
+    private volatile boolean stop = false;
+
+    public EventEngine(IWorkflowExecutionRunnableRepository workflowExecuteRunnableRepository,
+                       IEventFirer eventFirer) {
+        super("EventEngine");
+        this.workflowExecuteRunnableRepository = workflowExecuteRunnableRepository;
+        this.eventFirer = eventFirer;
+    }
+
+    @Override
+    public synchronized void start() {
+        this.stop = false;
+        super.start();
+        log.info(getClass().getName() + " started");
+    }
+
+    @Override
+    public void run() {
+        while (!stop) {
+            try {
+                Collection<IWorkflowExecutionRunnable> workflowExecutionRunnableCollection =
+                        workflowExecuteRunnableRepository.getActiveWorkflowExecutionRunnable();
+                if (CollectionUtils.isEmpty(workflowExecutionRunnableCollection)) {
+                    log.debug("There is no active WorkflowExecutionRunnable");
+                    this.wait(3_000);
+                    continue;
+                }
+                fireAllActiveEvents(workflowExecutionRunnableCollection);
+            } catch (Throwable throwable) {
+                log.error("Fire active event error", throwable);
+                ThreadUtils.sleep(3_000);
+            }
+        }
+    }
+
+    public void fireAllActiveEvents(Collection<IWorkflowExecutionRunnable> workflowExecutionRunnableList) {
+        StopWatch stopWatch = StopWatch.createStarted();
+
+        log.info("Fire all active events cost: {} ms", stopWatch.getTime());
+        stopWatch.stop();
+        for (IWorkflowExecutionRunnable workflowExecutionRunnable : workflowExecutionRunnableList) {
+            IWorkflowExecutionContext workflowExecutionContext =
+                    workflowExecutionRunnable.getWorkflowExecutionContext();
+            IWorkflowExecutionRunnableIdentify identify = workflowExecutionContext.getIdentify();
+            try {
+                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);
+                            if (ex != null) {
+                                log.error("Fire event for WorkflowExecutionRunnable: {} error", identify, ex);
+                            } else {
+                                if (fireCount > 0) {
+                                    log.info("Fire {} events for WorkflowExecutionRunnable: {} success", fireCount,
+                                            identify);
+                                }
+                            }
+                        });
+            } finally {
+                MDC.remove(Constants.WORKFLOW_INSTANCE_ID_MDC_KEY);
+            }
+        }
+    }
+
+    @Override
+    public synchronized void shutdown() {
+        if (stop) {
+            log.warn("EventEngine has already stopped");
+            return;
+        }
+        this.stop = true;
+        eventFirer.shutdown();
+    }
+
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/engine/EventEngineFactory.java

@@ -0,0 +1,59 @@
+/*
+ * 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.engine;
+
+import org.apache.dolphinscheduler.workflow.engine.workflow.IWorkflowExecutionRunnableRepository;
+import org.apache.dolphinscheduler.workflow.engine.workflow.SingletonWorkflowExecutionRunnableRepository;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class EventEngineFactory implements IEventEngineFactory {
+
+    private static final IWorkflowExecutionRunnableRepository DEFAULT_WORKFLOW_EXECUTION_RUNNABLE_FACTORY =
+            SingletonWorkflowExecutionRunnableRepository.getInstance();
+
+    private IWorkflowExecutionRunnableRepository workflowExecuteRunnableRepository;
+
+    private static final int DEFAULT_EVENT_FIRE_THREAD_POOL_SIZE = Runtime.getRuntime().availableProcessors() * 2;
+
+    private int eventFireThreadPoolSize = DEFAULT_EVENT_FIRE_THREAD_POOL_SIZE;
+
+    private EventEngineFactory() {
+    }
+
+    public static EventEngineFactory newEventEngineFactory() {
+        return new EventEngineFactory();
+    }
+
+    public EventEngineFactory withWorkflowExecuteRunnableRepository(IWorkflowExecutionRunnableRepository workflowExecuteRunnableRepository) {
+        this.workflowExecuteRunnableRepository = workflowExecuteRunnableRepository;
+        return this;
+    }
+
+    public int withEventFireThreadPoolSize(int eventFireThreadPoolSize) {
+        this.eventFireThreadPoolSize = eventFireThreadPoolSize;
+        return this.eventFireThreadPoolSize;
+    }
+
+    @Override
+    public IEventEngine createEventEngine() {
+        EventFirer eventFirer = new EventFirer(eventFireThreadPoolSize);
+        return new EventEngine(workflowExecuteRunnableRepository, eventFirer);
+    }
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/engine/EventFirer.java

@@ -0,0 +1,99 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.dolphinscheduler.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.IEventRepository;
+import org.apache.dolphinscheduler.workflow.engine.utils.ExceptionUtils;
+import org.apache.dolphinscheduler.workflow.engine.workflow.IEventfulExecutionRunnable;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ThreadPoolExecutor;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class EventFirer implements IEventFirer {
+
+    private final ThreadPoolExecutor eventFireThreadPool;
+
+    public EventFirer(int eventFireThreadPoolSize) {
+        this.eventFireThreadPool =
+                ThreadUtils.newDaemonFixedThreadExecutor("EventFireThreadPool", eventFireThreadPoolSize);
+    }
+
+    @Override
+    public CompletableFuture<Integer> fireActiveEvents(IEventfulExecutionRunnable eventfulExecutionRunnable) {
+        // todo: add MDC key
+        IEventRepository eventRepository = eventfulExecutionRunnable.getEventRepository();
+        if (eventRepository.getEventSize() == 0) {
+            return CompletableFuture.completedFuture(0);
+        }
+        return CompletableFuture.supplyAsync(() -> {
+            int fireCount = 0;
+            for (;;) {
+                IEvent event = eventRepository.poolEvent();
+                if (event == null) {
+                    break;
+                }
+
+                if (event instanceof IAsyncEvent) {
+                    fireAsyncEvent(event);
+                    fireCount++;
+                    continue;
+                }
+                try {
+                    fireSyncEvent(event);
+                    fireCount++;
+                } catch (Exception ex) {
+                    if (ExceptionUtils.isDatabaseConnectedFailedException(ex)) {
+                        // If the event is failed due to cannot connect to DB, we should retry it
+                        eventRepository.storeEventToHead(event);
+                    }
+                    throw ex;
+                }
+            }
+            return fireCount;
+        }, eventFireThreadPool);
+    }
+
+    @Override
+    public void shutdown() {
+        eventFireThreadPool.shutdown();
+    }
+
+    private void fireAsyncEvent(IEvent event) {
+        CompletableFuture.runAsync(() -> {
+            log.info("Begin fire IAsyncEvent: {}", event);
+            event.getEventOperation().operate();
+            log.info("Success fire IAsyncEvent: {}", event);
+        }, eventFireThreadPool).exceptionally(ex -> {
+            log.error("Failed to fire IAsyncEvent: {}", event, ex);
+            return null;
+        });
+    }
+
+    private void fireSyncEvent(IEvent event) {
+        log.info("Begin fire SyncEvent: {}", event);
+        event.getEventOperation().operate();
+        log.info("Success fire SyncEvent: {}", event);
+    }
+
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/engine/IDAGEngine.java

@@ -0,0 +1,78 @@
+/*
+ * 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.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;
+
+/**
+ * The IDAGEngine is responsible for triggering, killing, pausing, and finalizing task in {@link org.apache.dolphinscheduler.workflow.engine.dag.WorkflowDAG}.
+ * <p>All DAG operation should directly use the method in IDAGEngine, new {@link IWorkflowExecutionDAG} should be triggered by new IDAGEngine.
+ */
+public interface IDAGEngine {
+
+    /**
+     * Start the DAGEngine, will trigger the start tasks.
+     */
+    void start();
+
+    /**
+     * Trigger the tasks which are post of the given task.
+     * <P> If there are no task after the given taskNode, will try to finish the WorkflowExecutionRunnable(Send a task chain end event).
+     *
+     * @param taskIdentify the parent task identify
+     */
+    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(ITaskExecutionRunnableIdentify taskInstanceId);
+
+    /**
+     * Retry the given task.
+     *
+     * @param taskInstanceId taskInstanceId
+     */
+    void retryTask(ITaskExecutionRunnableIdentify taskInstanceId);
+
+    void pause();
+
+    /**
+     * Pause the given task.
+     */
+    void pauseTask(ITaskExecutionRunnableIdentify taskExecutionIdentify);
+
+    void kill();
+
+    /**
+     * Kill the given task.
+     */
+    void killTask(ITaskExecutionRunnableIdentify taskExecutionIdentify);
+
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/engine/IDAGEngineFactory.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.engine;
+
+import org.apache.dolphinscheduler.workflow.engine.workflow.IWorkflowExecutionContext;
+
+public interface IDAGEngineFactory {
+
+    IDAGEngine createDAGEngine(IWorkflowExecutionContext workflowExecutionContext);
+
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/engine/IEventAcceptor.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.engine;
+
+import org.apache.dolphinscheduler.workflow.engine.event.IEvent;
+
+public interface IEventAcceptor {
+
+    void accept(IEvent event);
+
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/engine/IEventEngine.java

@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.dolphinscheduler.workflow.engine.engine;
+
+public interface IEventEngine {
+
+    /**
+     * Start the event engine.
+     */
+    void start();
+
+    /**
+     * Shutdown the event engine. The event engine cannot be restarted after shutdown. This method will block until the event engine is completely shutdown.
+     */
+    void shutdown();
+
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/engine/IEventEngineFactory.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.engine;
+
+public interface IEventEngineFactory {
+
+    IEventEngine createEventEngine();
+
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/engine/IEventFirer.java

@@ -0,0 +1,42 @@
+/*
+ * 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.engine;
+
+import org.apache.dolphinscheduler.workflow.engine.workflow.IEventfulExecutionRunnable;
+
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * The event firer interface used to fire event for {@link IEventfulExecutionRunnable}.
+ *
+ */
+public interface IEventFirer {
+
+    /**
+     * Fire all active events in the event repository
+     *
+     * @return the count of fired success events
+     */
+    CompletableFuture<Integer> fireActiveEvents(IEventfulExecutionRunnable eventfulExecutionRunnable);
+
+    /**
+     * Shutdown the event firer, this method will block until all firing task has been completed.
+     */
+    void shutdown();
+
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/engine/IWorkflowEngine.java

@@ -0,0 +1,70 @@
+/*
+ * 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.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}.
+ * All operation on a workflow instance should be done through the WorkflowEngine.
+ */
+public interface IWorkflowEngine {
+
+    /**
+     * Start the workflow engine.
+     */
+    void start();
+
+    /**
+     * Trigger a workflow to start.
+     *
+     * @param workflowExecuteRunnable the workflow to start
+     */
+    void triggerWorkflow(IWorkflowExecutionRunnable workflowExecuteRunnable);
+
+    /**
+     * Pause a workflow instance.
+     *
+     * @param workflowExecutionRunnableIdentify the ID of the workflow to pause
+     * @throws WorkflowExecuteRunnableNotFoundException if the workflow is not found
+     */
+    void pauseWorkflow(IWorkflowExecutionRunnableIdentify workflowExecutionRunnableIdentify);
+
+    /**
+     * Kill a workflow instance.
+     *
+     * @param workflowExecutionRunnableIdentify the ID of the workflow to stop
+     * @throws WorkflowExecuteRunnableNotFoundException if the workflow is not found
+     */
+    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 workflowExecutionRunnableIdentify the ID of the workflow to finalize
+     */
+    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.
+     */
+    void shutdown();
+
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/engine/IWorkflowEngineFactory.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.engine;
+
+public interface IWorkflowEngineFactory {
+
+    IWorkflowEngine createWorkflowEngine();
+
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/engine/WorkflowEngine.java

@@ -0,0 +1,100 @@
+/*
+ * 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.engine;
+
+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;
+
+@Slf4j
+public class WorkflowEngine implements IWorkflowEngine {
+
+    private final IWorkflowExecutionRunnableRepository workflowExecuteRunnableRepository;
+
+    private final IEventEngine eventEngine;
+
+    public WorkflowEngine(IWorkflowExecutionRunnableRepository workflowExecuteRunnableRepository,
+                          IEventEngine eventEngine) {
+        this.workflowExecuteRunnableRepository = workflowExecuteRunnableRepository;
+        this.eventEngine = eventEngine;
+    }
+
+    @Override
+    public void start() {
+        eventEngine.start();
+    }
+
+    @Override
+    public void triggerWorkflow(IWorkflowExecutionRunnable workflowExecuteRunnable) {
+        IWorkflowExecutionContext workflowExecutionContext = workflowExecuteRunnable.getWorkflowExecutionContext();
+        IWorkflowExecutionRunnableIdentify workflowExecutionIdentify = workflowExecutionContext.getIdentify();
+        log.info("Triggering WorkflowExecutionRunnable: {}", workflowExecutionIdentify);
+        workflowExecuteRunnableRepository.storeWorkflowExecutionRunnable(workflowExecuteRunnable);
+        workflowExecuteRunnable
+                .storeEventToTail(WorkflowOperationEvent.triggerEvent(workflowExecuteRunnable));
+    }
+
+    @Override
+    public void pauseWorkflow(IWorkflowExecutionRunnableIdentify workflowExecutionRunnableIdentify) {
+        IWorkflowExecutionRunnable workflowExecuteRunnable =
+                workflowExecuteRunnableRepository.getWorkflowExecutionRunnable(workflowExecutionRunnableIdentify);
+        if (workflowExecuteRunnable == null) {
+            throw new WorkflowExecuteRunnableNotFoundException(workflowExecutionRunnableIdentify);
+        }
+        log.info("Pausing WorkflowExecutionRunnable: {}",
+                workflowExecuteRunnable.getWorkflowExecutionContext().getIdentify());
+        workflowExecuteRunnable
+                .storeEventToTail(WorkflowOperationEvent.pauseEvent(workflowExecuteRunnable));
+    }
+
+    @Override
+    public void killWorkflow(IWorkflowExecutionRunnableIdentify workflowExecutionRunnableIdentify) {
+        IWorkflowExecutionRunnable workflowExecuteRunnable =
+                workflowExecuteRunnableRepository.getWorkflowExecutionRunnable(workflowExecutionRunnableIdentify);
+        if (workflowExecuteRunnable == null) {
+            throw new WorkflowExecuteRunnableNotFoundException(workflowExecutionRunnableIdentify);
+        }
+        log.info("Killing WorkflowExecutionRunnable: {}",
+                workflowExecuteRunnable.getWorkflowExecutionContext().getIdentify());
+        workflowExecuteRunnable
+                .storeEventToTail(WorkflowOperationEvent.killEvent(workflowExecuteRunnable));
+    }
+
+    @Override
+    public void finalizeWorkflow(IWorkflowExecutionRunnableIdentify workflowExecutionRunnableIdentify) {
+        IWorkflowExecutionRunnable workflowExecutionRunnable =
+                workflowExecuteRunnableRepository.getWorkflowExecutionRunnable(workflowExecutionRunnableIdentify);
+        if (workflowExecutionRunnable == null) {
+            return;
+        }
+        // todo: If the workflowExecutionRunnable is not finished, we cannot finalize it.
+        log.info("Finalizing WorkflowExecutionRunnable: {}", workflowExecutionRunnable.getIdentity());
+        workflowExecuteRunnableRepository.removeWorkflowExecutionRunnable(workflowExecutionRunnableIdentify);
+    }
+
+    @Override
+    public void shutdown() {
+        eventEngine.shutdown();
+    }
+
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/engine/WorkflowEngineFactory.java

@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.dolphinscheduler.workflow.engine.engine;
+
+import org.apache.dolphinscheduler.workflow.engine.workflow.IWorkflowExecutionRunnableRepository;
+import org.apache.dolphinscheduler.workflow.engine.workflow.SingletonWorkflowExecutionRunnableRepository;
+
+public class WorkflowEngineFactory implements IWorkflowEngineFactory {
+
+    private static final IWorkflowExecutionRunnableRepository DEFAULT_WORKFLOW_EXECUTION_RUNNABLE_FACTORY =
+            SingletonWorkflowExecutionRunnableRepository.getInstance();
+
+    private IWorkflowExecutionRunnableRepository workflowExecuteRunnableRepository =
+            DEFAULT_WORKFLOW_EXECUTION_RUNNABLE_FACTORY;
+
+    private IEventEngine eventEngine;
+
+    private WorkflowEngineFactory() {
+    }
+
+    public static WorkflowEngineFactory newWorkflowEngineFactory() {
+        return new WorkflowEngineFactory();
+    }
+
+    public WorkflowEngineFactory withWorkflowExecuteRunnableRepository(IWorkflowExecutionRunnableRepository workflowExecuteRunnableRepository) {
+        this.workflowExecuteRunnableRepository = workflowExecuteRunnableRepository;
+        return this;
+    }
+
+    public WorkflowEngineFactory withEventEngine(IEventEngine eventEngine) {
+        this.eventEngine = eventEngine;
+        return this;
+    }
+
+    @Override
+    public IWorkflowEngine createWorkflowEngine() {
+        return new WorkflowEngine(workflowExecuteRunnableRepository, eventEngine);
+    }
+}

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 {
+}

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);
+    }
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/EventOperatorManager.java

@@ -0,0 +1,57 @@
+/*
+ * 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 java.util.HashMap;
+import java.util.Map;
+
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * The event operator manager interface used to get {@link ITaskEventOperator}.
+ */
+@Slf4j
+public class EventOperatorManager implements IEventOperatorManager<IEvent> {
+
+    private static final Map<IEventType, IEventOperator<IEvent>> EVENT_OPERATOR_MAP = new HashMap<>();
+
+    private static final EventOperatorManager INSTANCE = new EventOperatorManager();
+
+    private EventOperatorManager() {
+    }
+
+    public static EventOperatorManager getInstance() {
+        return INSTANCE;
+    }
+
+    public void registerEventOperator(IEventType eventType, IEventOperator<IEvent> eventOperator) {
+        EVENT_OPERATOR_MAP.put(eventType, eventOperator);
+    }
+
+    @Override
+    public IEventOperator<IEvent> getEventOperator(IEvent event) {
+        if (event == null) {
+            throw new IllegalArgumentException("event cannot be null");
+        }
+        if (event.getEventType() == null) {
+            throw new IllegalArgumentException("event operator class cannot be null");
+        }
+        return EVENT_OPERATOR_MAP.get(event.getEventType());
+    }
+
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/IAsyncEvent.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;
+
+/**
+ * Mark the event as AsyncEvent, if the event is marked as AsyncEvent, the event will be handled asynchronously and we don't .
+ */
+public interface IAsyncEvent {
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/IDelayEvent.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.event;
+
+public interface IDelayEvent {
+
+    long getEventCreateTime();
+
+    long getDelayTime();
+
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/IEvent.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.event;
+
+import org.apache.dolphinscheduler.workflow.engine.workflow.IWorkflowExecutionRunnableIdentify;
+
+public interface IEvent {
+
+    IWorkflowExecutionRunnableIdentify getWorkflowExecutionRunnableIdentify();
+
+}

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);
+
+}

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<T> {
+
+    void operate(T t);
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/IEventOperator.java

@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.dolphinscheduler.workflow.engine.event;
+
+/**
+ * The event operator interface used to handle event.
+ */
+public interface IEventOperator<E> {
+
+    /**
+     * Handle the given event
+     *
+     * @param event event
+     */
+    void handleEvent(E event);
+
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/IEventOperatorManager.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.event;
+
+/**
+ * The event operator manager interface used to get event operator.
+ */
+public interface IEventOperatorManager<E> {
+
+    void registerEventOperator(IEventType eventType, IEventOperator<E> eventOperator);
+
+    /**
+     * Get the {@link IEventOperator} for the given event.
+     *
+     * @param event event
+     * @return event operator for the given event
+     */
+    IEventOperator<E> getEventOperator(E event);
+
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/IEventRepository.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.event;
+
+import java.util.List;
+
+/**
+ * The event repository interface used to store event.
+ */
+public interface IEventRepository {
+
+    void storeEventToTail(IEvent event);
+
+    void storeEventToHead(IEvent event);
+
+    IEvent poolEvent();
+
+    int getEventSize();
+
+    List<IEvent> getAllEvent();
+
+}

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 {
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/ISyncEvent.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 ISyncEvent {
+
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/ITaskEvent.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 ITaskEvent extends IEvent {
+
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/ITaskEventOperator.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 ITaskEventOperator<E extends ITaskEvent> extends IEventOperator<E> {
+
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/IWorkflowEvent.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;
+
+import org.apache.dolphinscheduler.workflow.engine.workflow.IWorkflowExecutionRunnableIdentify;
+
+public interface IWorkflowEvent extends IEvent<IWorkflowExecutionRunnableIdentify> {
+
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/IWorkflowEventOperator.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 IWorkflowEventOperator<E extends IWorkflowEvent>
+        extends
+            IEventOperator<E> {
+}

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 {
+
+}

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 {
+
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/MemoryEventRepository.java

@@ -0,0 +1,64 @@
+/*
+ * 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 java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.LinkedBlockingDeque;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class MemoryEventRepository implements IEventRepository {
+
+    private final LinkedBlockingDeque<IEvent> eventQueue;
+
+    private MemoryEventRepository() {
+        this.eventQueue = new LinkedBlockingDeque<>();
+    }
+
+    public static MemoryEventRepository newInstance() {
+        return new MemoryEventRepository();
+    }
+
+    @Override
+    public void storeEventToTail(IEvent event) {
+        log.info("Store event to tail: {}", event);
+        eventQueue.offerLast(event);
+    }
+
+    @Override
+    public void storeEventToHead(IEvent event) {
+        eventQueue.offerFirst(event);
+    }
+
+    @Override
+    public IEvent poolEvent() {
+        return eventQueue.poll();
+    }
+
+    @Override
+    public int getEventSize() {
+        return eventQueue.size();
+    }
+
+    public List<IEvent> getAllEvent() {
+        return new ArrayList<>(eventQueue);
+    }
+
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/TaskOperationEvent.java

@@ -0,0 +1,55 @@
+/*
+ * 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.ITaskExecutionPlan;
+
+import lombok.Builder;
+import lombok.Data;
+
+@Data
+@Builder
+public class TaskOperationEvent implements ITaskEvent, ISyncEvent {
+
+    private IEventOperation eventOperation;
+
+    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);
+    }
+
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/TaskOperationEventOperator.java

@@ -0,0 +1,29 @@
+/*
+ * 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 lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class TaskOperationEventOperator implements ITaskEventOperator<TaskOperationEvent> {
+
+    @Override
+    public void handleEvent(TaskOperationEvent event) {
+
+    }
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/TaskOperationEventType.java

@@ -0,0 +1,29 @@
+/*
+ * 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 enum TaskOperationEventType implements IEventType {
+
+    FAILOVER,
+    START,
+    RETRY,
+    KILL,
+    PAUSE,
+    ;
+
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/WorkflowFailedEvent.java

@@ -0,0 +1,47 @@
+/*
+ * 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 lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class WorkflowFailedEvent implements IWorkflowEvent {
+
+    private String failedReason;
+
+    private IWorkflowExecutionRunnableIdentify workflowExecutionRunnableIdentify;
+
+    @Override
+    public IWorkflowExecutionRunnableIdentify getWorkflowExecutionRunnableIdentify() {
+        return null;
+    }
+
+    @Override
+    public IEventType getEventType() {
+        return null;
+    }
+
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/WorkflowFinalizeEvent.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 lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@Builder
+@AllArgsConstructor
+@NoArgsConstructor
+public class WorkflowFinalizeEvent implements IWorkflowEvent, ISyncEvent {
+
+    private Integer workflowInstanceId;
+
+    @Override
+    public IEventType getEventType() {
+        return null;
+    }
+
+    @Override
+    public IWorkflowExecutionRunnableIdentify getWorkflowExecutionRunnableIdentify() {
+        return null;
+    }
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/WorkflowFinalizeEventOperator.java

@@ -0,0 +1,41 @@
+/*
+ * 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.IWorkflowExecutionRunnableRepository;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class WorkflowFinalizeEventOperator
+        implements
+            IWorkflowEventOperator<WorkflowFinalizeEvent> {
+
+    private final IWorkflowExecutionRunnableRepository workflowExecuteRunnableRepository;
+
+    public WorkflowFinalizeEventOperator(IWorkflowExecutionRunnableRepository workflowExecuteRunnableRepository) {
+        this.workflowExecuteRunnableRepository = workflowExecuteRunnableRepository;
+    }
+
+    @Override
+    public void handleEvent(WorkflowFinalizeEvent event) {
+        Integer workflowInstanceId = event.getWorkflowInstanceId();
+        workflowExecuteRunnableRepository.removeWorkflowExecutionRunnable(workflowInstanceId);
+    }
+
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/WorkflowFinishEvent.java

@@ -0,0 +1,46 @@
+/*
+ * 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 lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class WorkflowFinishEvent implements IWorkflowEvent, ISyncEvent {
+
+    private IWorkflowExecutionRunnableIdentify workflowExecutionRunnableIdentify;
+
+    public static WorkflowFinishEvent of(IWorkflowExecutionRunnableIdentify workflowExecutionRunnableIdentify) {
+        return WorkflowFinishEvent.builder()
+                .workflowExecutionRunnableIdentify(workflowExecutionRunnableIdentify)
+                .build();
+    }
+
+    @Override
+    public IEventType getEventType() {
+        return null;
+    }
+
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/WorkflowOperationEvent.java

@@ -0,0 +1,46 @@
+/*
+ * 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;
+
+public class WorkflowOperationEvent implements IWorkflowEvent, ISyncEvent {
+
+    private final IWorkflowExecutionRunnableIdentify workflowExecutionRunnableIdentify;
+
+    public WorkflowOperationEvent(IWorkflowExecutionRunnableIdentify workflowExecutionRunnableIdentify) {
+        this.workflowExecutionRunnableIdentify = workflowExecutionRunnableIdentify;
+    }
+
+    public static WorkflowOperationEvent triggerEvent(IWorkflowExecutionRunnableIdentify workflowExecutionRunnableIdentify) {
+        return new WorkflowOperationEvent(workflowExecutionRunnableIdentify);
+    }
+
+    public static WorkflowOperationEvent pauseEvent(IWorkflowExecutionRunnableIdentify workflowExecutionRunnableIdentify) {
+        return new WorkflowOperationEvent(workflowExecutionRunnableIdentify);
+    }
+
+    public static WorkflowOperationEvent killEvent(IWorkflowExecutionRunnableIdentify workflowExecutionRunnableIdentify) {
+        return new WorkflowOperationEvent(workflowExecutionRunnableIdentify);
+    }
+
+    @Override
+    public IWorkflowExecutionRunnableIdentify getEventIdentify() {
+        return workflowExecutionRunnableIdentify;
+    }
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/WorkflowOperationEventOperator.java

@@ -0,0 +1,86 @@
+/*
+ * 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.utils.ExceptionUtils;
+import org.apache.dolphinscheduler.workflow.engine.workflow.IWorkflowExecutionContext;
+import org.apache.dolphinscheduler.workflow.engine.workflow.IWorkflowExecutionRunnable;
+import org.apache.dolphinscheduler.workflow.engine.workflow.IWorkflowExecutionRunnableRepository;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class WorkflowOperationEventOperator implements IWorkflowEventOperator<WorkflowOperationEvent> {
+
+    private final IWorkflowExecutionRunnableRepository workflowExecuteRunnableRepository;
+
+    public WorkflowOperationEventOperator(IWorkflowExecutionRunnableRepository workflowExecuteRunnableRepository) {
+        this.workflowExecuteRunnableRepository = workflowExecuteRunnableRepository;
+    }
+
+    @Override
+    public void handleEvent(WorkflowOperationEvent event) {
+        IWorkflowExecutionRunnable workflowExecutionRunnable =
+                workflowExecuteRunnableRepository
+                        .getWorkflowExecutionRunnable(event.getWorkflowExecutionRunnableIdentify());
+        if (workflowExecutionRunnable == null) {
+            log.warn("WorkflowExecutionRunnable not found: {}", event);
+            return;
+        }
+        WorkflowOperationEventType workflowOperationEvent = (WorkflowOperationEventType) event.getEventType();
+        switch (workflowOperationEvent) {
+            case TRIGGER:
+                triggerWorkflow(workflowExecutionRunnable);
+                break;
+            case PAUSE:
+                pauseWorkflow(workflowExecutionRunnable);
+                break;
+            case KILL:
+                killWorkflow(workflowExecutionRunnable);
+                break;
+            default:
+                log.error("Unknown operationType for event: {}", event);
+        }
+    }
+
+    private void triggerWorkflow(IWorkflowExecutionRunnable workflowExecutionRunnable) {
+        try {
+            workflowExecutionRunnable.start();
+        } catch (Throwable exception) {
+            if (ExceptionUtils.isDatabaseConnectedFailedException(exception)) {
+                throw exception;
+            }
+            IWorkflowExecutionContext workflowExecutionContext =
+                    workflowExecutionRunnable.getWorkflowExecutionContext();
+            log.error("Trigger workflow: {} failed", workflowExecutionContext.getIdentify(), exception);
+            WorkflowFailedEvent workflowExecutionRunnableFailedEvent = WorkflowFailedEvent.builder()
+                    .workflowExecutionRunnableIdentify(workflowExecutionRunnable.getIdentity())
+                    .failedReason(exception.getMessage())
+                    .build();
+            workflowExecutionRunnable.storeEventToTail(workflowExecutionRunnableFailedEvent);
+        }
+    }
+
+    private void pauseWorkflow(IWorkflowExecutionRunnable workflowExecutionRunnable) {
+        workflowExecutionRunnable.pause();
+    }
+
+    private void killWorkflow(IWorkflowExecutionRunnable workflowExecutionRunnable) {
+        workflowExecutionRunnable.kill();
+    }
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/event/WorkflowOperationEventType.java

@@ -0,0 +1,36 @@
+/*
+ * 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 enum WorkflowOperationEventType implements IEventType {
+
+    /**
+     * Trigger the workflow instance.
+     */
+    TRIGGER,
+    /**
+     * Pause the workflow instance, it will pause the running task instances.
+     */
+    PAUSE,
+    /**
+     * Kill the workflow instance, it will kill the running task instances.
+     */
+    KILL,
+    ;
+
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/exception/WorkflowExecuteRunnableNotFoundException.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.exception;
+
+import org.apache.dolphinscheduler.workflow.engine.workflow.IWorkflowExecutionRunnableIdentify;
+
+public class WorkflowExecuteRunnableNotFoundException extends RuntimeException {
+
+    public WorkflowExecuteRunnableNotFoundException(IWorkflowExecutionRunnableIdentify workflowExecutionRunnableIdentify) {
+        super("WorkflowExecuteRunnable not found: " + workflowExecutionRunnableIdentify);
+    }
+
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/utils/ExceptionUtils.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.utils;
+
+import org.springframework.dao.DataAccessResourceFailureException;
+
+public class ExceptionUtils {
+
+    public static boolean isDatabaseConnectedFailedException(Throwable e) {
+        return e instanceof DataAccessResourceFailureException;
+    }
+
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/utils/IWorkflowExecutionDAGStatusCheck.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.utils;
+
+public interface IWorkflowExecutionDAGStatusCheck {
+
+    boolean isSuccess();
+
+    boolean isFailed();
+
+    boolean isKilled();
+
+    boolean isPaused();
+
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/BaseWorkflowExecutionRunnable.java

@@ -0,0 +1,60 @@
+/*
+ * 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.event.IEventRepository;
+
+public abstract class BaseWorkflowExecutionRunnable implements IWorkflowExecutionRunnable {
+
+    protected final IWorkflowExecutionContext workflowExecutionContext;
+
+    protected WorkflowExecutionRunnableStatus workflowExecutionRunnableStatus;
+
+    public BaseWorkflowExecutionRunnable(IWorkflowExecutionContext workflowExecutionContext,
+                                         WorkflowExecutionRunnableStatus workflowExecutionRunnableStatus) {
+        this.workflowExecutionContext = workflowExecutionContext;
+        this.workflowExecutionRunnableStatus = workflowExecutionRunnableStatus;
+    }
+
+    @Override
+    public IWorkflowExecutionRunnableIdentify getIdentity() {
+        return workflowExecutionContext.getIdentify();
+    }
+
+    @Override
+    public IWorkflowExecutionContext getWorkflowExecutionContext() {
+        return workflowExecutionContext;
+    }
+
+    @Override
+    public IEventRepository getEventRepository() {
+        return workflowExecutionContext.getEventRepository();
+    }
+
+    protected void statusTransform(WorkflowExecutionRunnableStatus targetStatus, Runnable runnable) {
+        WorkflowExecutionRunnableStatus originStatus = workflowExecutionRunnableStatus;
+        try {
+            workflowExecutionRunnableStatus = targetStatus;
+            runnable.run();
+        } catch (Throwable throwable) {
+            workflowExecutionRunnableStatus = originStatus;
+            throw throwable;
+        }
+    }
+
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/DefaultTaskExecutionRunnableDelegate.java

@@ -0,0 +1,51 @@
+/*
+ * 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 DefaultTaskExecutionRunnableDelegate implements ITaskExecutionRunnableDelegate {
+
+    @Override
+    public void beforeStart() {
+
+    }
+
+    @Override
+    public void afterStart() {
+
+    }
+
+    @Override
+    public void beforePause() {
+
+    }
+
+    @Override
+    public void afterPause() {
+
+    }
+
+    @Override
+    public void beforeKill() {
+
+    }
+
+    @Override
+    public void afterKill() {
+
+    }
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/DefaultTaskExecutionRunnableDelegateFactory.java

@@ -0,0 +1,51 @@
+/*
+ * 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 class DefaultTaskExecutionRunnableDelegateFactory implements ITaskExecutionRunnableDelegateFactory {
+
+    private static final DefaultTaskExecutionRunnableDelegateFactory INSTANCE =
+            new DefaultTaskExecutionRunnableDelegateFactory();
+
+    private DefaultTaskExecutionRunnableDelegateFactory() {
+    }
+
+    public static DefaultTaskExecutionRunnableDelegateFactory getInstance() {
+        return INSTANCE;
+    }
+
+    @Override
+    public ITaskExecutionRunnableDelegate createTaskExecutionRunnable(ITaskIdentify taskIdentify,
+                                                                      IWorkflowExecutionContext workflowExecutionContext) {
+        return new DefaultTaskExecutionRunnableDelegate();
+    }
+
+    @Override
+    public ITaskExecutionRunnableDelegate createFailoverTaskExecutionRunnable(ITaskExecutionRunnable taskExecutionRunnable,
+                                                                              IWorkflowExecutionContext workflowExecutionContext) {
+        return new DefaultTaskExecutionRunnableDelegate();
+    }
+
+    @Override
+    public ITaskExecutionRunnableDelegate createRetryTaskExecutionRunnable(ITaskExecutionRunnable taskExecutionRunnable,
+                                                                           IWorkflowExecutionContext workflowExecutionContext) {
+        return new DefaultTaskExecutionRunnableDelegate();
+    }
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/DefaultWorkflowExecutionRunnableDelegate.java

@@ -0,0 +1,51 @@
+/*
+ * 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 DefaultWorkflowExecutionRunnableDelegate implements IWorkflowExecutionRunnableDelegate {
+
+    @Override
+    public void beforeStart() {
+
+    }
+
+    @Override
+    public void afterStart() {
+
+    }
+
+    @Override
+    public void beforePause() {
+
+    }
+
+    @Override
+    public void afterPause() {
+
+    }
+
+    @Override
+    public void beforeKill() {
+
+    }
+
+    @Override
+    public void afterKill() {
+
+    }
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/DefaultWorkflowExecutionRunnableDelegateFactory.java

@@ -0,0 +1,36 @@
+/*
+ * 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 DefaultWorkflowExecutionRunnableDelegateFactory implements IWorkflowExecutionRunnableDelegateFactory {
+
+    private static final DefaultWorkflowExecutionRunnableDelegateFactory INSTANCE =
+            new DefaultWorkflowExecutionRunnableDelegateFactory();
+
+    private DefaultWorkflowExecutionRunnableDelegateFactory() {
+    }
+
+    public static DefaultWorkflowExecutionRunnableDelegateFactory getInstance() {
+        return INSTANCE;
+    }
+
+    @Override
+    public IWorkflowExecutionRunnableDelegate createWorkflowExecutionRunnableDelegate(IWorkflowExecutionContext workflowExecutionContext) {
+        return new DefaultWorkflowExecutionRunnableDelegate();
+    }
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/IEventfulExecutionRunnable.java

@@ -0,0 +1,38 @@
+/*
+ * 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.event.IEvent;
+import org.apache.dolphinscheduler.workflow.engine.event.IEventRepository;
+
+public interface IEventfulExecutionRunnable {
+
+    IEventRepository getEventRepository();
+
+    default void storeEventToTail(IEvent event) {
+        getEventRepository().storeEventToTail(event);
+    }
+
+    default void storeEventToHead(IEvent event) {
+        getEventRepository().storeEventToHead(event);
+    }
+
+    default void onEvent(IEvent event) {
+        throw new UnsupportedOperationException("onEvent is not implemented");
+    }
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/ITaskExecutionContext.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;
+
+import org.apache.dolphinscheduler.workflow.engine.event.IEventRepository;
+
+public interface ITaskExecutionContext {
+
+    ITaskExecutionRunnableIdentify getIdentify();
+
+    IEventRepository getEventRepository();
+
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/ITaskExecutionContextFactory.java

@@ -0,0 +1,27 @@
+/*
+ * 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 ITaskExecutionContextFactory {
+
+    ITaskExecutionContext createTaskExecutionContext(ITaskIdentify taskIdentify,
+                                                     IWorkflowExecutionContext workflowExecutionContext);
+
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/ITaskExecutionPlan.java

@@ -0,0 +1,52 @@
+/*
+ * 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;
+
+/**
+ * The task execution plan interface which represents the task with its execution plan.
+ * The task execution plan contains the task identify, task and the list of task execution runnables.
+ */
+public interface ITaskExecutionPlan {
+
+    void start();
+
+    void failoverTask();
+
+    void retryTask();
+
+    void pauseTask();
+
+    void killTask();
+
+    ITaskIdentify getTaskIdentify();
+
+    ITaskExecutionRunnable getActiveTaskExecutionRunnable();
+
+    ITask getTask();
+
+    List<ITaskExecutionRunnable> getTaskExecutionRunnableList();
+
+    ITaskExecutionRunnable getTaskExecutionRunnable(ITaskExecutionRunnableIdentify taskExecutionRunnableIdentify);
+
+    ITaskExecutionRunnable storeTaskExecutionRunnable(ITaskExecutionRunnable taskExecutionRunnable);
+}

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();
+
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/ITaskExecutionRunnable.java

@@ -0,0 +1,62 @@
+/*
+ * 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;
+
+/**
+ * The TaskExecutionRunnable represent the running task, it is responsible for operate the task instance. e.g. dispatch, kill, pause.
+ */
+public interface ITaskExecutionRunnable extends IEventfulExecutionRunnable {
+
+    /**
+     * Start the task instance.
+     */
+    void start();
+
+    /**
+     * Kill the task instance.
+     */
+    void kill();
+
+    /**
+     * Pause the task instance.
+     */
+    void pause();
+
+    /**
+     * Get the task execution identify.
+     *
+     * @return the task execution identify
+     */
+    ITaskExecutionRunnableIdentify getIdentify();
+
+    /**
+     * Get the task execution context.
+     *
+     * @return the task execution context
+     */
+    ITaskExecutionContext getTaskExecutionContext();
+
+    /**
+     * Determine whether the current task is ready to trigger the post task node.
+     *
+     * @param taskNodeName post task name
+     * @return true if the current task can be accessed to the post task.
+     */
+    boolean isReadyToTrigger(String taskNodeName);
+
+}

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();
+
+}

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();
+
+}

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);
+
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/ITaskExecutionRunnableFactory.java

@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.dolphinscheduler.workflow.engine.workflow;
+
+import org.apache.dolphinscheduler.workflow.engine.dag.ITaskIdentify;
+
+public interface ITaskExecutionRunnableFactory {
+
+    ITaskExecutionRunnable createTaskExecutionRunnable(ITaskIdentify taskIdentify,
+                                                       IWorkflowExecutionContext workflowExecutionContext);
+
+    ITaskExecutionRunnable createFailoverTaskExecutionRunnable(ITaskExecutionRunnable taskExecutionRunnable,
+                                                               IWorkflowExecutionContext workflowExecutionContext);
+
+    ITaskExecutionRunnable createRetryTaskExecutionRunnable(ITaskExecutionRunnable taskExecutionRunnable,
+                                                            IWorkflowExecutionContext workflowExecutionContext);
+}

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();
+
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/ITaskExecutionRunnableRepository.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;
+
+import java.util.Collection;
+
+public interface ITaskExecutionRunnableRepository {
+
+    void storeTaskExecutionRunnable(ITaskExecutionRunnable taskExecutionRunnable);
+
+    ITaskExecutionRunnable getTaskExecutionRunnableById(Integer taskInstanceId);
+
+    ITaskExecutionRunnable getTaskExecutionRunnableByName(String taskInstanceName);
+
+    Collection<ITaskExecutionRunnable> getActiveTaskExecutionRunnable();
+
+    void removeTaskExecutionRunnable(Integer taskInstanceId);
+
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/IWorkflowExecutionContext.java

@@ -0,0 +1,38 @@
+/*
+ * 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;
+import org.apache.dolphinscheduler.workflow.engine.dag.WorkflowDAG;
+import org.apache.dolphinscheduler.workflow.engine.event.IEventRepository;
+
+import java.util.List;
+
+public interface IWorkflowExecutionContext {
+
+    IWorkflowExecutionRunnableIdentify getIdentify();
+
+    List<ITaskIdentify> getStartTaskIdentifies();
+
+    WorkflowDAG getWorkflowDAG();
+
+    WorkflowExecutionDAG getWorkflowExecutionDAG();
+
+    IEventRepository getEventRepository();
+
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/IWorkflowExecutionDAG.java

@@ -0,0 +1,36 @@
+/*
+ * 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.DAG;
+
+import java.util.List;
+
+/**
+ * The WorkflowExecutionDAG represent the running workflow DAG.
+ */
+public interface IWorkflowExecutionDAG extends DAG<ITaskExecutionRunnable, ITaskExecutionRunnableIdentify> {
+
+    /**
+     * Get TaskExecutionRunnable which is not finished.
+     *
+     * @return TaskExecutionRunnable
+     */
+    List<ITaskExecutionRunnable> getActiveTaskExecutionRunnable();
+
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/IWorkflowExecutionDAGFactory.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 IWorkflowExecutionDAGFactory {
+
+    IWorkflowExecutionDAG createWorkflowExecutionDAG(IWorkflowExecutionContext workflowExecutionContext);
+
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/IWorkflowExecutionRunnable.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.workflow;
+
+/**
+ * The IWorkflowExecuteRunnable represent a running workflow instance, it is responsible for operate the workflow instance. e.g. start, kill, pause.
+ */
+public interface IWorkflowExecutionRunnable
+        extends
+            IWorkflowOuterAction,
+            IWorkflowInnerAction,
+            IEventfulExecutionRunnable {
+
+    /**
+     * Get the identity of the workflow execution runnable.
+     *
+     * @return the identity of the workflow execution runnable
+     */
+    IWorkflowExecutionRunnableIdentify getIdentity();
+
+    /**
+     * Get the workflow execution context.
+     *
+     * @return the workflow execution context
+     */
+    IWorkflowExecutionContext getWorkflowExecutionContext();
+
+}

dolphinscheduler-workflow-engine/src/main/java/org/apache/dolphinscheduler/workflow/engine/workflow/IWorkflowExecutionRunnableDelegate.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 IWorkflowExecutionRunnableDelegate {
+
+    void start();
+
+    void pause();
+
+    void kill();
+
+}