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

@@ -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();

@@ -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(),

@@ -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(),

@@ -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"),
     /**

@@ -56,4 +56,8 @@ public static WorkflowExecutionTypeEnum of(int executionType) {
         throw new IllegalArgumentException("invalid status : " + executionType);
     }
 
+    public boolean isSerial() {
+        return this != PARALLEL;
+    }
+
 }
@@ -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;
+
+}
@@ -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())) {

@@ -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.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_workflow_task_lineage")
+public class WorkflowSerialQueue {
+
+    @TableId(value = "id", type = IdType.AUTO)
+    private Integer id;
+
+    private long workflowDefinitionCode;
+
+    private int workflowInstanceId;
+
+    private int state;
+
+    private Date createTime;
+
+    private Date updateTime;
+
+}
@@ -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);
+
+    void deleteByWorkflowInstanceId(@Param("workflowInstanceId") int workflowInstanceId);
+
+}
@@ -0,0 +1,25 @@
+/*
+ * 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.WorkflowSerialQueue;
+
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+
+public interface WorkflowSerialQueueMapper extends BaseMapper<WorkflowSerialQueue> {
+}