[DSIP-61][Master] Refactor thread pool and state event orchestration …

…in master (apache#16327)
qlangtech · Aug 26, 2024 · 9448806 · 9448806
1 parent e8098ac
commit 9448806
Show file tree

Hide file tree

Showing 519 changed files with 23,066 additions and 15,946 deletions.
diff --git a/.github/workflows/unit-test.yml b/.github/workflows/unit-test.yml
@@ -111,7 +111,7 @@ jobs:
   result:
     name: Unit Test
     runs-on: ubuntu-latest
-    timeout-minutes: 30
+    timeout-minutes: 60
     needs: [ unit-test, paths-filter ]
     if: always()
     steps:

diff --git a/docs/docs/en/guide/project/workflow-instance.md b/docs/docs/en/guide/project/workflow-instance.md
@@ -2,47 +2,73 @@
 
 ## View Workflow Instance
 
-Click `Project Management -> Workflow -> Workflow Instance`, enter the Workflow Instance page, as shown in the following figure:
+Click `Project Management -> Workflow -> Workflow Instance`, enter the Workflow Instance page, as shown in the following
+figure:
 
 ![workflow-instance](../../../../img/new_ui/dev/project/workflow-instance.png)
 
-Click the workflow name to enter the DAG view page, and check the task execution status, as shown in the following figure:
+Click the workflow name to enter the DAG view page, and check the task execution status, as shown in the following
+figure:
 
 ![instance-state](../../../../img/new_ui/dev/project/instance-state.png)
 
 ## View Task Log
 
-Enter the workflow instance page, click the workflow name, enter the DAG view page, double-click the task node, as shown in the following figure:
+Enter the workflow instance page, click the workflow name, enter the DAG view page, double-click the task node, as shown
+in the following figure:
 
 ![instance-log01](../../../../img/new_ui/dev/project/instance-log01.png)
 
-Click "View Log", a log window pops up, as shown in the figure below, you can also view the task log on the task instance page, refer to [Task View Log](./task-instance.md)
+Click "View Log", a log window pops up, as shown in the figure below, you can also view the task log on the task
+instance page, refer to [Task View Log](./task-instance.md)
 
 ![instance-log02](../../../../img/new_ui/dev/project/instance-log02.png)
 
 ## View Task History
 
-Click `Project Management -> Workflow -> Workflow Instance` to enter the workflow instance page, click the workflow name to enter the workflow DAG page;
+Click `Project Management -> Workflow -> Workflow Instance` to enter the workflow instance page, click the workflow name
+to enter the workflow DAG page;
 
-Double-click the task node, click `View History` to jump to the task instance page, and display the list of task instances run by the task definition.
+Double-click the task node, click `View History` to jump to the task instance page, and display the list of task
+instances run by the task definition.
 
 ![instance-history](../../../../img/new_ui/dev/project/instance-history.png)
 
 ## View Running Parameters
 
-Click `Project Management -> Workflow -> Workflow Instance` to enter the workflow instance page, click the workflow name to enter the workflow DAG page;
+Click `Project Management -> Workflow -> Workflow Instance` to enter the workflow instance page, click the workflow name
+to enter the workflow DAG page;
 
-Click the icon in the upper left corner <img src="../../../../img/run_params_button.png" width="35"/> to view the startup parameters of the workflow instance; click the icon <img src="../../../../img/global_param.png" width="35"/> to view the global parameters and local parameters of the workflow instance, as shown in the following figure:
+Click the icon in the upper left corner <img src="../../../../img/run_params_button.png" width="35"/> to view the
+startup parameters of the workflow instance; click the icon <img src="../../../../img/global_param.png" width="35"/> to
+view the global parameters and local parameters of the workflow instance, as shown in the following figure:
 
 ![instance-parameter](../../../../img/new_ui/dev/project/instance-parameter.png)
 
 ## Workflow Instance Operation Function
 
-Click `Project Management -> Workflow -> Workflow Instance`, enter the workflow instance page, as shown in the following figure:
+Click `Project Management -> Workflow -> Workflow Instance`, enter the workflow instance page, as shown in the following
+figure:
 
 ![workflow-instance](../../../../img/new_ui/dev/project/workflow-instance.png)
 
-- **Edit:** Only processes with success/failed/stop status can be edited. Click the "Edit" button or the workflow instance name to enter the DAG edit page. After the edit, click the "Save" button to confirm, as shown in the figure below. In the pop-up box, check "Whether to update the workflow definition", after saving, the information modified by the instance will be updated to the workflow definition; if not checked, the workflow definition would not be updated.
+| WorkflowInstanceState \ Operation | Edit | Rerun | Stop | Pause | Resume Suspend | Delete | Gantt Chart |
+|-----------------------------------|------|-------|------|-------|----------------|--------|-------------|
+| SUBMITTED_SUCCESS                 |      |       | √    | √     |                |        | √           |
+| SERIAL_WAIT                       |      |       | √    |       |                |        | √           |
+| WAIT_TO_RUN                       |      |       | √    |       |                |        | √           |
+| Executing                         |      |       | √    | √     |                |        | √           |
+| READY PAUSE                       |      |       |      |       |                |        | √           |
+| PAUSE                             | √    | √     |      |       | √              | √      | √           |
+| READY STOP                        |      |       |      |       |                |        | √           |
+| STOP                              | √    | √     |      |       | √              | √      | √           |
+| FAILURE                           | √    | √     |      |       |                | √      | √           |
+| SUCCESS                           | √    | √     |      |       |                | √      | √           |
+
+- **Edit:** Only processes with success/failed/stop status can be edited. Click the "Edit" button or the workflow
+  instance name to enter the DAG edit page. After the edit, click the "Save" button to confirm, as shown in the figure
+  below. In the pop-up box, check "Whether to update the workflow definition", after saving, the information modified by
+  the instance will be updated to the workflow definition; if not checked, the workflow definition would not be updated.
 
      <p align="center">
        <img src="../../../../img/editDag-en.png" width="80%" />
@@ -52,15 +78,18 @@ Click `Project Management -> Workflow -> Workflow Instance`, enter the workflow
 
 - **Recovery Failed:** For failed processes, you can perform failure recovery operations, starting from the failed node
 
-- **Stop:** **Stop** the running process, the background code will first `kill` the worker process, and then execute `kill -9` operation
+- **Stop:** **Stop** the running process, the background code will first `kill` the worker process, and then
+  execute `kill -9` operation
 
-- **Pause:** **Pause** the running process, the system status will change to **waiting for execution**, it will wait for the task to finish, and pause the next sequence task.
+- **Pause:** **Pause** the running process, the system status will change to **waiting for execution**, it will wait for
+  the task to finish, and pause the next sequence task.
 
 - **Resume pause:** Resume the paused process, start running directly from the **paused node**
 
 - **Delete:** Delete the workflow instance and the task instance under the workflow instance
 
-- **Gantt Chart:** The vertical axis of the Gantt chart is the topological sorting of task instances of the workflow instance, and the horizontal axis is the running time of the task instances, as shown in the figure:
+- **Gantt Chart:** The vertical axis of the Gantt chart is the topological sorting of task instances of the workflow
+  instance, and the horizontal axis is the running time of the task instances, as shown in the figure:
 
 ![instance-gantt](../../../../img/new_ui/dev/project/instance-gantt.png)
 
diff --git a/...ert-server/src/main/java/org/apache/dolphinscheduler/alert/plugin/AlertPluginManager.java b/...ert-server/src/main/java/org/apache/dolphinscheduler/alert/plugin/AlertPluginManager.java
@@ -84,12 +84,8 @@ private void installAlertPlugin() {
             String name = entry.getKey();
             AlertChannelFactory factory = entry.getValue();
 
-            log.info("Registering alert plugin: {} - {}", name, factory.getClass().getSimpleName());
-
             final AlertChannel alertChannel = factory.create();
 
-            log.info("Registered alert plugin: {} - {}", name, factory.getClass().getSimpleName());
-
             final List<PluginParams> params = new ArrayList<>(factory.params());
 
             final String paramsJson = PluginParamsTransfer.transferParamsToJson(params);
@@ -99,6 +95,8 @@ private void installAlertPlugin() {
             final int id = pluginDao.addOrUpdatePluginDefine(pluginDefine);
 
             alertPluginMap.put(id, alertChannel);
+
+            log.info("Success register alert plugin: {}", name);
         }
     }
 

diff --git a/...i-test-case/src/test/java/org/apache/dolphinscheduler/api/test/cases/ExecutorAPITest.java b/...i-test-case/src/test/java/org/apache/dolphinscheduler/api/test/cases/ExecutorAPITest.java
@@ -145,12 +145,4 @@ public void testStartProcessInstance() {
         }
     }
 
