[DSIP-46] Unify the logic task params data structure between front-en…

…d and back-end
apache · Jun 22, 2024 · 0639aa9 · 0639aa9
1 parent f0e0395
commit 0639aa9
Show file tree

Hide file tree

Showing 139 changed files with 1,226 additions and 2,495 deletions.
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/ApiApplicationServer.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/ApiApplicationServer.java
@@ -19,22 +19,14 @@
 
 import org.apache.dolphinscheduler.api.metrics.ApiServerMetrics;
 import org.apache.dolphinscheduler.common.CommonConfiguration;
-import org.apache.dolphinscheduler.common.enums.PluginType;
 import org.apache.dolphinscheduler.common.thread.DefaultUncaughtExceptionHandler;
 import org.apache.dolphinscheduler.dao.DaoConfiguration;
 import org.apache.dolphinscheduler.dao.PluginDao;
-import org.apache.dolphinscheduler.dao.entity.PluginDefine;
 import org.apache.dolphinscheduler.plugin.datasource.api.plugin.DataSourceProcessorProvider;
 import org.apache.dolphinscheduler.plugin.storage.api.StorageConfiguration;
-import org.apache.dolphinscheduler.plugin.task.api.TaskChannelFactory;
 import org.apache.dolphinscheduler.plugin.task.api.TaskPluginManager;
 import org.apache.dolphinscheduler.registry.api.RegistryConfiguration;
 import org.apache.dolphinscheduler.service.ServiceConfiguration;
-import org.apache.dolphinscheduler.spi.params.PluginParamsTransfer;
-import org.apache.dolphinscheduler.spi.params.base.PluginParams;
-
-import java.util.List;
-import java.util.Map;
 
 import lombok.extern.slf4j.Slf4j;
 
@@ -68,17 +60,7 @@ public static void main(String[] args) {
     @EventListener
     public void run(ApplicationReadyEvent readyEvent) {
         log.info("Received spring application context ready event will load taskPlugin and write to DB");
-        // install task plugin
-        TaskPluginManager.loadPlugin();
         DataSourceProcessorProvider.initialize();
-        for (Map.Entry<String, TaskChannelFactory> entry : TaskPluginManager.getTaskChannelFactoryMap().entrySet()) {
-            String taskPluginName = entry.getKey();
-            TaskChannelFactory taskChannelFactory = entry.getValue();
-            List<PluginParams> params = taskChannelFactory.getParams();
-            String paramsJson = PluginParamsTransfer.transferParamsToJson(params);
-
-            PluginDefine pluginDefine = new PluginDefine(taskPluginName, PluginType.TASK.getDesc(), paramsJson);
-            pluginDao.addOrUpdatePluginDefine(pluginDefine);
-        }
+        TaskPluginManager.loadPlugin();
     }
 }
diff --git a/...r-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ExecutorServiceImpl.java b/...r-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ExecutorServiceImpl.java
@@ -89,8 +89,8 @@
 import org.apache.dolphinscheduler.extract.master.transportor.StreamingTaskTriggerRequest;
 import org.apache.dolphinscheduler.extract.master.transportor.StreamingTaskTriggerResponse;
 import org.apache.dolphinscheduler.extract.master.transportor.WorkflowInstanceStateChangeEvent;
-import org.apache.dolphinscheduler.plugin.task.api.TaskConstants;
 import org.apache.dolphinscheduler.plugin.task.api.model.Property;
+import org.apache.dolphinscheduler.plugin.task.api.utils.TaskTypeUtils;
 import org.apache.dolphinscheduler.registry.api.enums.RegistryNodeType;
 import org.apache.dolphinscheduler.service.command.CommandService;
 import org.apache.dolphinscheduler.service.cron.CronUtils;
