Merge branch 'dev' into Fix-17701

Author: njnu-seafish

.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-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/WorkflowLineageServiceImpl.java

@@ -242,29 +242,38 @@ public List<DependentWorkflowDefinition> queryDownstreamDependentWorkflowDefinit
 
         List<WorkflowDefinition> workflowDefinitionList =
                 workflowDefinitionMapper.queryByCodes(workflowTaskLineageList.stream()
-                        .map(WorkflowTaskLineage::getDeptWorkflowDefinitionCode).distinct()
+                        .map(WorkflowTaskLineage::getWorkflowDefinitionCode).distinct()
                         .collect(Collectors.toList()));
         List<TaskDefinition> taskDefinitionList = taskDefinitionMapper.queryByCodeList(workflowTaskLineageList.stream()
-                .map(WorkflowTaskLineage::getDeptTaskDefinitionCode).distinct().collect(Collectors.toList()));
-        for (TaskDefinition taskDefinition : taskDefinitionList) {
+                .map(WorkflowTaskLineage::getTaskDefinitionCode).filter(code -> code != 0).distinct()
+                .collect(Collectors.toList()));
+
+        for (WorkflowTaskLineage workflowLineage : workflowTaskLineageList) {
             DependentWorkflowDefinition dependentWorkflowDefinition = new DependentWorkflowDefinition();
-            workflowTaskLineageList.stream()
-                    .filter(workflowLineage -> workflowLineage.getDeptTaskDefinitionCode() == taskDefinition.getCode())
-                    .findFirst()
-                    .ifPresent(workflowLineage -> {
-                        dependentWorkflowDefinition
-                                .setWorkflowDefinitionCode(workflowLineage.getDeptWorkflowDefinitionCode());
-                        dependentWorkflowDefinition.setTaskDefinitionCode(taskDefinition.getCode());
-                        dependentWorkflowDefinition.setTaskParams(taskDefinition.getTaskParams());
-                        dependentWorkflowDefinition.setWorkerGroup(taskDefinition.getWorkerGroup());
-                    });
+            dependentWorkflowDefinition.setWorkflowDefinitionCode(workflowLineage.getWorkflowDefinitionCode());
+            dependentWorkflowDefinition.setTaskDefinitionCode(workflowLineage.getTaskDefinitionCode());
+
+            // If taskDefinitionCode is 0, it means dependency on entire workflow, taskParams and workerGroup remain
+            // null
+            if (workflowLineage.getTaskDefinitionCode() != 0) {
+                taskDefinitionList.stream()
+                        .filter(taskDefinition -> taskDefinition.getCode() == workflowLineage.getTaskDefinitionCode())
+                        .findFirst()
+                        .ifPresent(taskDefinition -> {
+                            dependentWorkflowDefinition.setTaskParams(taskDefinition.getTaskParams());
+                            dependentWorkflowDefinition.setWorkerGroup(taskDefinition.getWorkerGroup());
+                        });
+            }
+
             workflowDefinitionList.stream()
-                    .filter(workflowDefinition -> workflowDefinition.getCode() == dependentWorkflowDefinition
+                    .filter(workflowDefinition -> workflowDefinition.getCode() == workflowLineage
                             .getWorkflowDefinitionCode())
                     .findFirst()
                     .ifPresent(workflowDefinition -> {
                         dependentWorkflowDefinition.setWorkflowDefinitionVersion(workflowDefinition.getVersion());
                     });
+
+            dependentWorkflowDefinitionList.add(dependentWorkflowDefinition);
         }
 
         return dependentWorkflowDefinitionList;

dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/impl/WorkflowLineageServiceImplTest.java

@@ -0,0 +1,146 @@
+/*
+ * 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.api.service.impl;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.when;
+
+import org.apache.dolphinscheduler.common.constants.Constants;
+import org.apache.dolphinscheduler.dao.entity.DependentWorkflowDefinition;
+import org.apache.dolphinscheduler.dao.entity.TaskDefinition;
+import org.apache.dolphinscheduler.dao.entity.WorkflowDefinition;
+import org.apache.dolphinscheduler.dao.entity.WorkflowTaskLineage;
+import org.apache.dolphinscheduler.dao.mapper.TaskDefinitionMapper;
+import org.apache.dolphinscheduler.dao.mapper.WorkflowDefinitionMapper;
+import org.apache.dolphinscheduler.dao.repository.WorkflowTaskLineageDao;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.springframework.test.util.ReflectionTestUtils;
+
+@ExtendWith(MockitoExtension.class)
+class WorkflowLineageServiceImplTest {
+
+    @InjectMocks
+    private WorkflowLineageServiceImpl workflowLineageService;
+
+    @Mock
+    private WorkflowTaskLineageDao workflowTaskLineageDao;
+
+    @Mock
+    private WorkflowDefinitionMapper workflowDefinitionMapper;
+
+    @Mock
+    private TaskDefinitionMapper taskDefinitionMapper;
+
+    @BeforeEach
+    void setUp() {
+        ReflectionTestUtils.setField(workflowLineageService, "workflowTaskLineageDao", workflowTaskLineageDao);
+        ReflectionTestUtils.setField(workflowLineageService, "workflowDefinitionMapper", workflowDefinitionMapper);
+        ReflectionTestUtils.setField(workflowLineageService, "taskDefinitionMapper", taskDefinitionMapper);
+    }
+
+    @Test
+    void shouldReturnEmptyListWhenNoLineageExist() {
+        long workflowCode = 100L;
+        when(workflowTaskLineageDao
+                .queryWorkFlowLineageByDept(Constants.DEFAULT_PROJECT_CODE, workflowCode, Constants.DEPENDENT_ALL_TASK))
+                        .thenReturn(Collections.emptyList());
+
+        List<DependentWorkflowDefinition> result =
+                workflowLineageService.queryDownstreamDependentWorkflowDefinitions(workflowCode);
+
+        assertThat(result).isEmpty();
+        verifyNoInteractions(workflowDefinitionMapper, taskDefinitionMapper);
+    }
+
+    @Test
+    void shouldBuildDependentWorkflowDefinitions() {
+        long upstreamWorkflowCode = 1L;
+
+        WorkflowTaskLineage taskLineage = new WorkflowTaskLineage();
+        taskLineage.setWorkflowDefinitionCode(200L);
+        taskLineage.setDeptWorkflowDefinitionCode(upstreamWorkflowCode);
+        taskLineage.setTaskDefinitionCode(300L);
+        taskLineage.setDeptTaskDefinitionCode(0L);
+
+        WorkflowTaskLineage workflowLineage = new WorkflowTaskLineage();
+        workflowLineage.setWorkflowDefinitionCode(201L);
+        workflowLineage.setDeptWorkflowDefinitionCode(upstreamWorkflowCode);
+        workflowLineage.setTaskDefinitionCode(0L);
+
+        when(workflowTaskLineageDao
+                .queryWorkFlowLineageByDept(Constants.DEFAULT_PROJECT_CODE, upstreamWorkflowCode,
+                        Constants.DEPENDENT_ALL_TASK))
+                                .thenReturn(Arrays.asList(taskLineage, workflowLineage));
+
+        WorkflowDefinition workflowDefinition200 = new WorkflowDefinition();
+        workflowDefinition200.setCode(200L);
+        workflowDefinition200.setVersion(3);
+
+        WorkflowDefinition workflowDefinition201 = new WorkflowDefinition();
+        workflowDefinition201.setCode(201L);
+        workflowDefinition201.setVersion(4);
+
+        when(workflowDefinitionMapper.queryByCodes(Arrays.asList(200L, 201L)))
+                .thenReturn(Arrays.asList(workflowDefinition200, workflowDefinition201));
+
+        TaskDefinition taskDefinition = new TaskDefinition();
+        taskDefinition.setCode(300L);
+        taskDefinition.setTaskParams("task-params");
+        taskDefinition.setWorkerGroup("test-group");
+
+        when(taskDefinitionMapper.queryByCodeList(Collections.singletonList(300L)))
+                .thenReturn(Collections.singletonList(taskDefinition));
+
+        List<DependentWorkflowDefinition> result =
+                workflowLineageService.queryDownstreamDependentWorkflowDefinitions(upstreamWorkflowCode);
+
+        assertThat(result).hasSize(2);
+
+        DependentWorkflowDefinition taskDependent = result.stream()
+                .filter(dependent -> dependent.getWorkflowDefinitionCode() == 200L)
+                .findFirst()
+                .orElseThrow(() -> new AssertionError("Expected DependentWorkflowDefinition with code 200 not found"));
+        assertThat(taskDependent.getTaskDefinitionCode()).isEqualTo(300L);
+        assertThat(taskDependent.getTaskParams()).isEqualTo("task-params");
+        assertThat(taskDependent.getWorkerGroup()).isEqualTo("test-group");
+        assertThat(taskDependent.getWorkflowDefinitionVersion()).isEqualTo(3);
+
+        DependentWorkflowDefinition workflowDependent = result.stream()
+                .filter(dependent -> dependent.getWorkflowDefinitionCode() == 201L)
+                .findFirst()
+                .orElseThrow(() -> new AssertionError("Expected DependentWorkflowDefinition with code 201 not found"));
+        assertThat(workflowDependent.getTaskDefinitionCode()).isEqualTo(0L);
+        assertThat(workflowDependent.getTaskParams()).isNull();
+        assertThat(workflowDependent.getWorkerGroup()).isNull();
+        assertThat(workflowDependent.getWorkflowDefinitionVersion()).isEqualTo(4);
+
+        verify(taskDefinitionMapper).queryByCodeList(Collections.singletonList(300L));
+    }
+}

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