-    @Test
-    @Order(2)
-    public void testStartCheckProcessDefinition() {
-        HttpResponse testStartCheckProcessDefinitionResponse =
-                executorPage.startCheckProcessDefinition(loginUser, projectCode, processDefinitionCode);
-        Assertions.assertTrue(testStartCheckProcessDefinitionResponse.getBody().getSuccess());
-    }
-
 }
diff --git a/...case/src/test/java/org/apache/dolphinscheduler/api/test/cases/ProcessInstanceAPITest.java b/...case/src/test/java/org/apache/dolphinscheduler/api/test/cases/ProcessInstanceAPITest.java
@@ -19,7 +19,6 @@
 
 package org.apache.dolphinscheduler.api.test.cases;
 
-import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import org.apache.dolphinscheduler.api.test.core.DolphinScheduler;
@@ -37,7 +36,6 @@
 import org.apache.dolphinscheduler.common.enums.WarningType;
 import org.apache.dolphinscheduler.dao.entity.User;
 
-import org.apache.commons.collections4.CollectionUtils;
 import org.apache.http.client.methods.CloseableHttpResponse;
 import org.apache.http.util.EntityUtils;
 
@@ -53,6 +51,7 @@
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Order;
 import org.junit.jupiter.api.Test;
 import org.testcontainers.shaded.org.awaitility.Awaitility;