@@ -361,7 +361,7 @@ public boolean checkSubProcessDefinitionValid(ProcessDefinition processDefinitio
         // find out the process definition code
         Set<Long> processDefinitionCodeSet = new HashSet<>();
         taskDefinitions.stream()
-                .filter(task -> TaskConstants.TASK_TYPE_SUB_PROCESS.equalsIgnoreCase(task.getTaskType())).forEach(
+                .filter(task -> TaskTypeUtils.isSubWorkflowTask(task.getTaskType())).forEach(
                         taskDefinition -> processDefinitionCodeSet.add(Long.valueOf(
                                 JSONUtils.getNodeString(taskDefinition.getTaskParams(),
                                         CMD_PARAM_SUB_PROCESS_DEFINE_CODE))));

diff --git a/.../main/java/org/apache/dolphinscheduler/api/service/impl/ProcessDefinitionServiceImpl.java b/.../main/java/org/apache/dolphinscheduler/api/service/impl/ProcessDefinitionServiceImpl.java
@@ -40,7 +40,6 @@
 import static org.apache.dolphinscheduler.common.constants.Constants.LOCAL_PARAMS;
 import static org.apache.dolphinscheduler.plugin.task.api.TaskConstants.LOCAL_PARAMS_LIST;
 import static org.apache.dolphinscheduler.plugin.task.api.TaskConstants.TASK_TYPE;
-import static org.apache.dolphinscheduler.plugin.task.api.TaskConstants.TASK_TYPE_SQL;
 
 import org.apache.dolphinscheduler.api.dto.DagDataSchedule;
 import org.apache.dolphinscheduler.api.dto.treeview.Instance;
@@ -115,6 +114,8 @@
 import org.apache.dolphinscheduler.plugin.task.api.model.Property;
 import org.apache.dolphinscheduler.plugin.task.api.parameters.ParametersNode;
 import org.apache.dolphinscheduler.plugin.task.api.parameters.SqlParameters;
+import org.apache.dolphinscheduler.plugin.task.api.utils.TaskTypeUtils;
+import org.apache.dolphinscheduler.plugin.task.sql.SqlTaskChannelFactory;
 import org.apache.dolphinscheduler.service.alert.ListenerEventAlertManager;
 import org.apache.dolphinscheduler.service.model.TaskNode;
 import org.apache.dolphinscheduler.service.process.ProcessService;
@@ -424,7 +425,6 @@ private List<TaskDefinitionLog> generateTaskDefinitionList(String taskDefinition
                 if (!TaskPluginManager.checkTaskParameters(ParametersNode.builder()
                         .taskType(taskDefinitionLog.getTaskType())
                         .taskParams(taskDefinitionLog.getTaskParams())
-                        .dependence(taskDefinitionLog.getDependence())
                         .build())) {
                     log.error(
                             "Generate task definition list failed, the given task definition parameter is invalided, taskName: {}, taskDefinition: {}",
@@ -1386,7 +1386,7 @@ private TaskDefinitionLog buildNormalSqlTaskDefinition(String taskName, DataSour
         sqlParameters.setLocalParams(Collections.emptyList());
         taskDefinition.setTaskParams(JSONUtils.toJsonString(sqlParameters));
         taskDefinition.setCode(CodeGenerateUtils.genCode());
-        taskDefinition.setTaskType(TASK_TYPE_SQL);
+        taskDefinition.setTaskType(SqlTaskChannelFactory.NAME);
         taskDefinition.setFailRetryTimes(0);
         taskDefinition.setFailRetryInterval(0);
         taskDefinition.setTimeoutFlag(TimeoutFlag.CLOSE);
@@ -1615,12 +1615,11 @@ public Map<String, Object> checkProcessNodeList(String processTaskRelationJson,
 
             // check whether the process definition json is normal
             for (TaskNode taskNode : taskNodes) {
-                if (!TaskPluginManager.checkTaskParameters(ParametersNode.builder()
-                        .taskType(taskNode.getType())
-                        .taskParams(taskNode.getTaskParams())
-                        .dependence(taskNode.getDependence())
-                        .switchResult(taskNode.getSwitchResult())
-                        .build())) {
+                if (!TaskPluginManager.checkTaskParameters(
+                        ParametersNode.builder()
+                                .taskType(taskNode.getType())
+                                .taskParams(taskNode.getParams())
+                                .build())) {
                     log.error("Task node {} parameter invalid.", taskNode.getName());
                     putMsg(result, Status.PROCESS_NODE_S_PARAMETER_INVALID, taskNode.getName());
                     return result;
@@ -1891,7 +1890,7 @@ public Map<String, Object> viewTree(User loginUser, long projectCode, long code,
 
                         long subProcessCode = 0L;
                         // if process is sub process, the return sub id, or sub id=0
-                        if (taskInstance.isSubProcess()) {
+                        if (TaskTypeUtils.isSubWorkflowTask(taskInstance.getTaskType())) {
                             TaskDefinition taskDefinition = taskDefinitionMap.get(taskInstance.getTaskCode());
                             subProcessCode = Long.parseLong(JSONUtils.parseObject(
                                     taskDefinition.getTaskParams()).path(CMD_PARAM_SUB_PROCESS_DEFINE_CODE).asText());

diff --git a/...rc/main/java/org/apache/dolphinscheduler/api/service/impl/ProcessInstanceServiceImpl.java b/...rc/main/java/org/apache/dolphinscheduler/api/service/impl/ProcessInstanceServiceImpl.java
@@ -26,8 +26,6 @@
 import static org.apache.dolphinscheduler.common.constants.Constants.LOCAL_PARAMS;
 import static org.apache.dolphinscheduler.common.constants.Constants.PROCESS_INSTANCE_STATE;
 import static org.apache.dolphinscheduler.common.constants.Constants.TASK_LIST;
-import static org.apache.dolphinscheduler.plugin.task.api.TaskConstants.TASK_TYPE_DEPENDENT;
-import static org.apache.dolphinscheduler.plugin.task.api.TaskConstants.TASK_TYPE_SUB_PROCESS;
 
 import org.apache.dolphinscheduler.api.constants.ApiFuncIdentificationConstant;
 import org.apache.dolphinscheduler.api.dto.DynamicSubWorkflowDto;
@@ -82,6 +80,7 @@
 import org.apache.dolphinscheduler.plugin.task.api.model.Property;
 import org.apache.dolphinscheduler.plugin.task.api.parameters.ParametersNode;
 import org.apache.dolphinscheduler.plugin.task.api.utils.ParameterUtils;
+import org.apache.dolphinscheduler.plugin.task.api.utils.TaskTypeUtils;
 import org.apache.dolphinscheduler.service.expand.CuringParamsService;
 import org.apache.dolphinscheduler.service.model.TaskNode;
 import org.apache.dolphinscheduler.service.process.ProcessService;
@@ -498,7 +497,7 @@ public List<DynamicSubWorkflowDto> queryDynamicSubWorkflowInstances(User loginUs
             throw new ServiceException(Status.TASK_INSTANCE_NOT_EXISTS, taskId);
         }
 
-        if (!taskInstance.isDynamic()) {
+        if (!TaskTypeUtils.isDynamicTask(taskInstance.getTaskType())) {
             putMsg(result, Status.TASK_INSTANCE_NOT_DYNAMIC_TASK, taskInstance.getName());
             throw new ServiceException(Status.TASK_INSTANCE_NOT_EXISTS, taskId);
         }
@@ -548,7 +547,7 @@ public List<DynamicSubWorkflowDto> queryDynamicSubWorkflowInstances(User loginUs
      */
     private void addDependResultForTaskList(User loginUser, List<TaskInstance> taskInstanceList) throws IOException {
         for (TaskInstance taskInstance : taskInstanceList) {
-            if (TASK_TYPE_DEPENDENT.equalsIgnoreCase(taskInstance.getTaskType())) {
+            if (TaskTypeUtils.isDependentTask(taskInstance.getTaskType())) {
                 log.info("DEPENDENT type task instance need to set dependent result, taskCode:{}, taskInstanceId:{}",
                         taskInstance.getTaskCode(), taskInstance.getId());
                 // TODO The result of dependent item should not be obtained from the log, waiting for optimization.
@@ -628,9 +627,7 @@ public Map<String, Object> querySubProcessInstanceByTaskId(User loginUser, long
             return result;
         }
 
-        if (!taskInstance.isSubProcess()) {
-            log.warn("Task instance is not {} type instance, projectCode:{}, taskInstanceId:{}.",
-                    TASK_TYPE_SUB_PROCESS, projectCode, taskId);
+        if (!TaskTypeUtils.isSubWorkflowTask(taskInstance.getTaskType())) {
             putMsg(result, Status.TASK_INSTANCE_NOT_SUB_WORKFLOW_INSTANCE, taskInstance.getName());
             return result;
         }
@@ -717,7 +714,6 @@ public Map<String, Object> updateProcessInstance(User loginUser, long projectCod
             if (!TaskPluginManager.checkTaskParameters(ParametersNode.builder()
                     .taskType(taskDefinitionLog.getTaskType())
                     .taskParams(taskDefinitionLog.getTaskParams())
-                    .dependence(taskDefinitionLog.getDependence())
                     .build())) {
                 log.error("Task parameters are invalid,  taskDefinitionName:{}.", taskDefinitionLog.getName());
                 putMsg(result, Status.PROCESS_NODE_S_PARAMETER_INVALID, taskDefinitionLog.getName());

diff --git a/...ain/java/org/apache/dolphinscheduler/api/service/impl/ProcessTaskRelationServiceImpl.java b/...ain/java/org/apache/dolphinscheduler/api/service/impl/ProcessTaskRelationServiceImpl.java
@@ -18,9 +18,6 @@
 package org.apache.dolphinscheduler.api.service.impl;
 
 import static java.util.stream.Collectors.toSet;
-import static org.apache.dolphinscheduler.plugin.task.api.TaskConstants.TASK_TYPE_CONDITIONS;
-import static org.apache.dolphinscheduler.plugin.task.api.TaskConstants.TASK_TYPE_DEPENDENT;
-import static org.apache.dolphinscheduler.plugin.task.api.TaskConstants.TASK_TYPE_SUB_PROCESS;
 
 import org.apache.dolphinscheduler.api.dto.taskRelation.TaskRelationCreateRequest;
 import org.apache.dolphinscheduler.api.dto.taskRelation.TaskRelationFilterRequest;
@@ -46,6 +43,7 @@
 import org.apache.dolphinscheduler.dao.mapper.ProjectMapper;
 import org.apache.dolphinscheduler.dao.mapper.TaskDefinitionLogMapper;
 import org.apache.dolphinscheduler.dao.mapper.TaskDefinitionMapper;
+import org.apache.dolphinscheduler.plugin.task.api.utils.TaskTypeUtils;
 import org.apache.dolphinscheduler.service.process.ProcessService;
 
 import org.apache.commons.collections4.CollectionUtils;
@@ -354,9 +352,9 @@ public Map<String, Object> deleteTaskProcessRelation(User loginUser, long projec
         }
         updateProcessDefiniteVersion(loginUser, result, processDefinition);
         updateRelation(loginUser, result, processDefinition, processTaskRelationList);
-        if (TASK_TYPE_CONDITIONS.equals(taskDefinition.getTaskType())
-                || TASK_TYPE_DEPENDENT.equals(taskDefinition.getTaskType())
-                || TASK_TYPE_SUB_PROCESS.equals(taskDefinition.getTaskType())) {
+        if (TaskTypeUtils.isConditionTask(taskDefinition.getTaskType())
+                || TaskTypeUtils.isSubWorkflowTask(taskDefinition.getTaskType())
+                || TaskTypeUtils.isDependentTask(taskDefinition.getTaskType())) {
             int deleteTaskDefinition = taskDefinitionMapper.deleteByCode(taskCode);
             if (0 == deleteTaskDefinition) {
                 log.error("Delete task definition error, taskDefinitionCode:{}.", taskCode);

diff --git a/...src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskDefinitionServiceImpl.java b/...src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskDefinitionServiceImpl.java
@@ -170,7 +170,6 @@ public Map<String, Object> createTaskDefinition(User loginUser,
             if (!TaskPluginManager.checkTaskParameters(ParametersNode.builder()
                     .taskType(taskDefinitionLog.getTaskType())
                     .taskParams(taskDefinitionLog.getTaskParams())
-                    .dependence(taskDefinitionLog.getDependence())
                     .build())) {
                 log.warn("Task definition {} parameters are invalid.", taskDefinitionLog.getName());
                 putMsg(result, Status.PROCESS_NODE_S_PARAMETER_INVALID, taskDefinitionLog.getName());
@@ -211,7 +210,6 @@ private void checkTaskDefinitionValid(User user, TaskDefinition taskDefinition,
         if (!TaskPluginManager.checkTaskParameters(ParametersNode.builder()
                 .taskType(taskDefinition.getTaskType())
                 .taskParams(taskDefinition.getTaskParams())
-                .dependence(taskDefinition.getDependence())
                 .build())) {
             throw new ServiceException(Status.PROCESS_NODE_S_PARAMETER_INVALID, taskDefinition.getName());
         }
@@ -324,7 +322,6 @@ public Map<String, Object> createTaskBindsWorkFlow(User loginUser,
         if (!TaskPluginManager.checkTaskParameters(ParametersNode.builder()
                 .taskType(taskDefinition.getTaskType())
                 .taskParams(taskDefinition.getTaskParams())
-                .dependence(taskDefinition.getDependence())
                 .build())) {
             log.error("Task definition {} parameters are invalid", taskDefinition.getName());
             putMsg(result, Status.PROCESS_NODE_S_PARAMETER_INVALID, taskDefinition.getName());
@@ -735,7 +732,6 @@ private TaskDefinitionLog updateTask(User loginUser, long projectCode, long task
         if (!TaskPluginManager.checkTaskParameters(ParametersNode.builder()
                 .taskType(taskDefinitionToUpdate.getTaskType())
                 .taskParams(taskDefinitionToUpdate.getTaskParams())
-                .dependence(taskDefinitionToUpdate.getDependence())
                 .build())) {
             log.warn("Task definition parameters are invalid, taskDefinitionName:{}.",
                     taskDefinitionToUpdate.getName());

diff --git a/...rc/main/java/org/apache/dolphinscheduler/api/service/impl/WorkFlowLineageServiceImpl.java b/...rc/main/java/org/apache/dolphinscheduler/api/service/impl/WorkFlowLineageServiceImpl.java
@@ -17,8 +17,6 @@
 
 package org.apache.dolphinscheduler.api.service.impl;
 
-import static org.apache.dolphinscheduler.plugin.task.api.TaskConstants.TASK_TYPE_DEPENDENT;
-
 import org.apache.dolphinscheduler.api.enums.Status;
 import org.apache.dolphinscheduler.api.exceptions.ServiceException;
 import org.apache.dolphinscheduler.api.service.WorkFlowLineageService;
@@ -39,6 +37,7 @@
 import org.apache.dolphinscheduler.plugin.task.api.model.DependentItem;
 import org.apache.dolphinscheduler.plugin.task.api.model.DependentTaskModel;
 import org.apache.dolphinscheduler.plugin.task.api.parameters.DependentParameters;
+import org.apache.dolphinscheduler.plugin.task.api.utils.TaskTypeUtils;
 
 import org.apache.commons.lang3.StringUtils;
 
@@ -147,7 +146,7 @@ private void getProcessDefinitionCodeByDependentDefinitionList(List<DependentPro
                                                                    List<Long> processDefinitionCodes) {
         for (DependentProcessDefinition dependentProcessDefinition : dependentDefinitionList) {
             for (DependentTaskModel dependentTaskModel : dependentProcessDefinition.getDependentParameters()
-                    .getDependTaskList()) {
+                    .getDependence().getDependTaskList()) {
                 for (DependentItem dependentItem : dependentTaskModel.getDependItemList()) {
                     if (!processDefinitionCodes.contains(dependentItem.getDefinitionCode())) {
                         processDefinitionCodes.add(dependentItem.getDefinitionCode());
@@ -220,12 +219,12 @@ private Set<Long> querySourceWorkFlowCodes(long projectCode, long workFlowCode,
         List<TaskDefinitionLog> taskDefinitionLogs = taskDefinitionLogMapper.queryByTaskDefinitions(taskDefinitionList);
         for (TaskDefinitionLog taskDefinitionLog : taskDefinitionLogs) {
             if (taskDefinitionLog.getProjectCode() == projectCode) {
-                if (taskDefinitionLog.getTaskType().equals(TASK_TYPE_DEPENDENT)) {
+                if (TaskTypeUtils.isDependentTask(taskDefinitionLog.getTaskType())) {
                     DependentParameters dependentParameters =
                             JSONUtils.parseObject(taskDefinitionLog.getDependence(), DependentParameters.class);
                     if (dependentParameters != null) {
                         List<DependentTaskModel> dependTaskList =
-                                dependentParameters.getDependTaskList();
+                                dependentParameters.getDependence().getDependTaskList();
                         if (!CollectionUtils.isEmpty(dependTaskList)) {
                             for (DependentTaskModel taskModel : dependTaskList) {
                                 List<DependentItem> dependItemList = taskModel.getDependItemList();
@@ -247,9 +246,9 @@ private Set<Long> querySourceWorkFlowCodes(long projectCode, long workFlowCode,
     /**
      * Query and return tasks dependence with string format, is a wrapper of queryTaskDepOnTask and task query method.
      *
-     * @param projectCode Project code want to query tasks dependence
+     * @param projectCode           Project code want to query tasks dependence
      * @param processDefinitionCode Process definition code want to query tasks dependence
-     * @param taskCode Task code want to query tasks dependence
+     * @param taskCode              Task code want to query tasks dependence
      * @return Optional of formatter message
      */
     @Override
@@ -271,7 +270,7 @@ public Optional<String> taskDepOnTaskMsg(long projectCode, long processDefinitio
     /**
      * Query tasks depend on process definition, include upstream or downstream
      *
-     * @param projectCode Project code want to query tasks dependence
+     * @param projectCode           Project code want to query tasks dependence
      * @param processDefinitionCode Process definition code want to query tasks dependence
      * @return Set of TaskMainInfo
      */
@@ -291,7 +290,7 @@ public Set<TaskMainInfo> queryTaskDepOnProcess(long projectCode, long processDef
      * Query downstream tasks depend on a process definition or a task
      *
      * @param processDefinitionCode Process definition code want to query tasks dependence
-     * @param taskCode Task code want to query tasks dependence
+     * @param taskCode              Task code want to query tasks dependence
      * @return downstream dependent tasks
      */
     @Override