Fix task cannot use workflow's environment (#16199)

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/dto/task/TaskCreateRequest.java

@@ -19,11 +19,11 @@
 
 import static org.apache.dolphinscheduler.common.constants.Constants.VERSION_FIRST;
 
-import org.apache.dolphinscheduler.common.constants.Constants;
 import org.apache.dolphinscheduler.common.enums.Flag;
 import org.apache.dolphinscheduler.common.enums.Priority;
 import org.apache.dolphinscheduler.common.enums.TimeoutFlag;
 import org.apache.dolphinscheduler.dao.entity.TaskDefinition;
+import org.apache.dolphinscheduler.dao.utils.WorkerGroupUtils;
 import org.apache.dolphinscheduler.plugin.task.api.enums.TaskTimeoutStrategy;
 
 import java.util.Date;
@@ -107,7 +107,7 @@ public TaskDefinition convert2TaskDefinition() {
         taskDefinition.setProjectCode(this.projectCode);
         taskDefinition.setTaskType(this.taskType);
         taskDefinition.setTaskParams(this.taskParams);
-        taskDefinition.setWorkerGroup(this.workerGroup == null ? Constants.DEFAULT_WORKER_GROUP : this.workerGroup);
+        taskDefinition.setWorkerGroup(WorkerGroupUtils.getWorkerGroupOrDefault(workerGroup));
         taskDefinition.setEnvironmentCode(this.environmentCode);
         taskDefinition.setFailRetryTimes(this.failRetryTimes);
         taskDefinition.setFailRetryInterval(this.failRetryInterval);

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/WorkerGroupService.java

@@ -18,7 +18,6 @@
 package org.apache.dolphinscheduler.api.service;
 
 import org.apache.dolphinscheduler.api.utils.Result;
-import org.apache.dolphinscheduler.dao.entity.TaskInstance;
 import org.apache.dolphinscheduler.dao.entity.User;
 
 import java.util.List;
@@ -77,13 +76,6 @@ Map<String, Object> saveWorkerGroup(User loginUser, int id, String name, String
      */
     Map<String, Object> getWorkerAddressList();
 
-    /**
-     * Get task instance's worker group
-     * @param taskInstance task instance
-     * @return worker group
-     */
-    String getTaskWorkerGroup(TaskInstance taskInstance);
-
     /**
      * Query worker group by process definition codes
      * @param processDefinitionCodeList processDefinitionCodeList

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ProcessDefinitionServiceImpl.java

@@ -34,7 +34,6 @@
 import static org.apache.dolphinscheduler.common.constants.CommandKeyConstants.CMD_PARAM_SUB_PROCESS_DEFINE_CODE;
 import static org.apache.dolphinscheduler.common.constants.Constants.COPY_SUFFIX;
 import static org.apache.dolphinscheduler.common.constants.Constants.DATA_LIST;
-import static org.apache.dolphinscheduler.common.constants.Constants.DEFAULT_WORKER_GROUP;
 import static org.apache.dolphinscheduler.common.constants.Constants.GLOBAL_PARAMS;
 import static org.apache.dolphinscheduler.common.constants.Constants.IMPORT_SUFFIX;
 import static org.apache.dolphinscheduler.common.constants.Constants.LOCAL_PARAMS;
@@ -109,6 +108,7 @@
 import org.apache.dolphinscheduler.dao.repository.ProcessDefinitionDao;
 import org.apache.dolphinscheduler.dao.repository.ProcessDefinitionLogDao;
 import org.apache.dolphinscheduler.dao.repository.TaskDefinitionLogDao;
+import org.apache.dolphinscheduler.dao.utils.WorkerGroupUtils;
 import org.apache.dolphinscheduler.plugin.task.api.TaskPluginManager;
 import org.apache.dolphinscheduler.plugin.task.api.enums.SqlType;
 import org.apache.dolphinscheduler.plugin.task.api.enums.TaskTimeoutStrategy;
@@ -1390,7 +1390,7 @@ private TaskDefinitionLog buildNormalSqlTaskDefinition(String taskName, DataSour
         taskDefinition.setFailRetryTimes(0);
         taskDefinition.setFailRetryInterval(0);
         taskDefinition.setTimeoutFlag(TimeoutFlag.CLOSE);
-        taskDefinition.setWorkerGroup(DEFAULT_WORKER_GROUP);
+        taskDefinition.setWorkerGroup(WorkerGroupUtils.getDefaultWorkerGroup());
         taskDefinition.setTaskPriority(Priority.MEDIUM);
         taskDefinition.setEnvironmentCode(-1);
         taskDefinition.setTimeout(0);

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ProjectWorkerGroupRelationServiceImpl.java

@@ -32,6 +32,7 @@
 import org.apache.dolphinscheduler.dao.mapper.ScheduleMapper;
 import org.apache.dolphinscheduler.dao.mapper.TaskDefinitionMapper;
 import org.apache.dolphinscheduler.dao.mapper.WorkerGroupMapper;
+import org.apache.dolphinscheduler.dao.utils.WorkerGroupUtils;
 
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.collections4.SetUtils;
@@ -118,7 +119,7 @@ public Result assignWorkerGroupsToProject(User loginUser, Long projectCode, List
                 workerGroupMapper.queryAllWorkerGroup().stream().map(WorkerGroup::getName).collect(
                         Collectors.toSet());
 
-        workerGroupNames.add(Constants.DEFAULT_WORKER_GROUP);
+        workerGroupNames.add(WorkerGroupUtils.getDefaultWorkerGroup());
 
         Set<String> assignedWorkerGroupNames = new HashSet<>(workerGroups);
 

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/WorkerGroupServiceImpl.java

@@ -32,7 +32,6 @@
 import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
 import org.apache.dolphinscheduler.dao.entity.Schedule;
 import org.apache.dolphinscheduler.dao.entity.TaskDefinition;
-import org.apache.dolphinscheduler.dao.entity.TaskInstance;
 import org.apache.dolphinscheduler.dao.entity.User;
 import org.apache.dolphinscheduler.dao.entity.WorkerGroup;
 import org.apache.dolphinscheduler.dao.mapper.EnvironmentWorkerGroupRelationMapper;
@@ -41,6 +40,7 @@
 import org.apache.dolphinscheduler.dao.mapper.ScheduleMapper;
 import org.apache.dolphinscheduler.dao.mapper.TaskDefinitionMapper;
 import org.apache.dolphinscheduler.dao.mapper.WorkerGroupMapper;
+import org.apache.dolphinscheduler.dao.utils.WorkerGroupUtils;
 import org.apache.dolphinscheduler.registry.api.RegistryClient;
 import org.apache.dolphinscheduler.registry.api.enums.RegistryNodeType;
 import org.apache.dolphinscheduler.service.process.ProcessService;
@@ -357,12 +357,12 @@ private List<WorkerGroup> getWorkerGroups(List<Integer> ids) {
             workerGroups = workerGroupMapper.queryAllWorkerGroup();
         }
         boolean containDefaultWorkerGroups = workerGroups.stream()
-                .anyMatch(workerGroup -> Constants.DEFAULT_WORKER_GROUP.equals(workerGroup.getName()));
+                .anyMatch(workerGroup -> WorkerGroupUtils.isWorkerGroupEmpty(workerGroup.getName()));
         if (!containDefaultWorkerGroups) {
             // there doesn't exist a default WorkerGroup, we will add all worker to the default worker group.
             Set<String> activeWorkerNodes = registryClient.getServerNodeSet(RegistryNodeType.WORKER);
             WorkerGroup defaultWorkerGroup = new WorkerGroup();
-            defaultWorkerGroup.setName(Constants.DEFAULT_WORKER_GROUP);
+            defaultWorkerGroup.setName(WorkerGroupUtils.getDefaultWorkerGroup());
             defaultWorkerGroup.setAddrList(String.join(Constants.COMMA, activeWorkerNodes));
             defaultWorkerGroup.setCreateTime(new Date());
             defaultWorkerGroup.setUpdateTime(new Date());
@@ -431,27 +431,6 @@ public Map<String, Object> getWorkerAddressList() {
         return result;
     }
 
-    @Override
-    public String getTaskWorkerGroup(TaskInstance taskInstance) {
-        if (taskInstance == null) {
-            return null;
-        }
-
-        String workerGroup = taskInstance.getWorkerGroup();
-
-        if (StringUtils.isNotEmpty(workerGroup)) {
-            return workerGroup;
-        }
-        int processInstanceId = taskInstance.getProcessInstanceId();
-        ProcessInstance processInstance = processService.findProcessInstanceById(processInstanceId);
-
-        if (processInstance != null) {
-            return processInstance.getWorkerGroup();
-        }
-        log.info("task : {} will use default worker group", taskInstance.getId());
-        return Constants.DEFAULT_WORKER_GROUP;
-    }
-
     @Override
     public Map<Long, String> queryWorkerGroupByProcessDefinitionCodes(List<Long> processDefinitionCodeList) {
         List<Schedule> processDefinitionScheduleList =

dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/ExecuteFunctionServiceTest.java

@@ -67,6 +67,7 @@
 import org.apache.dolphinscheduler.dao.mapper.TaskGroupQueueMapper;
 import org.apache.dolphinscheduler.dao.mapper.TenantMapper;
 import org.apache.dolphinscheduler.dao.repository.ProcessInstanceDao;
+import org.apache.dolphinscheduler.dao.utils.WorkerGroupUtils;
 import org.apache.dolphinscheduler.registry.api.enums.RegistryNodeType;
 import org.apache.dolphinscheduler.service.command.CommandService;
 import org.apache.dolphinscheduler.service.process.ProcessService;
@@ -271,7 +272,7 @@ public void testNoComplement() {
                 null, null,
                 null, null, null,
                 RunMode.RUN_MODE_SERIAL,
-                Priority.LOW, Constants.DEFAULT_WORKER_GROUP, tenantCode, 100L, 10, null, null,
+                Priority.LOW, WorkerGroupUtils.getDefaultWorkerGroup(), tenantCode, 100L, 10, null, null,
                 Constants.DRY_RUN_FLAG_NO,
                 Constants.TEST_FLAG_NO,
                 ComplementDependentMode.OFF_MODE, null,
@@ -298,7 +299,7 @@ public void testComplementWithStartNodeList() {
                 null, "123456789,987654321",
                 null, null, null,
                 RunMode.RUN_MODE_SERIAL,
-                Priority.LOW, Constants.DEFAULT_WORKER_GROUP, tenantCode, 100L, 110, null, null,
+                Priority.LOW, WorkerGroupUtils.getDefaultWorkerGroup(), tenantCode, 100L, 110, null, null,
                 Constants.DRY_RUN_FLAG_NO,
                 Constants.TEST_FLAG_NO,
                 ComplementDependentMode.OFF_MODE, null,
@@ -323,7 +324,7 @@ public void testComplementWithOldStartNodeList() {
                     null, "1123456789,987654321",
                     null, null, null,
                     RunMode.RUN_MODE_SERIAL,
-                    Priority.LOW, Constants.DEFAULT_WORKER_GROUP, tenantCode, 100L, 110, null, 0,
+                    Priority.LOW, WorkerGroupUtils.getDefaultWorkerGroup(), tenantCode, 100L, 110, null, 0,
                     Constants.DRY_RUN_FLAG_NO,
                     Constants.TEST_FLAG_NO,
                     ComplementDependentMode.OFF_MODE, null,
@@ -354,14 +355,14 @@ public void testComplementWithDependentMode() {
         dependentProcessDefinition.setProcessDefinitionCode(2);
         dependentProcessDefinition.setProcessDefinitionVersion(1);
         dependentProcessDefinition.setTaskDefinitionCode(1);
-        dependentProcessDefinition.setWorkerGroup(Constants.DEFAULT_WORKER_GROUP);
+        dependentProcessDefinition.setWorkerGroup(WorkerGroupUtils.getDefaultWorkerGroup());
         dependentProcessDefinition.setTaskParams(
                 "{\"localParams\":[],\"resourceList\":[],\"dependence\":{\"relation\":\"AND\",\"dependTaskList\":[{\"relation\":\"AND\",\"dependItemList\":[{\"depTaskCode\":2,\"status\":\"SUCCESS\"}]}]},\"conditionResult\":{\"successNode\":[1],\"failedNode\":[1]}}");
         Mockito.when(processService.queryDependentProcessDefinitionByProcessDefinitionCode(processDefinitionCode))
                 .thenReturn(Lists.newArrayList(dependentProcessDefinition));
 
         Map<Long, String> processDefinitionWorkerGroupMap = new HashMap<>();
-        processDefinitionWorkerGroupMap.put(1L, Constants.DEFAULT_WORKER_GROUP);
+        processDefinitionWorkerGroupMap.put(1L, WorkerGroupUtils.getDefaultWorkerGroup());
         Mockito.when(workerGroupService.queryWorkerGroupByProcessDefinitionCodes(Lists.newArrayList(1L)))
                 .thenReturn(processDefinitionWorkerGroupMap);
 
@@ -370,7 +371,7 @@ public void testComplementWithDependentMode() {
         command.setCommandType(CommandType.COMPLEMENT_DATA);
         command.setCommandParam(
                 "{\"StartNodeList\":\"1\",\"complementStartDate\":\"2020-01-01 00:00:00\",\"complementEndDate\":\"2020-01-31 23:00:00\"}");
-        command.setWorkerGroup(Constants.DEFAULT_WORKER_GROUP);
+        command.setWorkerGroup(WorkerGroupUtils.getDefaultWorkerGroup());
         command.setProcessDefinitionCode(processDefinitionCode);
         command.setExecutorId(1);
 
@@ -383,7 +384,7 @@ public void testComplementWithDependentMode() {
         childDependent.setProcessDefinitionCode(3);
         childDependent.setProcessDefinitionVersion(1);
         childDependent.setTaskDefinitionCode(4);
-        childDependent.setWorkerGroup(Constants.DEFAULT_WORKER_GROUP);
+        childDependent.setWorkerGroup(WorkerGroupUtils.getDefaultWorkerGroup());
         childDependent.setTaskParams(
                 "{\"localParams\":[],\"resourceList\":[],\"dependence\":{\"relation\":\"AND\",\"dependTaskList\":[{\"relation\":\"AND\",\"dependItemList\":[{\"depTaskCode\":3,\"status\":\"SUCCESS\"}]}]},\"conditionResult\":{\"successNode\":[1],\"failedNode\":[1]}}");
         Mockito.when(processService.queryDependentProcessDefinitionByProcessDefinitionCode(
@@ -409,7 +410,8 @@ public void testDateError() {
                 null, null,
                 null, null, null,
                 RunMode.RUN_MODE_SERIAL,
-                Priority.LOW, Constants.DEFAULT_WORKER_GROUP, tenantCode, 100L, 110, null, 2, Constants.DRY_RUN_FLAG_NO,
+                Priority.LOW, WorkerGroupUtils.getDefaultWorkerGroup(), tenantCode, 100L, 110, null, 2,
+                Constants.DRY_RUN_FLAG_NO,
                 Constants.TEST_FLAG_NO,
                 ComplementDependentMode.OFF_MODE, null,
                 false,
@@ -434,7 +436,7 @@ public void testSerial() {
                 null, null,
                 null, null, null,
                 RunMode.RUN_MODE_SERIAL,
-                Priority.LOW, Constants.DEFAULT_WORKER_GROUP, tenantCode, 100L, 110, null, null,
+                Priority.LOW, WorkerGroupUtils.getDefaultWorkerGroup(), tenantCode, 100L, 110, null, null,
                 Constants.DRY_RUN_FLAG_NO,
                 Constants.TEST_FLAG_NO,
                 ComplementDependentMode.OFF_MODE, null,
@@ -460,7 +462,8 @@ public void testParallelWithOutSchedule() {
                 null, null,
                 null, null, null,
                 RunMode.RUN_MODE_PARALLEL,
-                Priority.LOW, Constants.DEFAULT_WORKER_GROUP, tenantCode, 100L, 110, null, 2, Constants.DRY_RUN_FLAG_NO,
+                Priority.LOW, WorkerGroupUtils.getDefaultWorkerGroup(), tenantCode, 100L, 110, null, 2,
+                Constants.DRY_RUN_FLAG_NO,
                 Constants.TEST_FLAG_NO,
                 ComplementDependentMode.OFF_MODE, null,
                 false,
@@ -486,7 +489,7 @@ public void testParallelWithSchedule() {
                 null, null,
                 null, null, null,
                 RunMode.RUN_MODE_PARALLEL,
-                Priority.LOW, Constants.DEFAULT_WORKER_GROUP, tenantCode, 100L, 110, null, 15,
+                Priority.LOW, WorkerGroupUtils.getDefaultWorkerGroup(), tenantCode, 100L, 110, null, 15,
                 Constants.DRY_RUN_FLAG_NO,
                 Constants.TEST_FLAG_NO,
                 ComplementDependentMode.OFF_MODE, null,
@@ -514,7 +517,7 @@ public void testNoMasterServers() {
                 null,
                 RunMode.RUN_MODE_PARALLEL,
                 Priority.LOW,
-                Constants.DEFAULT_WORKER_GROUP,
+                WorkerGroupUtils.getDefaultWorkerGroup(),
                 tenantCode,
                 100L,
                 110,
@@ -553,7 +556,7 @@ public void testOfTestRun() {
                 null, null,
                 null, null, 0,
                 RunMode.RUN_MODE_PARALLEL,
-                Priority.LOW, Constants.DEFAULT_WORKER_GROUP, tenantCode, 100L, 110, null, 15,
+                Priority.LOW, WorkerGroupUtils.getDefaultWorkerGroup(), tenantCode, 100L, 110, null, 15,
                 Constants.DRY_RUN_FLAG_NO,
                 Constants.TEST_FLAG_YES,
                 ComplementDependentMode.OFF_MODE, null,

dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/ProcessDefinitionServiceTest.java

@@ -72,6 +72,7 @@
 import org.apache.dolphinscheduler.dao.repository.ProcessDefinitionDao;
 import org.apache.dolphinscheduler.dao.repository.ProcessDefinitionLogDao;
 import org.apache.dolphinscheduler.dao.repository.TaskDefinitionLogDao;
+import org.apache.dolphinscheduler.dao.utils.WorkerGroupUtils;
 import org.apache.dolphinscheduler.service.alert.ListenerEventAlertManager;
 import org.apache.dolphinscheduler.service.process.ProcessService;
 import org.apache.dolphinscheduler.spi.enums.DbType;
@@ -1143,7 +1144,7 @@ private Schedule getSchedule() {
         schedule.setProcessInstancePriority(Priority.MEDIUM);
         schedule.setWarningType(WarningType.NONE);
         schedule.setWarningGroupId(1);
-        schedule.setWorkerGroup(Constants.DEFAULT_WORKER_GROUP);
+        schedule.setWorkerGroup(WorkerGroupUtils.getDefaultWorkerGroup());
         return schedule;
     }
 

dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/WorkerGroupServiceTest.java

@@ -30,7 +30,6 @@
 import org.apache.dolphinscheduler.common.enums.AuthorizationType;
 import org.apache.dolphinscheduler.common.enums.UserType;
 import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
-import org.apache.dolphinscheduler.dao.entity.TaskInstance;
 import org.apache.dolphinscheduler.dao.entity.User;
 import org.apache.dolphinscheduler.dao.entity.WorkerGroup;
 import org.apache.dolphinscheduler.dao.mapper.EnvironmentWorkerGroupRelationMapper;
@@ -65,8 +64,6 @@
 @MockitoSettings(strictness = Strictness.LENIENT)
 public class WorkerGroupServiceTest {
 
-    private static final Logger logger = LoggerFactory.getLogger(WorkerGroupServiceTest.class);
-
     private static final Logger baseServiceLogger = LoggerFactory.getLogger(BaseServiceImpl.class);
 
     private static final Logger serviceLogger = LoggerFactory.getLogger(WorkerGroupService.class);
@@ -288,47 +285,6 @@ public void testQueryAllGroupWithDefault() {
         Assertions.assertEquals("default", workerGroups.toArray()[0]);
     }
 
-    @Test
-    public void giveNull_whenGetTaskWorkerGroup_expectNull() {
-        String nullWorkerGroup = workerGroupService.getTaskWorkerGroup(null);
-        Assertions.assertNull(nullWorkerGroup);
-    }
-
-    @Test
-    public void giveCorrectTaskInstance_whenGetTaskWorkerGroup_expectTaskWorkerGroup() {
-        TaskInstance taskInstance = new TaskInstance();
-        taskInstance.setId(1);
-        taskInstance.setWorkerGroup("cluster1");
-
-        String workerGroup = workerGroupService.getTaskWorkerGroup(taskInstance);
-        Assertions.assertEquals("cluster1", workerGroup);
-    }
-
-    @Test
-    public void giveNullWorkerGroup_whenGetTaskWorkerGroup_expectProcessWorkerGroup() {
-        TaskInstance taskInstance = new TaskInstance();
-        taskInstance.setId(1);
-        taskInstance.setProcessInstanceId(1);
-        ProcessInstance processInstance = new ProcessInstance();
-        processInstance.setId(1);
-        processInstance.setWorkerGroup("cluster1");
-        Mockito.when(processService.findProcessInstanceById(1)).thenReturn(processInstance);
-
-        String workerGroup = workerGroupService.getTaskWorkerGroup(taskInstance);
-        Assertions.assertEquals("cluster1", workerGroup);
-    }
-
-    @Test
-    public void giveNullTaskAndProcessWorkerGroup_whenGetTaskWorkerGroup_expectDefault() {
-        TaskInstance taskInstance = new TaskInstance();
-        taskInstance.setId(1);
-        taskInstance.setProcessInstanceId(1);
-        Mockito.when(processService.findProcessInstanceById(1)).thenReturn(null);
-
-        String defaultWorkerGroup = workerGroupService.getTaskWorkerGroup(taskInstance);
-        Assertions.assertEquals(Constants.DEFAULT_WORKER_GROUP, defaultWorkerGroup);
-    }
-
     /**
      * get Group
      */

dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/constants/Constants.java

@@ -47,6 +47,7 @@ private Constants() {
     public static final String RESOURCE_TYPE_UDF = "udfs";
 
     public static final String EMPTY_STRING = "";
+    public static final int MAX_FILE_SIZE = 1024 * 1024 * 1024;
 
     /**
      * resource.hdfs.fs.defaultFS
@@ -532,14 +533,9 @@ private Constants() {
      * session timeout
      */
     public static final int SESSION_TIME_OUT = 7200;
-    public static final int MAX_FILE_SIZE = 1024 * 1024 * 1024;
     public static final String UDF = "UDF";
     public static final String CLASS = "class";
 
-    /**
-     * default worker group
-     */
-    public static final String DEFAULT_WORKER_GROUP = "default";
     /**
      * authorize writable perm
      */