@@ -81,8 +80,6 @@ public class ProcessInstanceAPITest {
 
     private static long processDefinitionCode;
 
-    private static long triggerCode;
-
     private static int processInstanceId;
 
     @BeforeAll
@@ -154,16 +151,11 @@ public void testQueryProcessInstancesByTriggerCode() {
                     .atMost(30, TimeUnit.SECONDS)
                     .untilAsserted(() -> {
                         // query workflow instance by trigger code
-                        triggerCode = (long) startProcessInstanceResponse.getBody().getData();
-                        HttpResponse queryProcessInstancesByTriggerCodeResponse = processInstancePage
-                                .queryProcessInstancesByTriggerCode(loginUser, projectCode, triggerCode);
-                        assertTrue(queryProcessInstancesByTriggerCodeResponse.getBody().getSuccess());
-                        List<LinkedHashMap<String, Object>> body =
-                                (List<LinkedHashMap<String, Object>>) queryProcessInstancesByTriggerCodeResponse
-                                        .getBody().getData();
-                        assertTrue(CollectionUtils.isNotEmpty(body));
-                        assertEquals("SUCCESS", body.get(0).get("state"));
-                        processInstanceId = (int) body.get(0).get("id");
+                        HttpResponse queryProcessInstanceListResponse =
+                                processInstancePage.queryProcessInstanceList(loginUser, projectCode, 1, 10);
+                        assertTrue(queryProcessInstanceListResponse.getBody().getSuccess());
+                        assertTrue(queryProcessInstanceListResponse.getBody().getData().toString()
+                                .contains("test_import"));
                     });
         } catch (Exception e) {
             log.error("failed", e);
@@ -182,6 +174,7 @@ public void testQueryProcessInstanceList() {
 
     @Test
     @Order(3)
+    @Disabled
     public void testQueryTaskListByProcessId() {
         HttpResponse queryTaskListByProcessIdResponse =
                 processInstancePage.queryTaskListByProcessId(loginUser, projectCode, processInstanceId);
@@ -191,6 +184,7 @@ public void testQueryTaskListByProcessId() {
 
     @Test
     @Order(4)
+    @Disabled
     public void testQueryProcessInstanceById() {
         HttpResponse queryProcessInstanceByIdResponse =
                 processInstancePage.queryProcessInstanceById(loginUser, projectCode, processInstanceId);
@@ -200,6 +194,7 @@ public void testQueryProcessInstanceById() {
 
     @Test
     @Order(5)
+    @Disabled
     public void testDeleteProcessInstanceById() {
         HttpResponse deleteProcessInstanceByIdResponse =
                 processInstancePage.deleteProcessInstanceById(loginUser, projectCode, processInstanceId);

diff --git a/...-case/src/test/java/org/apache/dolphinscheduler/api/test/pages/workflow/ExecutorPage.java b/...-case/src/test/java/org/apache/dolphinscheduler/api/test/pages/workflow/ExecutorPage.java
@@ -40,8 +40,11 @@ public class ExecutorPage {
 
     private String sessionId;
 
-    public HttpResponse startProcessInstance(User loginUser, long projectCode, long processDefinitionCode,
-                                             String scheduleTime, FailureStrategy failureStrategy,
+    public HttpResponse startProcessInstance(User loginUser,
+                                             long projectCode,
+                                             long processDefinitionCode,
+                                             String scheduleTime,
+                                             FailureStrategy failureStrategy,
                                              WarningType warningType) {
         Map<String, Object> params = new HashMap<>();
         params.put("loginUser", loginUser);
@@ -82,18 +85,6 @@ public HttpResponse execute(User loginUser, long projectCode, int processInstanc
         return requestClient.post(url, headers, params);
     }
 
-    public HttpResponse startCheckProcessDefinition(User loginUser, long projectCode, long processDefinitionCode) {
-        Map<String, Object> params = new HashMap<>();
-        params.put("loginUser", loginUser);
-        params.put("processDefinitionCode", processDefinitionCode);
-        Map<String, String> headers = new HashMap<>();
-        headers.put(Constants.SESSION_ID_KEY, sessionId);
-
-        RequestClient requestClient = new RequestClient();
-        String url = String.format("/projects/%s/executors/start-check", projectCode);
-        return requestClient.post(url, headers, params);
-    }
-
     public HttpResponse executeTask(User loginUser, long projectCode, int processInstanceId, String startNodeList,
                                     TaskDependType taskDependType) {
         Map<String, Object> params = new HashMap<>();