Merge pull request #12 from apache/dev

update

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/WorkerGroupController.java

@@ -52,35 +52,7 @@ public class WorkerGroupController extends BaseController {
     WorkerGroupService workerGroupService;
 
 
-    /**
-     * create or update a worker group
-     *
-     * @param loginUser login user
-     * @param id        worker group id
-     * @param name      worker group name
-     * @param ipList    ip list
-     * @return create or update result code
-     */
-    @ApiOperation(value = "saveWorkerGroup", notes = "CREATE_WORKER_GROUP_NOTES")
-    @ApiImplicitParams({
-            @ApiImplicitParam(name = "id", value = "WORKER_GROUP_ID", dataType = "Int", example = "10", defaultValue = "0"),
-            @ApiImplicitParam(name = "name", value = "WORKER_GROUP_NAME", required = true, dataType = "String"),
-            @ApiImplicitParam(name = "ipList", value = "WORKER_IP_LIST", required = true, dataType = "String")
-    })
-    @PostMapping(value = "/save")
-    @ResponseStatus(HttpStatus.OK)
-    @ApiException(SAVE_ERROR)
-    public Result saveWorkerGroup(@ApiIgnore @RequestAttribute(value = Constants.SESSION_USER) User loginUser,
-                                  @RequestParam(value = "id", required = false, defaultValue = "0") int id,
-                                  @RequestParam(value = "name") String name,
-                                  @RequestParam(value = "ipList") String ipList
-    ) {
-        logger.info("save worker group: login user {}, id:{}, name: {}, ipList: {} ",
-                loginUser.getUserName(), id, name, ipList);
 
-        Map<String, Object> result = workerGroupService.saveWorkerGroup(loginUser, id, name, ipList);
-        return returnDataList(result);
-    }
 
     /**
      * query worker groups paging
@@ -132,28 +104,5 @@ public class WorkerGroupController extends BaseController {
         return returnDataList(result);
     }
 
-    /**
-     * delete worker group by id
-     *
-     * @param loginUser login user
-     * @param id        group id
-     * @return delete result code
-     */
-    @ApiOperation(value = "deleteById", notes = "DELETE_WORKER_GROUP_BY_ID_NOTES")
-    @ApiImplicitParams({
-            @ApiImplicitParam(name = "id", value = "WORKER_GROUP_ID", required = true, dataType = "Int", example = "10"),
-
-    })
-    @GetMapping(value = "/delete-by-id")
-    @ResponseStatus(HttpStatus.OK)
-    @ApiException(DELETE_WORKER_GROUP_FAIL)
-    public Result deleteById(@ApiIgnore @RequestAttribute(value = Constants.SESSION_USER) User loginUser,
-                             @RequestParam("id") Integer id
-    ) {
-        logger.info("delete worker group: login user {}, id:{} ",
-                loginUser.getUserName(), id);
 
-        Map<String, Object> result = workerGroupService.deleteWorkerGroupById(id);
-        return returnDataList(result);
-    }
 }

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

