[DSIP-92][Master] Refactor workflow serial strategy (#17531)

Author: ruanwenjun

.github/workflows/backend.yml

@@ -143,7 +143,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        version: ["3.1.9", "3.2.0"]
+        version: ["3.1.9", "3.2.0", "3.3.1"]
         case:
           - name: schema-check-with-mysql
             script: .github/workflows/schema-check/mysql/start-job.sh

docs/docs/en/guide/upgrade/incompatible.md

@@ -2,29 +2,23 @@
 
 This document records the incompatible updates between each version. You need to check this document before you upgrade to related version.
 
-## dev
-
-* Upgrade mysql driver version from 8.0.16 to 8.0.33 ([#14684](https://github.com/apache/dolphinscheduler/pull/14684))
-* Change env `PYTHON_HOME` to `PYTHON_LAUNCHER` and `DATAX_HOME` to `DATAX_LAUNCHER` ([#14523](https://github.com/apache/dolphinscheduler/pull/14523))
-* Change regex matching sql params in SQL task plugin ([#13378](https://github.com/apache/dolphinscheduler/pull/13378))
-* Remove the spark version of spark task ([#11860](https://github.com/apache/dolphinscheduler/pull/11860)).
-* Change the default unix shell executor from sh to bash ([#12180](https://github.com/apache/dolphinscheduler/pull/12180)).
-* Remove `deleteSource` in `download()` of `StorageOperate` ([#14084](https://github.com/apache/dolphinscheduler/pull/14084))
-* Remove default key for attribute `data-quality.jar.name` in `common.properties` ([#15551](https://github.com/apache/dolphinscheduler/pull/15551))
-* Rename attribute `data-quality.jar.name` to `data-quality.jar.dir` in `common.properties` and represent for directory ([#15563](https://github.com/apache/dolphinscheduler/pull/15563))
-
-## 3.2.0
-
-* Remove parameter `description` from public interfaces of new resource center  ([#14394](https://github.com/apache/dolphinscheduler/pull/14394))
-
 ## 3.0.0
 
 * Copy and import workflow without 'copy' suffix [#10607](https://github.com/apache/dolphinscheduler/pull/10607)
 * Use semicolon as default sql segment separator [#10869](https://github.com/apache/dolphinscheduler/pull/10869)
 
 ## 3.2.0
 
+* Rename attribute `data-quality.jar.name` to `data-quality.jar.dir` in `common.properties` and represent for directory ([#15563](https://github.com/apache/dolphinscheduler/pull/15563))
+* Remove default key for attribute `data-quality.jar.name` in `common.properties` ([#15551](https://github.com/apache/dolphinscheduler/pull/15551))
+* Remove `deleteSource` in `download()` of `StorageOperate` ([#14084](https://github.com/apache/dolphinscheduler/pull/14084))
+* Change the default unix shell executor from sh to bash ([#12180](https://github.com/apache/dolphinscheduler/pull/12180)).
+* Remove the spark version of spark task ([#11860](https://github.com/apache/dolphinscheduler/pull/11860)).
+* Change regex matching sql params in SQL task plugin ([#13378](https://github.com/apache/dolphinscheduler/pull/13378))
+* Change env `PYTHON_HOME` to `PYTHON_LAUNCHER` and `DATAX_HOME` to `DATAX_LAUNCHER` ([#14523](https://github.com/apache/dolphinscheduler/pull/14523))
+* Upgrade mysql driver version from 8.0.16 to 8.0.33 ([#14684](https://github.com/apache/dolphinscheduler/pull/14684))
 * Add required field `database` in /datasources/tables && /datasources/tableColumns Api [#14406](https://github.com/apache/dolphinscheduler/pull/14406)
+* Remove parameter `description` from public interfaces of new resource center  ([#14394](https://github.com/apache/dolphinscheduler/pull/14394))
 
 ## 3.3.0
 
@@ -41,4 +35,5 @@ This document records the incompatible updates between each version. You need to
 ## 3.4.0
 
 * Renamed the publicKey field to privateKey in the SSH connection parameters under the datasource configuration. ([#17666])(https://github.com/apache/dolphinscheduler/pull/17666)
+* Add table t_ds_serial_command. ([#17531])(https://github.com/apache/dolphinscheduler/pull/17531)
 

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/dto/workflow/WorkflowCreateRequest.java

@@ -56,7 +56,7 @@ public class WorkflowCreateRequest {
     private int timeout;
 
     @Schema(allowableValues = "PARALLEL / SERIAL_WAIT / SERIAL_DISCARD / SERIAL_PRIORITY", example = "PARALLEL", description = "default PARALLEL if not provide.")
-    private String executionType;
+    private String executionType = WorkflowExecutionTypeEnum.PARALLEL.name();
 
     public WorkflowDefinition convert2WorkflowDefinition() {
         WorkflowDefinition workflowDefinition = new WorkflowDefinition();

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/executor/workflow/PauseWorkflowInstanceExecutorDelegate.java

@@ -21,6 +21,7 @@
 import org.apache.dolphinscheduler.common.enums.WorkflowExecutionStatus;
 import org.apache.dolphinscheduler.dao.entity.User;
 import org.apache.dolphinscheduler.dao.entity.WorkflowInstance;
+import org.apache.dolphinscheduler.dao.repository.SerialCommandDao;
 import org.apache.dolphinscheduler.dao.repository.WorkflowInstanceDao;
 import org.apache.dolphinscheduler.extract.base.client.Clients;
 import org.apache.dolphinscheduler.extract.master.IWorkflowControlClient;
@@ -31,6 +32,7 @@
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
+import org.springframework.transaction.support.TransactionTemplate;
 
 @Slf4j
 @Component
@@ -41,6 +43,12 @@ public class PauseWorkflowInstanceExecutorDelegate
     @Autowired
     private WorkflowInstanceDao workflowInstanceDao;
 
+    @Autowired
+    private TransactionTemplate transactionTemplate;
+
+    @Autowired
+    private SerialCommandDao serialCommandDao;
+
     @Override
     public Void execute(PauseWorkflowInstanceOperation workflowInstanceControlRequest) {
         final WorkflowInstance workflowInstance = workflowInstanceControlRequest.workflowInstance;
@@ -64,10 +72,15 @@ private void exceptionIfWorkflowInstanceCannotPause(WorkflowInstance workflowIns
     }
 
     private void directPauseInDB(WorkflowInstance workflowInstance) {
-        workflowInstanceDao.updateWorkflowInstanceState(
-                workflowInstance.getId(),
-                workflowInstance.getState(),
-                WorkflowExecutionStatus.PAUSE);
+        // todo: move the pause logic to master
+        transactionTemplate.execute(status -> {
+            workflowInstanceDao.updateWorkflowInstanceState(
+                    workflowInstance.getId(),
+                    workflowInstance.getState(),
+                    WorkflowExecutionStatus.PAUSE);
+            serialCommandDao.deleteByWorkflowInstanceId(workflowInstance.getId());
+            return null;
+        });
         log.info("Update workflow instance {} state from: {} to {} success",
                 workflowInstance.getName(),
                 workflowInstance.getState().name(),

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/executor/workflow/StopWorkflowInstanceExecutorDelegate.java

@@ -21,6 +21,7 @@
 import org.apache.dolphinscheduler.common.enums.WorkflowExecutionStatus;
 import org.apache.dolphinscheduler.dao.entity.User;
 import org.apache.dolphinscheduler.dao.entity.WorkflowInstance;
+import org.apache.dolphinscheduler.dao.repository.SerialCommandDao;
 import org.apache.dolphinscheduler.dao.repository.WorkflowInstanceDao;
 import org.apache.dolphinscheduler.extract.base.client.Clients;
 import org.apache.dolphinscheduler.extract.master.IWorkflowControlClient;
@@ -31,6 +32,7 @@
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
+import org.springframework.transaction.support.TransactionTemplate;
 
 @Slf4j
 @Component
@@ -41,6 +43,12 @@ public class StopWorkflowInstanceExecutorDelegate
     @Autowired
     private WorkflowInstanceDao workflowInstanceDao;
 
+    @Autowired
+    private TransactionTemplate transactionTemplate;
+
+    @Autowired
+    private SerialCommandDao serialCommandDao;
+
     @Override
     public Void execute(StopWorkflowInstanceOperation workflowInstanceControlRequest) {
         final WorkflowInstance workflowInstance = workflowInstanceControlRequest.workflowInstance;
@@ -65,10 +73,15 @@ void exceptionIfWorkflowInstanceCannotStop(WorkflowInstance workflowInstance) {
     }
 
     void directStopInDB(WorkflowInstance workflowInstance) {
-        workflowInstanceDao.updateWorkflowInstanceState(
-                workflowInstance.getId(),
-                workflowInstance.getState(),
-                WorkflowExecutionStatus.STOP);
+        // todo: move the stop logic to master
+        transactionTemplate.execute(status -> {
+            workflowInstanceDao.updateWorkflowInstanceState(
+                    workflowInstance.getId(),
+                    workflowInstance.getState(),
+                    WorkflowExecutionStatus.STOP);
+            serialCommandDao.deleteByWorkflowInstanceId(workflowInstance.getId());
+            return null;
+        });
         log.info("Update workflow instance {} state from: {} to {} success",
                 workflowInstance.getName(),
                 workflowInstance.getState().name(),

dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/CommandType.java

@@ -89,7 +89,6 @@ public enum CommandType {
     STOP(9, "stop a workflow"),
     /**
      * Recover from the serial-wait state.
-     * todo: We may need to remove these command, and use the workflow instance origin command type when notify from serial wait.
      */
     RECOVER_SERIAL_WAIT(11, "recover serial wait"),
     /**

dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/WorkflowExecutionTypeEnum.java

@@ -27,7 +27,6 @@
 public enum WorkflowExecutionTypeEnum {
 
     PARALLEL(0, "parallel"),
-    // todo: the serial is unstable, so we don't support them now
     SERIAL_WAIT(1, "serial wait"),
     SERIAL_DISCARD(2, "serial discard"),
     SERIAL_PRIORITY(3, "serial priority");
@@ -56,4 +55,8 @@ public static WorkflowExecutionTypeEnum of(int executionType) {
         throw new IllegalArgumentException("invalid status : " + executionType);
     }
 
+    public boolean isSerial() {
+        return this != PARALLEL;
+    }
+
 }

dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/SerialCommand.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.dao.entity;
+
+import java.util.Date;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import com.baomidou.mybatisplus.annotation.IdType;
+import com.baomidou.mybatisplus.annotation.TableId;
+import com.baomidou.mybatisplus.annotation.TableName;
+
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+@TableName("t_ds_serial_command")
+public class SerialCommand {
+
+    @TableId(value = "id", type = IdType.AUTO)
+    private Integer id;
+
+    private Integer workflowInstanceId;
+
+    private Long workflowDefinitionCode;
+
+    private Integer workflowDefinitionVersion;
+
+    private String command;
+
+    private int state;
+
+    private Date createTime;
+
+    private Date updateTime;
+
+}

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

@@ -210,11 +210,6 @@ public CommandType getCmdTypeIfComplement() {
         return commandType;
     }
 
-    /**
-     * set state with desc
-     * @param state
-     * @param stateDesc
-     */
     public void setStateWithDesc(WorkflowExecutionStatus state, String stateDesc) {
         this.setState(state);
         if (StringUtils.isEmpty(this.getStateHistory())) {

dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/SerialCommandMapper.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.dao.mapper;
+
+import org.apache.dolphinscheduler.dao.entity.SerialCommand;
+
+import org.apache.ibatis.annotations.Param;
+
+import java.util.List;
+
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+
+public interface SerialCommandMapper extends BaseMapper<SerialCommand> {
+
+    List<SerialCommand> fetchSerialCommands(@Param("fetchSize") int fetchSize);
+
+    int deleteByWorkflowInstanceId(@Param("workflowInstanceId") int workflowInstanceId);
+
+}