@@ -96,9 +96,6 @@ public class ProcessDefinitionService extends BaseDAGService {
     @Autowired
     private ProcessService processService;
 
-    @Autowired
-    private WorkerGroupMapper workerGroupMapper;
-
     /**
      * create process definition
      *
@@ -310,14 +307,14 @@ public class ProcessDefinitionService extends BaseDAGService {
             putMsg(result, Status.PROCESS_DEFINE_NOT_EXIST, processId);
             return result;
         } else {
-           return createProcessDefinition(
-                   loginUser,
-                   projectName,
-                   processDefinition.getName()+"_copy_"+System.currentTimeMillis(),
-                   processDefinition.getProcessDefinitionJson(),
-                   processDefinition.getDescription(),
-                   processDefinition.getLocations(),
-                   processDefinition.getConnects());
+            return createProcessDefinition(
+                    loginUser,
+                    projectName,
+                    processDefinition.getName()+"_copy_"+System.currentTimeMillis(),
+                    processDefinition.getProcessDefinitionJson(),
+                    processDefinition.getDescription(),
+                    processDefinition.getLocations(),
+                    processDefinition.getConnects());
         }
     }
 
@@ -408,19 +405,19 @@ public class ProcessDefinitionService extends BaseDAGService {
     public Map<String, Object> verifyProcessDefinitionName(User loginUser, String projectName, String name) {
 
         Map<String, Object> result = new HashMap<>();
-            Project project = projectMapper.queryByName(projectName);
+        Project project = projectMapper.queryByName(projectName);
 
-            Map<String, Object> checkResult = projectService.checkProjectAndAuth(loginUser, project, projectName);
-            Status resultEnum = (Status) checkResult.get(Constants.STATUS);
-            if (resultEnum != Status.SUCCESS) {
-                return checkResult;
-            }
-            ProcessDefinition processDefinition = processDefineMapper.queryByDefineName(project.getId(), name);
-            if (processDefinition == null) {
-                putMsg(result, Status.SUCCESS);
-            } else {
-                putMsg(result, Status.PROCESS_INSTANCE_EXIST, name);
-            }
+        Map<String, Object> checkResult = projectService.checkProjectAndAuth(loginUser, project, projectName);
+        Status resultEnum = (Status) checkResult.get(Constants.STATUS);
+        if (resultEnum != Status.SUCCESS) {
+            return checkResult;
+        }
+        ProcessDefinition processDefinition = processDefineMapper.queryByDefineName(project.getId(), name);
+        if (processDefinition == null) {
+            putMsg(result, Status.SUCCESS);
+        } else {
+            putMsg(result, Status.PROCESS_INSTANCE_EXIST, name);
+        }
         return result;
     }
 

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

@@ -91,8 +91,7 @@ public class ProcessInstanceService extends BaseDAGService {
     @Autowired
     LoggerService loggerService;
 
-    @Autowired
-    WorkerGroupMapper workerGroupMapper;
+
 
     @Autowired
     UsersService usersService;

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

@@ -16,24 +16,24 @@
  */
 package org.apache.dolphinscheduler.api.service;
 
+import com.baomidou.mybatisplus.core.metadata.IPage;
+import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
 import org.apache.dolphinscheduler.api.enums.Status;
 import org.apache.dolphinscheduler.api.utils.PageInfo;
 import org.apache.dolphinscheduler.common.Constants;
 import org.apache.dolphinscheduler.common.utils.CollectionUtils;
+import org.apache.dolphinscheduler.common.utils.DateUtils;
 import org.apache.dolphinscheduler.common.utils.StringUtils;
-import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
+import org.apache.dolphinscheduler.dao.entity.AccessToken;
 import org.apache.dolphinscheduler.dao.entity.User;
 import org.apache.dolphinscheduler.dao.entity.WorkerGroup;
 import org.apache.dolphinscheduler.dao.mapper.ProcessInstanceMapper;
-import org.apache.dolphinscheduler.dao.mapper.WorkerGroupMapper;
-import com.baomidou.mybatisplus.core.metadata.IPage;
-import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
 import org.apache.dolphinscheduler.service.zk.ZookeeperCachedOperator;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
-import org.springframework.transaction.annotation.Transactional;
 
 import java.util.*;
+import java.util.stream.Collectors;
 
 /**
  * work group service
@@ -42,90 +42,13 @@ import java.util.*;
 public class WorkerGroupService extends BaseService {
 
 
-    @Autowired
-    WorkerGroupMapper workerGroupMapper;
-
     @Autowired
     ProcessInstanceMapper processInstanceMapper;
 
     @Autowired
     protected ZookeeperCachedOperator zookeeperCachedOperator;
 
-    /**
-     * create or update a worker group
-     *
-     * @param loginUser login user
-     * @param id worker group id
-     * @param name worker group name
-     * @param ipList ip list
-     * @return create or update result code
-     */
-    public Map<String, Object> saveWorkerGroup(User loginUser,int id, String name, String ipList){
-
-        Map<String, Object> result = new HashMap<>(5);
-
-        //only admin can operate
-        if (checkAdmin(loginUser, result)){
-            return result;
-        }
-
-        if(StringUtils.isEmpty(name)){
-            putMsg(result, Status.NAME_NULL);
-            return result;
-        }
-        Date now = new Date();
-        WorkerGroup workerGroup = null;
-        if(id != 0){
-            workerGroup = workerGroupMapper.selectById(id);
-            //check exist
-            if (workerGroup == null){
-                workerGroup = new WorkerGroup();
-                workerGroup.setCreateTime(now);
-            }
-        }else{
-            workerGroup = new WorkerGroup();
-            workerGroup.setCreateTime(now);
-        }
-        workerGroup.setName(name);
-        workerGroup.setIpList(ipList);
-        workerGroup.setUpdateTime(now);
 
-        if(checkWorkerGroupNameExists(workerGroup)){
-            putMsg(result, Status.NAME_EXIST, workerGroup.getName());
-            return result;
-        }
-        if(workerGroup.getId() != 0 ){
-            workerGroupMapper.updateById(workerGroup);
-        }else{
-            workerGroupMapper.insert(workerGroup);
-        }
-        putMsg(result, Status.SUCCESS);
-        return result;
-    }
-
-    /**
-     * check worker group name exists
-     * @param workerGroup
-     * @return
-     */
-    private boolean checkWorkerGroupNameExists(WorkerGroup workerGroup) {
-
-        List<WorkerGroup> workerGroupList = workerGroupMapper.queryWorkerGroupByName(workerGroup.getName());
-
-        if(CollectionUtils.isNotEmpty(workerGroupList)){
-            // new group has same name..
-            if(workerGroup.getId() == 0){
-                return true;
-            }
-            // update group...
-            for(WorkerGroup group : workerGroupList){
-                if(group.getId() != workerGroup.getId()){
-                    return true;
-                }
-            }
-        }
-        return false;
-    }
 
     /**
      * query worker group paging
@@ -138,66 +61,100 @@ public class WorkerGroupService extends BaseService {
      */
     public Map<String,Object> queryAllGroupPaging(User loginUser, Integer pageNo, Integer pageSize, String searchVal) {
 
+        // list from index
+        Integer fromIndex = (pageNo - 1) * pageSize;
+        // list to index
+        Integer toIndex = (pageNo - 1) * pageSize + pageSize;
+
         Map<String, Object> result = new HashMap<>(5);
         if (checkAdmin(loginUser, result)) {
             return result;
         }
 
-        Page<WorkerGroup> page = new Page(pageNo, pageSize);
-        IPage<WorkerGroup> workerGroupIPage = workerGroupMapper.queryListPaging(
-                page, searchVal);
+        List<WorkerGroup> workerGroups = getWorkerGroups(true);
+
+        List<WorkerGroup> resultDataList = new ArrayList<>();
+
+        if (CollectionUtils.isNotEmpty(workerGroups)){
+            List<WorkerGroup> searchValDataList = new ArrayList<>();
+
+            if (StringUtils.isNotEmpty(searchVal)){
+                for (WorkerGroup workerGroup : workerGroups){
+                    if (workerGroup.getName().contains(searchVal)){
+                        searchValDataList.add(workerGroup);
+                    }
+                }
+            }else {
+                searchValDataList = workerGroups;
+            }
+
+            if (searchValDataList.size() < pageSize){
+                toIndex = (pageNo - 1) * pageSize + searchValDataList.size();
+            }
+            resultDataList = searchValDataList.subList(fromIndex, toIndex);
+        }
+
         PageInfo<WorkerGroup> pageInfo = new PageInfo<>(pageNo, pageSize);
-        pageInfo.setTotalCount((int)workerGroupIPage.getTotal());
-        pageInfo.setLists(workerGroupIPage.getRecords());
+        pageInfo.setTotalCount(resultDataList.size());
+        pageInfo.setLists(resultDataList);
+
         result.put(Constants.DATA_LIST, pageInfo);
         putMsg(result, Status.SUCCESS);
         return result;
     }
 
+
+
     /**
-     * delete worker group by id
-     * @param id worker group id
-     * @return delete result code
+     * query all worker group
+     *
+     * @return all worker group list
      */
-    @Transactional(rollbackFor = Exception.class)
-    public Map<String,Object> deleteWorkerGroupById(Integer id) {
+    public Map<String,Object> queryAllGroup() {
+        Map<String, Object> result = new HashMap<>();
 
-        Map<String, Object> result = new HashMap<>(5);
+        List<WorkerGroup> workerGroups = getWorkerGroups(false);
 
-        List<ProcessInstance> processInstances = processInstanceMapper.queryByWorkerGroupIdAndStatus(id, Constants.NOT_TERMINATED_STATES);
-        if(CollectionUtils.isNotEmpty(processInstances)){
-            putMsg(result, Status.DELETE_WORKER_GROUP_BY_ID_FAIL, processInstances.size());
-            return result;
-        }
-        workerGroupMapper.deleteById(id);
-        processInstanceMapper.updateProcessInstanceByWorkerGroupId(id, Constants.DEFAULT_WORKER_ID);
+        Set<String> availableWorkerGroupSet = workerGroups.stream()
+                .map(workerGroup -> workerGroup.getName())
+                .collect(Collectors.toSet());
+        result.put(Constants.DATA_LIST, availableWorkerGroupSet);
         putMsg(result, Status.SUCCESS);
         return result;
     }
 
+
     /**
-     * query all worker group
+     *  get worker groups
      *
-     * @return all worker group list
+     * @param isPaging whether paging
+     * @return WorkerGroup list
      */
-    public Map<String,Object> queryAllGroup() {
-        Map<String, Object> result = new HashMap<>();
+    private List<WorkerGroup> getWorkerGroups(boolean isPaging) {
         String workerPath = zookeeperCachedOperator.getZookeeperConfig().getDsRoot()+"/nodes" +"/worker";
         List<String> workerGroupList = zookeeperCachedOperator.getChildrenKeys(workerPath);
 
         // available workerGroup list
         List<String> availableWorkerGroupList = new ArrayList<>();
 
+        List<WorkerGroup> workerGroups = new ArrayList<>();
+
         for (String workerGroup : workerGroupList){
             String workerGroupPath= workerPath + "/" + workerGroup;
             List<String> childrenNodes = zookeeperCachedOperator.getChildrenKeys(workerGroupPath);
             if (CollectionUtils.isNotEmpty(childrenNodes)){
                 availableWorkerGroupList.add(workerGroup);
+                WorkerGroup wg = new WorkerGroup();
+                wg.setName(workerGroup);
+                if (isPaging){
+                    wg.setIpList(childrenNodes);
+                    String registeredIpValue = zookeeperCachedOperator.get(workerGroupPath + "/" + childrenNodes.get(0));
+                    wg.setCreateTime(DateUtils.stringToDate(registeredIpValue.split(",")[3]));
+                    wg.setUpdateTime(DateUtils.stringToDate(registeredIpValue.split(",")[4]));
+                }
+                workerGroups.add(wg);
             }
         }
-
-        result.put(Constants.DATA_LIST, availableWorkerGroupList);
-        putMsg(result, Status.SUCCESS);
-        return result;
+        return workerGroups;
     }
 }

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

@@ -80,8 +80,7 @@ public class ProcessInstanceServiceTest {
     @Mock
     LoggerService loggerService;
 
-    @Mock
-    WorkerGroupMapper workerGroupMapper;
+
 
     @Mock
     UsersService usersService;
@@ -486,7 +485,6 @@ public class ProcessInstanceServiceTest {
      */
     private WorkerGroup getWorkGroup() {
         WorkerGroup workerGroup = new WorkerGroup();
-        workerGroup.setId(1);
         workerGroup.setName("test_workergroup");
         return workerGroup;
     }

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

@@ -26,10 +26,10 @@ import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
 import org.apache.dolphinscheduler.dao.entity.User;
 import org.apache.dolphinscheduler.dao.entity.WorkerGroup;
 import org.apache.dolphinscheduler.dao.mapper.ProcessInstanceMapper;
-import org.apache.dolphinscheduler.dao.mapper.WorkerGroupMapper;
 import org.apache.dolphinscheduler.service.zk.ZookeeperCachedOperator;
 import org.apache.dolphinscheduler.service.zk.ZookeeperConfig;
 import org.junit.Assert;
+import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.mockito.InjectMocks;
@@ -43,6 +43,7 @@ import org.slf4j.LoggerFactory;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 
 @RunWith(MockitoJUnitRunner.class)
 public class WorkerGroupServiceTest {
@@ -51,100 +52,55 @@ public class WorkerGroupServiceTest {
 
     @InjectMocks
     private WorkerGroupService workerGroupService;
-    @Mock
-    private WorkerGroupMapper workerGroupMapper;
+
     @Mock
     private ProcessInstanceMapper processInstanceMapper;
+
     @Mock
     private ZookeeperCachedOperator zookeeperCachedOperator;
 
-    private String groupName="groupName000001";
 
-    /**
-     *  create or update a worker group
-     */
-    @Test
-    public void testSaveWorkerGroup(){
+    @Before
+    public void init(){
+        ZookeeperConfig zookeeperConfig = new ZookeeperConfig();
+        zookeeperConfig.setDsRoot("/dolphinscheduler_qzw");
+        Mockito.when(zookeeperCachedOperator.getZookeeperConfig()).thenReturn(zookeeperConfig);
 
-        User user = new User();
-        // general user add
-        user.setUserType(UserType.GENERAL_USER);
-        Map<String, Object> result = workerGroupService.saveWorkerGroup(user, 0, groupName, "127.0.0.1");
-        logger.info(result.toString());
-        Assert.assertEquals( Status.USER_NO_OPERATION_PERM.getMsg(),(String) result.get(Constants.MSG));
+        String workerPath = zookeeperCachedOperator.getZookeeperConfig().getDsRoot()+"/nodes" +"/worker";
 
-        //success
-        user.setUserType(UserType.ADMIN_USER);
-        result = workerGroupService.saveWorkerGroup(user, 0, groupName, "127.0.0.1");
-        logger.info(result.toString());
-        Assert.assertEquals(Status.SUCCESS.getMsg(),(String)result.get(Constants.MSG));
-        // group name exist
-        Mockito.when(workerGroupMapper.selectById(2)).thenReturn(getWorkerGroup(2));
-        Mockito.when(workerGroupMapper.queryWorkerGroupByName(groupName)).thenReturn(getList());
-        result = workerGroupService.saveWorkerGroup(user, 2, groupName, "127.0.0.1");
-        logger.info(result.toString());
-        Assert.assertEquals(Status.NAME_EXIST,result.get(Constants.STATUS));
+        List<String> workerGroupStrList = new ArrayList<>();
+        workerGroupStrList.add("default");
+        workerGroupStrList.add("test");
+        Mockito.when(zookeeperCachedOperator.getChildrenKeys(workerPath)).thenReturn(workerGroupStrList);
+
+        List<String> defaultIpList = new ArrayList<>();
+        defaultIpList.add("192.168.220.188:1234");
+        defaultIpList.add("192.168.220.189:1234");
 
+        Mockito.when(zookeeperCachedOperator.getChildrenKeys(workerPath + "/default")).thenReturn(defaultIpList);
+
+        Mockito.when(zookeeperCachedOperator.get(workerPath + "/default" + "/" + defaultIpList.get(0))).thenReturn("0.02,0.23,0.03,2020-05-08 11:24:14,2020-05-08 14:22:24");
     }
 
     /**
      *  query worker group paging
      */
     @Test
-    public  void testQueryAllGroupPaging(){
-
+    public void testQueryAllGroupPaging(){
         User user = new User();
         // general user add
-        user.setUserType(UserType.GENERAL_USER);
-        Map<String, Object> result = workerGroupService.queryAllGroupPaging(user, 1, 10, groupName);
-        logger.info(result.toString());
-        Assert.assertEquals((String) result.get(Constants.MSG), Status.USER_NO_OPERATION_PERM.getMsg());
-        //success
         user.setUserType(UserType.ADMIN_USER);
-        Page<WorkerGroup> page = new Page<>(1,10);
-        page.setRecords(getList());
-        page.setSize(1L);
-        Mockito.when(workerGroupMapper.queryListPaging(Mockito.any(Page.class), Mockito.eq(groupName))).thenReturn(page);
-        result = workerGroupService.queryAllGroupPaging(user, 1, 10, groupName);
-        logger.info(result.toString());
-        Assert.assertEquals(Status.SUCCESS.getMsg(),(String)result.get(Constants.MSG));
-        PageInfo<WorkerGroup>  pageInfo = (PageInfo<WorkerGroup>) result.get(Constants.DATA_LIST);
-        Assert.assertTrue(CollectionUtils.isNotEmpty(pageInfo.getLists()));
+        Map<String, Object> result = workerGroupService.queryAllGroupPaging(user, 1, 10, null);
+        PageInfo<WorkerGroup> pageInfo = (PageInfo) result.get(Constants.DATA_LIST);
+        Assert.assertEquals(pageInfo.getLists().size(),1);
     }
 
-    /**
-     * delete group by id
-     */
-    @Test
-    public  void testDeleteWorkerGroupById(){
-
-        //DELETE_WORKER_GROUP_BY_ID_FAIL
-        Mockito.when(processInstanceMapper.queryByWorkerGroupIdAndStatus(1, Constants.NOT_TERMINATED_STATES)).thenReturn(getProcessInstanceList());
-        Map<String, Object> result = workerGroupService.deleteWorkerGroupById(1);
-        logger.info(result.toString());
-        Assert.assertEquals(Status.DELETE_WORKER_GROUP_BY_ID_FAIL.getCode(),((Status) result.get(Constants.STATUS)).getCode());
-
-        //correct
-        result = workerGroupService.deleteWorkerGroupById(2);
-        logger.info(result.toString());
-        Assert.assertEquals(Status.SUCCESS.getMsg(),(String)result.get(Constants.MSG));
-
-    }
 
     @Test
     public void testQueryAllGroup() throws Exception {
-        ZookeeperConfig zookeeperConfig = new ZookeeperConfig();
-        zookeeperConfig.setDsRoot("/ds");
-        Mockito.when(zookeeperCachedOperator.getZookeeperConfig()).thenReturn(zookeeperConfig);
-        List<String> workerGroupStrList = new ArrayList<>();
-        workerGroupStrList.add("workerGroup1");
-        Mockito.when(zookeeperCachedOperator.getChildrenKeys(Mockito.anyString())).thenReturn(workerGroupStrList);
-
         Map<String, Object> result = workerGroupService.queryAllGroup();
-        logger.info(result.toString());
-        Assert.assertEquals(Status.SUCCESS.getMsg(),(String)result.get(Constants.MSG));
-        List<WorkerGroup> workerGroupList = (List<WorkerGroup>) result.get(Constants.DATA_LIST);
-        Assert.assertTrue(workerGroupList.size()>0);
+        Set<String> workerGroups = (Set<String>) result.get(Constants.DATA_LIST);
+        Assert.assertEquals(workerGroups.size(), 1);
     }
 
 
@@ -158,25 +114,5 @@ public class WorkerGroupServiceTest {
         processInstances.add(new ProcessInstance());
         return processInstances;
     }
-    /**
-     * get Group
-     * @return
-     */
-    private WorkerGroup getWorkerGroup(int id){
-        WorkerGroup workerGroup = new WorkerGroup();
-        workerGroup.setName(groupName);
-        workerGroup.setId(id);
-        return workerGroup;
-    }
-    private WorkerGroup getWorkerGroup(){
-
-        return getWorkerGroup(1);
-    }
-
-   private List<WorkerGroup> getList(){
-        List<WorkerGroup> list = new ArrayList<>();
-        list.add(getWorkerGroup());
-        return list;
-   }
 
 }

dolphinscheduler-common/src/main/resources/common.properties

@@ -19,22 +19,22 @@
 resource.storage.type=NONE
 
 # resource store on HDFS/S3 path, resource file will store to this hadoop hdfs path, self configuration, please make sure the directory exists on hdfs and have read write permissions。"/dolphinscheduler" is recommended
-resource.upload.path=/dolphinscheduler
+#resource.upload.path=/dolphinscheduler
 
 # user data local directory path, please make sure the directory exists and have read write permissions
 #data.basedir.path=/tmp/dolphinscheduler
 
 # whether kerberos starts
-hadoop.security.authentication.startup.state=false
+#hadoop.security.authentication.startup.state=false
 
 # java.security.krb5.conf path
-java.security.krb5.conf.path=/opt/krb5.conf
+#java.security.krb5.conf.path=/opt/krb5.conf
 
 # login user from keytab username
-login.user.keytab.username=hdfs-mycluster@ESZ.COM
+#login.user.keytab.username=hdfs-mycluster@ESZ.COM
 
 # loginUserFromKeytab path
-login.user.keytab.path=/opt/hdfs.headless.keytab
+#login.user.keytab.path=/opt/hdfs.headless.keytab
 
 #resource.view.suffixs
 #resource.view.suffixs=txt,log,sh,conf,cfg,py,java,sql,hql,xml,properties
@@ -46,18 +46,18 @@ hdfs.root.user=hdfs
 fs.defaultFS=hdfs://mycluster:8020
 
 # if resource.storage.type=S3，s3 endpoint
-fs.s3a.endpoint=http://192.168.199.91:9010
+#fs.s3a.endpoint=http://192.168.199.91:9010
 
 # if resource.storage.type=S3，s3 access key
-fs.s3a.access.key=A3DXS30FO22544RE
+#fs.s3a.access.key=A3DXS30FO22544RE
 
 # if resource.storage.type=S3，s3 secret key
-fs.s3a.secret.key=OloCLq3n+8+sdPHUhJ21XrSxTC+JK
+#fs.s3a.secret.key=OloCLq3n+8+sdPHUhJ21XrSxTC+JK
 
 # if not use hadoop resourcemanager, please keep default value; if resourcemanager HA enable, please type the HA ips ; if resourcemanager is single, make this value empty
 yarn.resourcemanager.ha.rm.ids=192.168.xx.xx,192.168.xx.xx
 
-# if resourcemanager HA enable or not use resourcemanager, please keep the default value; If resourcemanager is single, you only need to replace ark1 to actual resourcemanager hostname.
+# If resourcemanager HA enable or not use resourcemanager, please keep the default value; If resourcemanager is single, you only need to replace ark1 to actual resourcemanager hostname.
 yarn.application.status.address=http://ark1:8088/ws/v1/cluster/apps/%s
 
 # system env path

dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/WorkerGroup.java

@@ -21,41 +21,22 @@ import com.baomidou.mybatisplus.annotation.TableId;
 import com.baomidou.mybatisplus.annotation.TableName;
 
 import java.util.Date;
+import java.util.List;
 
 /**
- * worker group for task running
+ * worker group
  */
-@TableName("t_ds_worker_group")
 public class WorkerGroup {
 
-    @TableId(value="id", type=IdType.AUTO)
-    private int id;
-
     private String name;
 
-    private String ipList;
+    private List<String> ipList;
 
     private Date createTime;
 
     private Date updateTime;
 
 
-    public int getId() {
-        return id;
-    }
-
-    public void setId(int id) {
-        this.id = id;
-    }
-
-    public String getIpList() {
-        return ipList;
-    }
-
-    public void setIpList(String ipList) {
-        this.ipList = ipList;
-    }
-
     public Date getCreateTime() {
         return createTime;
     }
@@ -72,18 +53,6 @@ public class WorkerGroup {
         this.updateTime = updateTime;
     }
 
-    @Override
-    public String toString() {
-        return "Worker group model{" +
-                "id= " + id +
-                ",name= " + name +
-                ",ipList= " + ipList +
-                ",createTime= " + createTime +
-                ",updateTime= " + updateTime +
-
-                "}";
-    }
-
     public String getName() {
         return name;
     }
@@ -91,4 +60,14 @@ public class WorkerGroup {
     public void setName(String name) {
         this.name = name;
     }
+
+    public List<String> getIpList() {
+        return ipList;
+    }
+
+    public void setIpList(List<String> ipList) {
+        this.ipList = ipList;
+    }
+
+
 }

dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/WorkerGroupMapper.java

@@ -1,54 +0,0 @@
-/*
- * 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.WorkerGroup;
-import com.baomidou.mybatisplus.core.mapper.BaseMapper;
-import com.baomidou.mybatisplus.core.metadata.IPage;
-import org.apache.ibatis.annotations.Param;
-
-import java.util.List;
-
-/**
- * worker group mapper interface
- */
-public interface WorkerGroupMapper extends BaseMapper<WorkerGroup> {
-
-    /**
-     * query all worker group
-     * @return worker group list
-     */
-    List<WorkerGroup> queryAllWorkerGroup();
-
-    /**
-     * query worer grouop by name
-     * @param name name
-     * @return worker group list
-     */
-    List<WorkerGroup> queryWorkerGroupByName(@Param("name") String name);
-
-    /**
-     * worker group page
-     * @param page page
-     * @param searchVal searchVal
-     * @return worker group IPage
-     */
-    IPage<WorkerGroup> queryListPaging(IPage<WorkerGroup> page,
-                                       @Param("searchVal") String searchVal);
-
-}
-

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/WorkerServer.java

@@ -71,9 +71,9 @@ public class WorkerServer {
     private SpringApplicationContext springApplicationContext;
 
     /**
-     * master server startup
+     * worker server startup
      *
-     * master server not use web service
+     * worker server not use web service
      * @param args arguments
      */
     public static void main(String[] args) {
@@ -140,4 +140,4 @@ public class WorkerServer {
         }
     }
 
-}
+}

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/processor/TaskExecuteProcessor.java

@@ -101,9 +101,15 @@ public class TaskExecuteProcessor implements NettyRequestProcessor {
         taskCallbackService.addRemoteChannel(taskExecutionContext.getTaskInstanceId(),
                 new NettyRemoteChannel(channel, command.getOpaque()));
 
-        this.doAck(taskExecutionContext);
+        try {
+            this.doAck(taskExecutionContext);
+        }catch (Exception e){
+            ThreadUtils.sleep(Constants.SLEEP_TIME_MILLIS);
+            this.doAck(taskExecutionContext);
+        }
+
         // submit task
-        workerExecService.submit(new TaskExecuteThread(taskExecutionContext,taskCallbackService));
+        workerExecService.submit(new TaskExecuteThread(taskExecutionContext, taskCallbackService));
     }
 
     private void doAck(TaskExecutionContext taskExecutionContext){

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/runner/TaskExecuteThread.java

@@ -18,10 +18,12 @@ package org.apache.dolphinscheduler.server.worker.runner;
 
 
 import com.alibaba.fastjson.JSONObject;
+import org.apache.dolphinscheduler.common.Constants;
 import org.apache.dolphinscheduler.common.enums.ExecutionStatus;
 import org.apache.dolphinscheduler.common.model.TaskNode;
 import org.apache.dolphinscheduler.common.process.Property;
 import org.apache.dolphinscheduler.common.task.TaskTimeoutParameter;
+import org.apache.dolphinscheduler.common.thread.ThreadUtils;
 import org.apache.dolphinscheduler.common.utils.*;
 import org.apache.dolphinscheduler.remote.command.TaskExecuteResponseCommand;
 import org.apache.dolphinscheduler.server.entity.TaskExecutionContext;
@@ -131,7 +133,12 @@ public class TaskExecuteThread implements Runnable {
             responseCommand.setProcessId(task.getProcessId());
             responseCommand.setAppIds(task.getAppIds());
         } finally {
-            taskCallbackService.sendResult(taskExecutionContext.getTaskInstanceId(), responseCommand.convert2Command());
+            try {
+                taskCallbackService.sendResult(taskExecutionContext.getTaskInstanceId(), responseCommand.convert2Command());
+            }catch (Exception e){
+                ThreadUtils.sleep(Constants.SLEEP_TIME_MILLIS);
+                taskCallbackService.sendResult(taskExecutionContext.getTaskInstanceId(), responseCommand.convert2Command());
+            }
         }
     }
 

dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/registry/DependencyConfig.java

@@ -113,10 +113,7 @@ public class DependencyConfig {
         return Mockito.mock(ResourceMapper.class);
     }
 
-    @Bean
-    public WorkerGroupMapper workerGroupMapper(){
-        return Mockito.mock(WorkerGroupMapper.class);
-    }
+
 
     @Bean
     public ErrorCommandMapper errorCommandMapper(){

dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/worker/processor/TaskCallbackServiceTestConfig.java

@@ -107,10 +107,6 @@ public class TaskCallbackServiceTestConfig {
         return Mockito.mock(ResourceMapper.class);
     }
 
-    @Bean
-    public WorkerGroupMapper workerGroupMapper(){
-        return Mockito.mock(WorkerGroupMapper.class);
-    }
 
     @Bean
     public ErrorCommandMapper errorCommandMapper(){

dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessService.java

@@ -86,8 +86,7 @@ public class ProcessService {
     @Autowired
     private ResourceMapper resourceMapper;
 
-    @Autowired
-    private WorkerGroupMapper workerGroupMapper;
+
 
     @Autowired
     private ErrorCommandMapper errorCommandMapper;
@@ -1670,15 +1669,7 @@ public class ProcessService {
         return queue;
     }
 
-    /**
-     * query worker group by id
-     * @param workerGroupId workerGroupId
-     * @return WorkerGroup
-     */
-    public WorkerGroup queryWorkerGroupById(int workerGroupId){
 
-        return workerGroupMapper.selectById(workerGroupId);
-    }
 
     /**
      * get task worker group

dolphinscheduler-ui/src/js/conf/home/pages/resource/pages/file/pages/details/index.vue

@@ -118,7 +118,7 @@
        * up
        */
       _onUp: _.debounce(function () {
-        this.loadingIndex = this.loadingIndex - 2
+        this.loadingIndex = this.loadingIndex - 3
         console.log('_onUp')
         this._editorOff()
 
@@ -131,7 +131,7 @@
        * down
        */
       _onDown: _.debounce(function () {
-        this.loadingIndex = this.loadingIndex + 2
+        this.loadingIndex = this.loadingIndex + 3
         console.log('_onDown')
         this._editorOff()
 
@@ -166,11 +166,11 @@
           // down
           if ((scrollTop + h) > totalHeight) {
             if (this.isData) {
-              this._onDown()
+              // this._onDown()
             }
           }
           // up
-          if (scrollTop < 2) {
+          if (scrollTop < 3) {
             if (this.loadingIndex > 0) {
               this._onUp()
             }
@@ -210,7 +210,7 @@
         return {
           id: this.$route.params.id,
           skipLineNum: parseInt(`${this.loadingIndex ? this.loadingIndex + '000' : 0}`),
-          limit: parseInt(`${this.loadingIndex ? this.loadingIndex + 2 : 2}000`)
+          limit: parseInt(`${this.loadingIndex ? this.loadingIndex + 3 : 3}000`)
         }
       }
     },

dolphinscheduler-ui/src/js/conf/home/pages/security/pages/workerGroups/_source/list.vue

@@ -34,9 +34,6 @@
           <th>
             <span>{{$t('Update Time')}}</span>
           </th>
-          <th width="70">
-            <span>{{$t('Operation')}}</span>
-          </th>
         </tr>
         <tr v-for="(item, $index) in list" :key="$index">
           <td>
@@ -48,7 +45,7 @@
             </span>
           </td>
           <td>
-            <span>{{item.ipList}}</span>
+            <span>{{item.ipList.join(',')}}</span>
           </td>
           <td>
             <span v-if="item.createTime">{{item.createTime | formatDate}}</span>
@@ -58,24 +55,6 @@
             <span v-if="item.updateTime">{{item.updateTime | formatDate}}</span>
             <span v-else>-</span>
           </td>
-          <td>
-            <x-button type="info" shape="circle" size="xsmall" data-toggle="tooltip" icon="ans-icon-edit" :title="$t('Edit')" @click="_edit(item)">
-            </x-button>
-            <x-poptip
-                    :ref="'poptip-delete-' + $index"
-                    placement="bottom-end"
-                    width="90">
-              <p>{{$t('Delete?')}}</p>
-              <div style="text-align: right; margin: 0;padding-top: 4px;">
-                <x-button type="text" size="xsmall" shape="circle" @click="_closeDelete($index)">{{$t('Cancel')}}</x-button>
-                <x-button type="primary" size="xsmall" shape="circle" @click="_delete(item,$index)">{{$t('Confirm')}}</x-button>
-              </div>
-              <template slot="reference">
-                <x-button type="error" shape="circle" size="xsmall" data-toggle="tooltip" icon="ans-icon-trash" :title="$t('delete')">
-                </x-button>
-              </template>
-            </x-poptip>
-          </td>
         </tr>
       </table>
     </div>
@@ -128,8 +107,7 @@
     created () {
       this.list = this.workerGroupList
     },
-    mounted () {
-    },
-    components: { }
+    mounted () {},
+    components: {},
   }
 </script>

dolphinscheduler-ui/src/js/conf/home/pages/security/pages/workerGroups/index.vue

@@ -17,11 +17,7 @@
 <template>
   <m-list-construction :title="$t('Worker group manage')">
     <template slot="conditions">
-      <m-conditions @on-conditions="_onConditions">
-        <template slot="button-group" v-if="isADMIN">
-          <x-button type="ghost" size="small" @click="_create('')">{{$t('Create worker group')}}</x-button>
-        </template>
-      </m-conditions>
+      <m-conditions @on-conditions="_onConditions"></m-conditions>
     </template>
     <template slot="content">
       <template v-if="workerGroupList.length || total>0">
@@ -141,8 +137,7 @@
         this.searchParams.pageNo = _.isEmpty(a.query) ? 1 : a.query.pageNo
       }
     },
-    created () {
-    },
+    created () {},
     mounted () {
       this.$modal.destroy()
     },

dolphinscheduler-ui/src/js/conf/home/router/index.js

@@ -374,6 +374,14 @@ const router = new Router({
             title: `${i18n.$t('Queue manage')}`
           }
         },
+        {
+          path: '/security/worker-groups',
+          name: 'worker-groups-manage',
+          component: resolve => require(['../pages/security/pages/workerGroups/index'], resolve),
+          meta: {
+            title: `${i18n.$t('Worker group manage')}`
+          }
+        },
         {
           path: '/security/token',
           name: 'token-manage',

dolphinscheduler-ui/src/js/module/components/secondaryMenu/_source/menu.js

@@ -100,6 +100,15 @@ const menu = {
       icon: 'ans-icon-danger-solid',
       children: []
     },
+    {
+      name: `${i18n.$t('Worker group manage')}`,
+      id: 4,
+      path: 'worker-groups-manage',
+      isOpen: true,
+      disabled: true,
+      icon: 'ans-icon-diary',
+      children: []
+    },
     {
       name: `${i18n.$t('Queue manage')}`,
       id: 3,

sql/soft_version

@@ -1 +1 @@
-1.2.2
+1.3.0