[1.3.6-prepare][Feature-5062][*] Create/Edit/Delete Worker Group in Web UI #5113 (#5122)

* [1.3.6-prepare][Feature-5062][*] Create/Edit/Delete Worker Group in Web UI #5113

* [1.3.6-prepare][Feature-5062][*] Improve getServerMaps

* [Improvement][Test] Fix unit test

* [Improvement][SQL] Add Unique for name in table t_ds_worker_group

* [1.3.6-prepare][Feature-5062][*] Remove redundant changes

* [1.3.6-prepare][Improvement][Server] Optimize and reduce database and zookeeper query

* [1.3.6-prepare][Improvement][WorkerGroup] Replace worker address input with multiple-select

* [1.3.6-prepare][Improvement][WorkerGroup] Fix unit test

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

@@ -16,27 +16,38 @@
  */
 package org.apache.dolphinscheduler.api.controller;
 
+import static org.apache.dolphinscheduler.api.enums.Status.DELETE_WORKER_GROUP_FAIL;
+import static org.apache.dolphinscheduler.api.enums.Status.QUERY_WORKER_ADDRESS_LIST_FAIL;
+import static org.apache.dolphinscheduler.api.enums.Status.QUERY_WORKER_GROUP_FAIL;
+import static org.apache.dolphinscheduler.api.enums.Status.SAVE_ERROR;
 
 import org.apache.dolphinscheduler.api.exceptions.ApiException;
 import org.apache.dolphinscheduler.api.service.WorkerGroupService;
+import org.apache.dolphinscheduler.api.utils.RegexUtils;
 import org.apache.dolphinscheduler.api.utils.Result;
 import org.apache.dolphinscheduler.common.Constants;
 import org.apache.dolphinscheduler.common.utils.ParameterUtils;
 import org.apache.dolphinscheduler.dao.entity.User;
-import io.swagger.annotations.Api;
-import io.swagger.annotations.ApiImplicitParam;
-import io.swagger.annotations.ApiImplicitParams;
-import io.swagger.annotations.ApiOperation;
+
+import java.util.Map;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.http.HttpStatus;
-import org.springframework.web.bind.annotation.*;
-import springfox.documentation.annotations.ApiIgnore;
-
-import java.util.Map;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestAttribute;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.ResponseStatus;
+import org.springframework.web.bind.annotation.RestController;
 
-import static org.apache.dolphinscheduler.api.enums.Status.*;
+import io.swagger.annotations.Api;
+import io.swagger.annotations.ApiImplicitParam;
+import io.swagger.annotations.ApiImplicitParams;
+import io.swagger.annotations.ApiOperation;
+import springfox.documentation.annotations.ApiIgnore;
 
 /**
  * worker group controller
@@ -51,8 +62,34 @@ public class WorkerGroupController extends BaseController {
     @Autowired
     WorkerGroupService workerGroupService;
 
-
-
+    /**
+     * create or update a worker group
+     *
+     * @param loginUser login user
+     * @param id        worker group id
+     * @param name      worker group name
+     * @param addrList  addr 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 = "addrList", value = "WORKER_ADDR_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 = "addrList") String addrList
+    ) {
+        logger.info("save worker group: login user {}, id:{}, name: {}, addrList: {} ",
+                RegexUtils.escapeNRT(loginUser.getUserName()), id, RegexUtils.escapeNRT(name), RegexUtils.escapeNRT(addrList));
+        Map<String, Object> result = workerGroupService.saveWorkerGroup(loginUser, id, name, addrList);
+        return returnDataList(result);
+    }
 
     /**
      * query worker groups paging
@@ -65,21 +102,20 @@ public class WorkerGroupController extends BaseController {
      */
     @ApiOperation(value = "queryAllWorkerGroupsPaging", notes = "QUERY_WORKER_GROUP_PAGING_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")
+            @ApiImplicitParam(name = "pageNo", value = "PAGE_NO", dataType = "Int", example = "1"),
+            @ApiImplicitParam(name = "pageSize", value = "PAGE_SIZE", dataType = "Int", example = "20"),
+            @ApiImplicitParam(name = "searchVal", value = "SEARCH_VAL", dataType = "String")
     })
     @GetMapping(value = "/list-paging")
     @ResponseStatus(HttpStatus.OK)
     @ApiException(QUERY_WORKER_GROUP_FAIL)
     public Result queryAllWorkerGroupsPaging(@ApiIgnore @RequestAttribute(value = Constants.SESSION_USER) User loginUser,
                                              @RequestParam("pageNo") Integer pageNo,
-                                             @RequestParam(value = "searchVal", required = false) String searchVal,
-                                             @RequestParam("pageSize") Integer pageSize
+                                             @RequestParam("pageSize") Integer pageSize,
+                                             @RequestParam(value = "searchVal", required = false) String searchVal
     ) {
         logger.info("query all worker group paging: login user {}, pageNo:{}, pageSize:{}, searchVal:{}",
-                loginUser.getUserName(), pageNo, pageSize, searchVal);
-
+                RegexUtils.escapeNRT(loginUser.getUserName()), pageNo, pageSize, searchVal);
         searchVal = ParameterUtils.handleEscapes(searchVal);
         Map<String, Object> result = workerGroupService.queryAllGroupPaging(loginUser, pageNo, pageSize, searchVal);
         return returnDataListPaging(result);
@@ -95,14 +131,48 @@ public class WorkerGroupController extends BaseController {
     @GetMapping(value = "/all-groups")
     @ResponseStatus(HttpStatus.OK)
     @ApiException(QUERY_WORKER_GROUP_FAIL)
-    public Result queryAllWorkerGroups(@ApiIgnore @RequestAttribute(value = Constants.SESSION_USER) User loginUser
-    ) {
-        logger.info("query all worker group: login user {}",
-                loginUser.getUserName());
-
+    public Result queryAllWorkerGroups(@ApiIgnore @RequestAttribute(value = Constants.SESSION_USER) User loginUser) {
+        logger.info("query all worker group: login user {}", RegexUtils.escapeNRT(loginUser.getUserName()));
         Map<String, Object> result = workerGroupService.queryAllGroup();
         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"),
+    })
+    @PostMapping(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:{} ", RegexUtils.escapeNRT(loginUser.getUserName()), id);
+        Map<String, Object> result = workerGroupService.deleteWorkerGroupById(loginUser, id);
+        return returnDataList(result);
+    }
+
+    /**
+     * query worker address list
+     *
+     * @param loginUser login user
+     * @return all worker address list
+     */
+    @ApiOperation(value = "queryWorkerAddressList", notes = "QUERY_WORKER_ADDRESS_LIST_NOTES")
+    @GetMapping(value = "/worker-address-list")
+    @ResponseStatus(HttpStatus.OK)
+    @ApiException(QUERY_WORKER_ADDRESS_LIST_FAIL)
+    public Result queryWorkerAddressList(@ApiIgnore @RequestAttribute(value = Constants.SESSION_USER) User loginUser) {
+        logger.info("query worker address list: login user {}", RegexUtils.escapeNRT(loginUser.getUserName()));
+        Map<String, Object> result = workerGroupService.getWorkerAddressList();
+        return returnDataList(result);
+    }
 
 }

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/enums/Status.java

@@ -16,10 +16,10 @@
  */
 package org.apache.dolphinscheduler.api.enums;
 
-import org.springframework.context.i18n.LocaleContextHolder;
-
 import java.util.Locale;
 
+import org.springframework.context.i18n.LocaleContextHolder;
+
 /**
  *  status enum
  */
@@ -180,6 +180,11 @@ public enum Status {
     DATASOURCE_OTHER_PARAMS_ILLEGAL(10171, "datasource other params illegal", "数据源其他参数不合法"),
     DATASOURCE_NAME_ILLEGAL(10172, "datasource name illegal", "数据源名称不合法"),
     DATASOURCE_HOST_ILLEGAL(10173, "datasource host illegal", "数据源HOST不合法"),
+    DELETE_WORKER_GROUP_NOT_EXIST(10174, "delete worker group not exist ", "删除worker分组不存在"),
+    CREATE_WORKER_GROUP_FORBIDDEN_IN_DOCKER(10175, "create worker group forbidden in docker ", "创建worker分组在docker中禁止"),
+    DELETE_WORKER_GROUP_FORBIDDEN_IN_DOCKER(10176, "delete worker group forbidden in docker ", "删除worker分组在docker中禁止"),
+    WORKER_ADDRESS_INVALID(10177, "worker address {0} invalid", "worker地址[{0}]无效"),
+    QUERY_WORKER_ADDRESS_LIST_FAIL(10178, "query worker address list fail ", "查询worker地址列表失败"),
 
     UDF_FUNCTION_NOT_EXIST(20001, "UDF function not found", "UDF函数不存在"),
     UDF_FUNCTION_EXISTS(20002, "UDF function already exists", "UDF函数已存在"),

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

@@ -18,12 +18,6 @@ package org.apache.dolphinscheduler.api.service;
 
 import static org.apache.dolphinscheduler.common.utils.Preconditions.checkNotNull;
 
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.function.Function;
-import java.util.stream.Collectors;
-
 import org.apache.dolphinscheduler.api.enums.Status;
 import org.apache.dolphinscheduler.api.utils.ZookeeperMonitor;
 import org.apache.dolphinscheduler.common.Constants;
@@ -34,8 +28,16 @@ import org.apache.dolphinscheduler.dao.MonitorDBDao;
 import org.apache.dolphinscheduler.dao.entity.MonitorRecord;
 import org.apache.dolphinscheduler.dao.entity.User;
 import org.apache.dolphinscheduler.dao.entity.ZookeeperRecord;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
+
 import com.google.common.collect.Sets;
 
 /**
@@ -150,7 +152,7 @@ public class MonitorService extends BaseService {
 
     checkNotNull(zookeeperMonitor);
     ZKNodeType zkNodeType = isMaster ? ZKNodeType.MASTER : ZKNodeType.WORKER;
-    return zookeeperMonitor.getServersList(zkNodeType);
+    return zookeeperMonitor.getServerList(zkNodeType);
   }
 
 }

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

@@ -18,25 +18,31 @@ package org.apache.dolphinscheduler.api.service;
 
 import org.apache.dolphinscheduler.api.enums.Status;
 import org.apache.dolphinscheduler.api.utils.PageInfo;
+import org.apache.dolphinscheduler.api.utils.ZookeeperMonitor;
 import org.apache.dolphinscheduler.common.Constants;
+import org.apache.dolphinscheduler.common.enums.ZKNodeType;
 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.User;
 import org.apache.dolphinscheduler.dao.entity.WorkerGroup;
 import org.apache.dolphinscheduler.dao.mapper.ProcessInstanceMapper;
-import org.apache.dolphinscheduler.remote.utils.Host;
+import org.apache.dolphinscheduler.dao.mapper.WorkerGroupMapper;
 import org.apache.dolphinscheduler.service.zk.ZookeeperCachedOperator;
 
 import java.util.ArrayList;
+import java.util.Date;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.Set;
 import java.util.stream.Collectors;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
 
 /**
  * work group service
@@ -44,14 +50,115 @@ import org.springframework.stereotype.Service;
 @Service
 public class WorkerGroupService extends BaseService {
 
+    private static final Logger logger = LoggerFactory.getLogger(WorkerGroupService.class);
 
     @Autowired
-    ProcessInstanceMapper processInstanceMapper;
+    WorkerGroupMapper workerGroupMapper;
 
     @Autowired
     protected ZookeeperCachedOperator zookeeperCachedOperator;
 
+    @Autowired
+    private ZookeeperMonitor zookeeperMonitor;
 
+    @Autowired
+    ProcessInstanceMapper processInstanceMapper;
+
+    /**
+     * create or update a worker group
+     *
+     * @param loginUser login user
+     * @param id worker group id
+     * @param name worker group name
+     * @param addrList addr list
+     * @return create or update result code
+     */
+    public Map<String, Object> saveWorkerGroup(User loginUser, int id, String name, String addrList) {
+        Map<String, Object> result = new HashMap<>();
+        if (checkAdmin(loginUser, result)) {
+            return result;
+        }
+        if (Constants.DOCKER_MODE && !Constants.KUBERNETES_MODE) {
+            putMsg(result, Status.CREATE_WORKER_GROUP_FORBIDDEN_IN_DOCKER);
+            return result;
+        }
+        if (StringUtils.isEmpty(name)) {
+            putMsg(result, Status.NAME_NULL);
+            return result;
+        }
+        Date now = new Date();
+        WorkerGroup workerGroup;
+        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.setAddrList(addrList);
+        workerGroup.setUpdateTime(now);
+
+        if (checkWorkerGroupNameExists(workerGroup)) {
+            putMsg(result, Status.NAME_EXIST, workerGroup.getName());
+            return result;
+        }
+        String invalidAddr = checkWorkerGroupAddrList(workerGroup);
+        if (invalidAddr != null) {
+            putMsg(result, Status.WORKER_ADDRESS_INVALID, invalidAddr);
+            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 worker group
+     * @return boolean
+     */
+    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;
+            }
+            // check group id
+            for (WorkerGroup group : workerGroupList) {
+                if (group.getId() != workerGroup.getId()) {
+                    return true;
+                }
+            }
+        }
+        // check zookeeper
+        String workerGroupPath = zookeeperCachedOperator.getZookeeperConfig().getDsRoot() + Constants.ZOOKEEPER_DOLPHINSCHEDULER_WORKERS + Constants.SLASH + workerGroup.getName();
+        return zookeeperCachedOperator.isExisted(workerGroupPath);
+    }
+
+    /**
+     * check worker group addr list
+     * @param workerGroup worker group
+     * @return boolean
+     */
+    private String checkWorkerGroupAddrList(WorkerGroup workerGroup) {
+        Map<String, String> serverMaps = zookeeperMonitor.getServerMaps(ZKNodeType.WORKER, true);
+        for (String addr : workerGroup.getAddrList().split(Constants.COMMA)) {
+            if (!serverMaps.containsKey(addr)) {
+                return addr;
+            }
+        }
+        return null;
+    }
 
     /**
      * query worker group paging
@@ -62,39 +169,39 @@ public class WorkerGroupService extends BaseService {
      * @param pageSize page size
      * @return worker group list page
      */
-    public Map<String,Object> queryAllGroupPaging(User loginUser, Integer pageNo, Integer pageSize, String searchVal) {
-
+    public Map<String, Object> queryAllGroupPaging(User loginUser, Integer pageNo, Integer pageSize, String searchVal) {
         // list from index
-        Integer fromIndex = (pageNo - 1) * pageSize;
+        int fromIndex = (pageNo - 1) * pageSize;
         // list to index
-        Integer toIndex = (pageNo - 1) * pageSize + pageSize;
+        int toIndex = (pageNo - 1) * pageSize + pageSize;
 
-        Map<String, Object> result = new HashMap<>(5);
+        Map<String, Object> result = new HashMap<>();
         if (checkAdmin(loginUser, result)) {
             return result;
         }
 
         List<WorkerGroup> workerGroups = getWorkerGroups(true);
-
         List<WorkerGroup> resultDataList = new ArrayList<>();
 
-        if (CollectionUtils.isNotEmpty(workerGroups)){
+        if (CollectionUtils.isNotEmpty(workerGroups)) {
             List<WorkerGroup> searchValDataList = new ArrayList<>();
 
-            if (StringUtils.isNotEmpty(searchVal)){
-                for (WorkerGroup workerGroup : workerGroups){
-                    if (workerGroup.getName().contains(searchVal)){
+            if (StringUtils.isNotEmpty(searchVal)) {
+                for (WorkerGroup workerGroup : workerGroups) {
+                    if (workerGroup.getName().contains(searchVal)) {
                         searchValDataList.add(workerGroup);
                     }
                 }
-            }else {
+            } else {
                 searchValDataList = workerGroups;
             }
 
-            if (searchValDataList.size() < pageSize){
-                toIndex = (pageNo - 1) * pageSize + searchValDataList.size();
+            if (fromIndex < searchValDataList.size()) {
+                if (toIndex > searchValDataList.size()) {
+                    toIndex = searchValDataList.size();
+                }
+                resultDataList = searchValDataList.subList(fromIndex, toIndex);
             }
-            resultDataList = searchValDataList.subList(fromIndex, toIndex);
         }
 
         PageInfo<WorkerGroup> pageInfo = new PageInfo<>(pageNo, pageSize);
@@ -106,57 +213,121 @@ public class WorkerGroupService extends BaseService {
         return result;
     }
 
-
-
     /**
      * query all worker group
      *
      * @return all worker group list
      */
-    public Map<String,Object> queryAllGroup() {
+    public Map<String, Object> queryAllGroup() {
         Map<String, Object> result = new HashMap<>();
-
         List<WorkerGroup> workerGroups = getWorkerGroups(false);
-
-        Set<String> availableWorkerGroupSet = workerGroups.stream()
-                .map(workerGroup -> workerGroup.getName())
-                .collect(Collectors.toSet());
-        result.put(Constants.DATA_LIST, availableWorkerGroupSet);
+        List<String> availableWorkerGroupList = workerGroups.stream()
+                .map(WorkerGroup::getName)
+                .collect(Collectors.toList());
+        int index = availableWorkerGroupList.indexOf(Constants.DEFAULT_WORKER_GROUP);
+        if (index > -1) {
+            availableWorkerGroupList.remove(index);
+            availableWorkerGroupList.add(0, Constants.DEFAULT_WORKER_GROUP);
+        }
+        result.put(Constants.DATA_LIST, availableWorkerGroupList);
         putMsg(result, Status.SUCCESS);
         return result;
     }
 
-
     /**
-     *  get worker groups
+     * get worker groups
      *
      * @param isPaging whether paging
      * @return WorkerGroup list
      */
     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);
+        // worker groups from database
+        List<WorkerGroup> workerGroups = workerGroupMapper.queryAllWorkerGroup();
+        // worker groups from zookeeper
+        String workerPath = zookeeperCachedOperator.getZookeeperConfig().getDsRoot() + Constants.ZOOKEEPER_DOLPHINSCHEDULER_WORKERS;
+        List<String> workerGroupList = null;
+        try {
+            workerGroupList = zookeeperCachedOperator.getChildrenKeys(workerPath);
+        } catch (Exception e) {
+            logger.error("getWorkerGroups exception: {}, workerPath: {}, isPaging: {}", e.getMessage(), workerPath, isPaging);
+        }
+
+        if (CollectionUtils.isEmpty(workerGroupList)) {
+            if (CollectionUtils.isEmpty(workerGroups) && !isPaging) {
                 WorkerGroup wg = new WorkerGroup();
-                wg.setName(workerGroup);
-                if (isPaging){
-                    wg.setIpList(childrenNodes.stream().map(node -> Host.of(node).getIp()).collect(Collectors.toList()));
-                    String registeredValue = zookeeperCachedOperator.get(workerGroupPath + "/" + childrenNodes.get(0));
-                    wg.setCreateTime(DateUtils.stringToDate(registeredValue.split(",")[6]));
-                    wg.setUpdateTime(DateUtils.stringToDate(registeredValue.split(",")[7]));
-                }
+                wg.setName(Constants.DEFAULT_WORKER_GROUP);
                 workerGroups.add(wg);
             }
+            return workerGroups;
+        }
+
+        for (String workerGroup : workerGroupList) {
+            String workerGroupPath = workerPath + Constants.SLASH + workerGroup;
+            List<String> childrenNodes = null;
+            try {
+                childrenNodes = zookeeperCachedOperator.getChildrenKeys(workerGroupPath);
+            } catch (Exception e) {
+                logger.error("getChildrenNodes exception: {}, workerGroupPath: {}", e.getMessage(), workerGroupPath);
+            }
+            if (childrenNodes == null || childrenNodes.isEmpty()) {
+                continue;
+            }
+            WorkerGroup wg = new WorkerGroup();
+            wg.setName(workerGroup);
+            if (isPaging) {
+                wg.setAddrList(String.join(Constants.COMMA, childrenNodes));
+                String registeredValue = zookeeperCachedOperator.get(workerGroupPath + Constants.SLASH + childrenNodes.get(0));
+                wg.setCreateTime(DateUtils.stringToDate(registeredValue.split(Constants.COMMA)[6]));
+                wg.setUpdateTime(DateUtils.stringToDate(registeredValue.split(Constants.COMMA)[7]));
+                wg.setSystemDefault(true);
+            }
+            workerGroups.add(wg);
         }
         return workerGroups;
     }
+
+    /**
+     * delete worker group by id
+     * @param id worker group id
+     * @return delete result code
+     */
+    @Transactional(rollbackFor = Exception.class)
+    public Map<String, Object> deleteWorkerGroupById(User loginUser, Integer id) {
+        Map<String, Object> result = new HashMap<>();
+        if (checkAdmin(loginUser, result)) {
+            return result;
+        }
+        if (Constants.DOCKER_MODE && !Constants.KUBERNETES_MODE) {
+            putMsg(result, Status.DELETE_WORKER_GROUP_FORBIDDEN_IN_DOCKER);
+            return result;
+        }
+        WorkerGroup workerGroup = workerGroupMapper.selectById(id);
+        if (workerGroup == null) {
+            putMsg(result, Status.DELETE_WORKER_GROUP_NOT_EXIST);
+            return result;
+        }
+        List<ProcessInstance> processInstances = processInstanceMapper.queryByWorkerGroupNameAndStatus(workerGroup.getName(), 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.updateProcessInstanceByWorkerGroupName(workerGroup.getName(), "");
+        putMsg(result, Status.SUCCESS);
+        return result;
+    }
+
+    /**
+     * query all worker address list
+     *
+     * @return all worker address list
+     */
+    public Map<String, Object> getWorkerAddressList() {
+        Map<String, Object> result = new HashMap<>();
+        List<String> serverNodeList = zookeeperMonitor.getServerNodeList(ZKNodeType.WORKER, true);
+        result.put(Constants.DATA_LIST, serverNodeList);
+        putMsg(result, Status.SUCCESS);
+        return result;
+    }
+
 }

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/utils/RegexUtils.java

@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.dolphinscheduler.api.utils;
+
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * This is Regex expression utils.
+ */
+public class RegexUtils {
+
+    /**
+     * check number regex expression
+     */
+    private static final String CHECK_NUMBER = "^-?\\d+(\\.\\d+)?$";
+
+    private RegexUtils() {
+    }
+
+    /**
+     * check if the input is number
+     *
+     * @param str input
+     * @return
+     */
+    public static boolean isNumeric(String str) {
+        Pattern pattern = Pattern.compile(CHECK_NUMBER);
+        Matcher isNum = pattern.matcher(str);
+        return isNum.matches();
+    }
+
+    public static String escapeNRT(String str) {
+        // Logging should not be vulnerable to injection attacks: Replace pattern-breaking characters
+        if (str != null && !str.isEmpty()) {
+            return str.replaceAll("[\n|\r|\t]", "_");
+        }
+        return null;
+    }
+
+}

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/utils/ZookeeperMonitor.java

@@ -17,90 +17,89 @@
 package org.apache.dolphinscheduler.api.utils;
 
 import org.apache.dolphinscheduler.common.enums.ZKNodeType;
-import org.apache.dolphinscheduler.common.utils.StringUtils;
 import org.apache.dolphinscheduler.common.model.Server;
+import org.apache.dolphinscheduler.common.utils.StringUtils;
 import org.apache.dolphinscheduler.dao.entity.ZookeeperRecord;
 import org.apache.dolphinscheduler.service.zk.AbstractZKClient;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.stereotype.Component;
 
 import java.util.ArrayList;
 import java.util.Date;
 import java.util.List;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Component;
 
 /**
- *	monitor zookeeper info
+ * monitor zookeeper info
  */
 @Component
 public class ZookeeperMonitor extends AbstractZKClient {
 
-	private static final Logger LOG = LoggerFactory.getLogger(ZookeeperMonitor.class);
-
-	/**
-	 *
-	 * @return zookeeper info list
-	 */
-	public List<ZookeeperRecord> zookeeperInfoList(){
-		String zookeeperServers = getZookeeperQuorum().replaceAll("[\\t\\n\\x0B\\f\\r]", "");
-		try{
-			return zookeeperInfoList(zookeeperServers);
-		}catch(Exception e){
-			LOG.error(e.getMessage(),e);
-		}
-		return null;
-	}
-
-	/**
-	 * get master servers
-	 * @return master server information
-	 */
-	public List<Server> getMasterServers(){
-	    return getServersList(ZKNodeType.MASTER);
-	}
-
-	/**
-	 * master construct is the same with worker, use the master instead
-	 * @return worker server informations
-	 */
-	public List<Server> getWorkerServers(){
-	    return getServersList(ZKNodeType.WORKER);
-	}
-
-	private static List<ZookeeperRecord> zookeeperInfoList(String zookeeperServers) {
-
-		List<ZookeeperRecord> list = new ArrayList<>(5);
-
-		if(StringUtils.isNotBlank(zookeeperServers)){
-			String[] zookeeperServersArray = zookeeperServers.split(",");
-			
-			for (String zookeeperServer : zookeeperServersArray) {
-				ZooKeeperState state = new ZooKeeperState(zookeeperServer);
-				boolean ok = state.ruok();
-				if(ok){
-					state.getZookeeperInfo();
-				}
-				
-				String hostName = zookeeperServer;
-				int connections = state.getConnections();
-				int watches = state.getWatches();
-				long sent = state.getSent();
-				long received = state.getReceived();
-				String mode =  state.getMode();
-				float minLatency =  state.getMinLatency();
-				float avgLatency = state.getAvgLatency();
-				float maxLatency = state.getMaxLatency();
-				int nodeCount = state.getNodeCount();
-				int status = ok ? 1 : 0;
-				Date date = new Date();
-
-				ZookeeperRecord zookeeperRecord = new ZookeeperRecord(hostName,connections,watches,sent,received,mode,minLatency,avgLatency,maxLatency,nodeCount,status,date);
-				list.add(zookeeperRecord);
-
-			}
-		}
-
-		return list;
-	}
+    private static final Logger LOG = LoggerFactory.getLogger(ZookeeperMonitor.class);
+
+    /**
+     *
+     * @return zookeeper info list
+     */
+    public List<ZookeeperRecord> zookeeperInfoList() {
+        String zookeeperServers = getZookeeperQuorum().replaceAll("[\\t\\n\\x0B\\f\\r]", "");
+        try {
+            return zookeeperInfoList(zookeeperServers);
+        } catch (Exception e) {
+            LOG.error(e.getMessage(),e);
+        }
+        return null;
+    }
+
+    /**
+     * get master servers
+     * @return master server information
+     */
+    public List<Server> getMasterServers() {
+        return getServerList(ZKNodeType.MASTER);
+    }
+
+    /**
+     * master construct is the same with worker, use the master instead
+     * @return worker server informations
+     */
+    public List<Server> getWorkerServers() {
+        return getServerList(ZKNodeType.WORKER);
+    }
+
+    private static List<ZookeeperRecord> zookeeperInfoList(String zookeeperServers) {
+
+        List<ZookeeperRecord> list = new ArrayList<>(5);
+
+        if(StringUtils.isNotBlank(zookeeperServers)) {
+            String[] zookeeperServersArray = zookeeperServers.split(",");
+
+            for (String zookeeperServer : zookeeperServersArray) {
+                ZooKeeperState state = new ZooKeeperState(zookeeperServer);
+                boolean ok = state.ruok();
+                if (ok) {
+                    state.getZookeeperInfo();
+                }
+
+                int connections = state.getConnections();
+                int watches = state.getWatches();
+                long sent = state.getSent();
+                long received = state.getReceived();
+                String mode =  state.getMode();
+                float minLatency =  state.getMinLatency();
+                float avgLatency = state.getAvgLatency();
+                float maxLatency = state.getMaxLatency();
+                int nodeCount = state.getNodeCount();
+                int status = ok ? 1 : 0;
+                Date date = new Date();
+
+                ZookeeperRecord zookeeperRecord = new ZookeeperRecord(zookeeperServer,connections,watches,sent,received,mode,minLatency,avgLatency,maxLatency,nodeCount,status,date);
+                list.add(zookeeperRecord);
+
+            }
+        }
+
+        return list;
+    }
 }

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

@@ -16,18 +16,25 @@
  */
 package org.apache.dolphinscheduler.api.service;
 
-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.api.utils.ZookeeperMonitor;
 import org.apache.dolphinscheduler.common.Constants;
 import org.apache.dolphinscheduler.common.enums.UserType;
-import org.apache.dolphinscheduler.common.utils.CollectionUtils;
+import org.apache.dolphinscheduler.common.enums.ZKNodeType;
 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 java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
@@ -35,84 +42,155 @@ import org.junit.runner.RunWith;
 import org.mockito.InjectMocks;
 import org.mockito.Mock;
 import org.mockito.Mockito;
-import org.mockito.internal.matchers.Any;
 import org.mockito.junit.MockitoJUnitRunner;
-import org.slf4j.Logger;
-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 {
 
-    private static final Logger logger = LoggerFactory.getLogger(WorkerGroupServiceTest.class);
-
     @InjectMocks
     private WorkerGroupService workerGroupService;
 
+    @Mock
+    private WorkerGroupMapper workerGroupMapper;
+
     @Mock
     private ProcessInstanceMapper processInstanceMapper;
 
     @Mock
     private ZookeeperCachedOperator zookeeperCachedOperator;
 
+    @Mock
+    private ZookeeperMonitor zookeeperMonitor;
+
+    private String groupName = "groupName000001";
 
     @Before
-    public void init(){
+    public void init() {
         ZookeeperConfig zookeeperConfig = new ZookeeperConfig();
         zookeeperConfig.setDsRoot("/dolphinscheduler_qzw");
         Mockito.when(zookeeperCachedOperator.getZookeeperConfig()).thenReturn(zookeeperConfig);
 
-        String workerPath = zookeeperCachedOperator.getZookeeperConfig().getDsRoot()+"/nodes" +"/worker";
+        String workerPath = zookeeperCachedOperator.getZookeeperConfig().getDsRoot() + Constants.ZOOKEEPER_DOLPHINSCHEDULER_WORKERS;
 
         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");
+        List<String> defaultAddressList = new ArrayList<>();
+        defaultAddressList.add("192.168.220.188:1234");
+        defaultAddressList.add("192.168.220.189:1234");
+
+        Mockito.when(zookeeperCachedOperator.getChildrenKeys(workerPath + "/default")).thenReturn(defaultAddressList);
 
-        Mockito.when(zookeeperCachedOperator.getChildrenKeys(workerPath + "/default")).thenReturn(defaultIpList);
+        Mockito.when(zookeeperCachedOperator.get(workerPath + "/default" + "/" + defaultAddressList.get(0))).thenReturn("0.01,0.17,0.03,25.83,8.0,1.0,2020-07-21 11:17:59,2020-07-21 14:39:20,0,13238");
+    }
+
+    /**
+     *  create or update a worker group
+     */
+    @Test
+    public void testSaveWorkerGroup() {
+        // worker server maps
+        Map<String, String> serverMaps = new HashMap<>();
+        serverMaps.put("127.0.0.1:1234", "0.3,0.07,4.4,7.42,16.0,0.3,2021-03-19 20:17:58,2021-03-19 20:25:29,0,79214");
+        Mockito.when(zookeeperMonitor.getServerMaps(ZKNodeType.WORKER, true)).thenReturn(serverMaps);
+
+        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:1234");
+        Assert.assertEquals(Status.USER_NO_OPERATION_PERM.getMsg(), result.get(Constants.MSG));
 
-        Mockito.when(zookeeperCachedOperator.get(workerPath + "/default" + "/" + defaultIpList.get(0))).thenReturn("0.01,0.17,0.03,25.83,8.0,1.0,2020-07-21 11:17:59,2020-07-21 14:39:20,0,13238");
+        // success
+        user.setUserType(UserType.ADMIN_USER);
+        result = workerGroupService.saveWorkerGroup(user, 0, groupName, "127.0.0.1:1234");
+        Assert.assertEquals(Status.SUCCESS.getMsg(), 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:1234");
+        Assert.assertEquals(Status.NAME_EXIST, result.get(Constants.STATUS));
     }
 
     /**
-     *  query worker group paging
+     * query worker group paging
      */
     @Test
-    public void testQueryAllGroupPaging(){
+    public void testQueryAllGroupPaging() {
         User user = new User();
         // general user add
         user.setUserType(UserType.ADMIN_USER);
         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);
+        Assert.assertEquals(pageInfo.getLists().size(), 1);
     }
 
-
     @Test
-    public void testQueryAllGroup() throws Exception {
+    public void testQueryAllGroup() {
         Map<String, Object> result = workerGroupService.queryAllGroup();
-        Set<String> workerGroups = (Set<String>) result.get(Constants.DATA_LIST);
+        List<String> workerGroups = (List<String>) result.get(Constants.DATA_LIST);
         Assert.assertEquals(workerGroups.size(), 1);
     }
 
+    /**
+     * delete group by id
+     */
+    @Test
+    public  void testDeleteWorkerGroupById() {
+        User user = new User();
+        user.setUserType(UserType.ADMIN_USER);
+        WorkerGroup wg2 = getWorkerGroup(2);
+        Mockito.when(workerGroupMapper.selectById(2)).thenReturn(wg2);
+        Mockito.when(processInstanceMapper.queryByWorkerGroupNameAndStatus(wg2.getName(), Constants.NOT_TERMINATED_STATES)).thenReturn(getProcessInstanceList());
+        Map<String, Object> result = workerGroupService.deleteWorkerGroupById(user, 1);
+        Assert.assertEquals(Status.DELETE_WORKER_GROUP_NOT_EXIST.getCode(), ((Status) result.get(Constants.STATUS)).getCode());
+        result = workerGroupService.deleteWorkerGroupById(user, 2);
+        Assert.assertEquals(Status.DELETE_WORKER_GROUP_BY_ID_FAIL.getCode(), ((Status) result.get(Constants.STATUS)).getCode());
+        // correct
+        WorkerGroup wg3 = getWorkerGroup(3);
+        Mockito.when(workerGroupMapper.selectById(3)).thenReturn(wg3);
+        Mockito.when(processInstanceMapper.queryByWorkerGroupNameAndStatus(wg3.getName(), Constants.NOT_TERMINATED_STATES)).thenReturn(new ArrayList<>());
+        result = workerGroupService.deleteWorkerGroupById(user, 3);
+        Assert.assertEquals(Status.SUCCESS.getMsg(), result.get(Constants.MSG));
+    }
 
     /**
      * get processInstances
-     * @return
      */
-    private List<ProcessInstance> getProcessInstanceList(){
-
+    private List<ProcessInstance> getProcessInstanceList() {
         List<ProcessInstance> processInstances = new ArrayList<>();
         processInstances.add(new ProcessInstance());
         return processInstances;
     }
 
+    @Test
+    public void testQueryAllGroupWithDefault() {
+        Map<String, Object> result = workerGroupService.queryAllGroup();
+        List<String> workerGroups = (List<String>) result.get(Constants.DATA_LIST);
+        Assert.assertEquals(1, workerGroups.size());
+        Assert.assertEquals("default", workerGroups.toArray()[0]);
+    }
+
+    /**
+     * 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/java/org/apache/dolphinscheduler/common/Constants.java

@@ -760,8 +760,8 @@ public final class Constants {
     /**
      * master/worker server use for zk
      */
-    public static final String MASTER_PREFIX = "master";
-    public static final String WORKER_PREFIX = "worker";
+    public static final String MASTER_TYPE = "master";
+    public static final String WORKER_TYPE = "worker";
     public static final String DELETE_ZK_OP = "delete";
     public static final String ADD_ZK_OP = "add";
     public static final String ALIAS = "alias";

dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/ZKNodeType.java

@@ -25,7 +25,6 @@ public enum ZKNodeType {
      * 0 master node;
      * 1 worker node;
      * 2 dead_server node;
-     * 3 task_queue node;
      */
-    MASTER, WORKER, DEAD_SERVER, TASK_QUEUE;
+    MASTER, WORKER, DEAD_SERVER;
 }

dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/ResInfo.java

@@ -112,4 +112,17 @@ public class ResInfo {
         return masterServer;
     }
 
+    /**
+     * is valid heartbeat info for zk
+     * @param heartBeatInfo heartbeat info
+     * @return heartbeat info is valid
+     */
+    public static boolean isValidHeartbeatForZKInfo(String heartBeatInfo) {
+        if (StringUtils.isNotEmpty(heartBeatInfo)) {
+            String[] parts = heartBeatInfo.split(Constants.COMMA);
+            return parts.length == Constants.HEARTBEAT_FOR_ZOOKEEPER_INFO_LENGTH;
+        }
+        return false;
+    }
+
 }

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

@@ -16,23 +16,59 @@
  */
 package org.apache.dolphinscheduler.dao.entity;
 
-
 import java.util.Date;
-import java.util.List;
+
+import com.baomidou.mybatisplus.annotation.IdType;
+import com.baomidou.mybatisplus.annotation.TableField;
+import com.baomidou.mybatisplus.annotation.TableId;
+import com.baomidou.mybatisplus.annotation.TableName;
+import com.fasterxml.jackson.annotation.JsonFormat;
 
 /**
  * worker group
  */
+@TableName("t_ds_worker_group")
 public class WorkerGroup {
 
+    @TableId(value = "id", type = IdType.AUTO)
+    private int id;
+
     private String name;
 
-    private List<String> ipList;
+    private String addrList;
 
+    @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss", timezone = "GMT+8")
     private Date createTime;
 
+    @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss", timezone = "GMT+8")
     private Date updateTime;
 
+    @TableField(exist = false)
+    private boolean systemDefault;
+
+    public int getId() {
+        return id;
+    }
+
+    public void setId(int id) {
+        this.id = id;
+    }
+
+    public String getName() {
+        return name;
+    }
+
+    public void setName(String name) {
+        this.name = name;
+    }
+
+    public String getAddrList() {
+        return addrList;
+    }
+
+    public void setAddrList(String addrList) {
+        this.addrList = addrList;
+    }
 
     public Date getCreateTime() {
         return createTime;
@@ -50,21 +86,24 @@ public class WorkerGroup {
         this.updateTime = updateTime;
     }
 
-    public String getName() {
-        return name;
-    }
-
-    public void setName(String name) {
-        this.name = name;
+    public boolean getSystemDefault() {
+        return systemDefault;
     }
 
-    public List<String> getIpList() {
-        return ipList;
+    public void setSystemDefault(boolean systemDefault) {
+        this.systemDefault = systemDefault;
     }
 
-    public void setIpList(List<String> ipList) {
-        this.ipList = ipList;
+    @Override
+    public String toString() {
+        return "WorkerGroup{"
+                + "id= " + id
+                + ", name= " + name
+                + ", addrList= " + addrList
+                + ", createTime= " + createTime
+                + ", updateTime= " + updateTime
+                + ", systemDefault= " + systemDefault
+                + "}";
     }
 
-
 }

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

@@ -16,17 +16,19 @@
  */
 package org.apache.dolphinscheduler.dao.mapper;
 
+import org.apache.dolphinscheduler.common.enums.ExecutionStatus;
 import org.apache.dolphinscheduler.dao.entity.ExecuteStatusCount;
 import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
-import com.baomidou.mybatisplus.core.mapper.BaseMapper;
-import com.baomidou.mybatisplus.core.metadata.IPage;
-import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
-import org.apache.dolphinscheduler.common.enums.ExecutionStatus;
+
 import org.apache.ibatis.annotations.Param;
 
 import java.util.Date;
 import java.util.List;
 
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import com.baomidou.mybatisplus.core.metadata.IPage;
+import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
+
 /**
  * process instance mapper interface
  */
@@ -59,12 +61,12 @@ public interface ProcessInstanceMapper extends BaseMapper<ProcessInstance> {
 
     /**
      * query process instance by worker group and stateArray
-     * @param workerGroupId workerGroupId
+     * @param workerGroupName workerGroupName
      * @param states states array
      * @return process instance list
      */
-    List<ProcessInstance> queryByWorkerGroupIdAndStatus(@Param("workerGroupId") int workerGroupId,
-                                                   @Param("states") int[] states);
+    List<ProcessInstance> queryByWorkerGroupNameAndStatus(@Param("workerGroupName") String workerGroupName,
+                                                          @Param("states") int[] states);
 
     /**
      * process instance page
@@ -130,12 +132,13 @@ public interface ProcessInstanceMapper extends BaseMapper<ProcessInstance> {
                                         @Param("destTenantId") int destTenantId);
 
     /**
-     * update process instance by worker groupId
-     * @param originWorkerGroupId originWorkerGroupId
-     * @param destWorkerGroupId destWorkerGroupId
+     * update process instance by worker group name
+     * @param originWorkerGroupName originWorkerGroupName
+     * @param destWorkerGroupName destWorkerGroupName
      * @return update result
      */
-    int updateProcessInstanceByWorkerGroupId(@Param("originWorkerGroupId") int originWorkerGroupId, @Param("destWorkerGroupId") int destWorkerGroupId);
+    int updateProcessInstanceByWorkerGroupName(@Param("originWorkerGroupName") String originWorkerGroupName,
+                                               @Param("destWorkerGroupName") String destWorkerGroupName);
 
     /**
      * count process instance state by user

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

@@ -0,0 +1,45 @@
+/*
+ * 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 org.apache.ibatis.annotations.Param;
+
+import java.util.List;
+
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+
+/**
+ * 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);
+
+}

dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/ProcessInstanceMapper.xml

@@ -51,12 +51,12 @@
         order by id asc
     </select>
 
-    <select id="queryByWorkerGroupIdAndStatus" resultType="org.apache.dolphinscheduler.dao.entity.ProcessInstance">
+    <select id="queryByWorkerGroupNameAndStatus" resultType="org.apache.dolphinscheduler.dao.entity.ProcessInstance">
         select *
         from t_ds_process_instance
         where 1=1
-        <if test="workerGroupId != -1">
-            and worker_group_id =#{workerGroupId}
+        <if test="workerGroupName != ''">
+            and worker_group =#{workerGroupName}
         </if>
         and state in
         <foreach collection="states" item="i" open="(" close=")" separator=",">
@@ -116,10 +116,10 @@
         where tenant_id = #{originTenantId}
     </update>
 
-    <update id="updateProcessInstanceByWorkerGroupId">
+    <update id="updateProcessInstanceByWorkerGroupName">
         update t_ds_process_instance
-        set worker_group_id = #{destWorkerGroupId}
-        where worker_group_id = #{originWorkerGroupId}
+        set worker_group = #{destWorkerGroupName}
+        where worker_group = #{originWorkerGroupName}
     </update>
 
     <select id="countInstanceStateByUser" resultType="org.apache.dolphinscheduler.dao.entity.ExecuteStatusCount">

dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/WorkerGroupMapper.xml

@@ -0,0 +1,31 @@
+<?xml version="1.0" encoding="UTF-8" ?>
+<!--
+  ~ 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.
+  -->
+
+<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd" >
+<mapper namespace="org.apache.dolphinscheduler.dao.mapper.WorkerGroupMapper">
+    <select id="queryAllWorkerGroup" resultType="org.apache.dolphinscheduler.dao.entity.WorkerGroup">
+        select *
+        from t_ds_worker_group
+        order by update_time desc
+    </select>
+    <select id="queryWorkerGroupByName" resultType="org.apache.dolphinscheduler.dao.entity.WorkerGroup">
+        select *
+        from t_ds_worker_group
+        where name = #{name}
+    </select>
+</mapper>

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/MasterServer.java

@@ -28,7 +28,6 @@ import org.apache.dolphinscheduler.server.master.processor.TaskAckProcessor;
 import org.apache.dolphinscheduler.server.master.processor.TaskKillResponseProcessor;
 import org.apache.dolphinscheduler.server.master.processor.TaskResponseProcessor;
 import org.apache.dolphinscheduler.server.master.runner.MasterSchedulerService;
-import org.apache.dolphinscheduler.server.worker.WorkerServer;
 import org.apache.dolphinscheduler.server.zk.ZKMasterClient;
 import org.apache.dolphinscheduler.service.bean.SpringApplicationContext;
 import org.apache.dolphinscheduler.service.quartz.QuartzExecutors;
@@ -44,8 +43,15 @@ import org.springframework.boot.builder.SpringApplicationBuilder;
 import org.springframework.context.annotation.ComponentScan;
 import org.springframework.context.annotation.FilterType;
 
+/**
+ *  master server
+ */
 @ComponentScan(value = "org.apache.dolphinscheduler", excludeFilters = {
-        @ComponentScan.Filter(type = FilterType.ASSIGNABLE_TYPE, classes = {WorkerServer.class})
+        @ComponentScan.Filter(type = FilterType.REGEX, pattern = {
+                "org.apache.dolphinscheduler.server.worker.*",
+                "org.apache.dolphinscheduler.server.monitor.*",
+                "org.apache.dolphinscheduler.server.log.*"
+        })
 })
 public class MasterServer implements IStoppable {
 

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/dispatch/executor/NettyExecutorManager.java

@@ -17,7 +17,6 @@
 
 package org.apache.dolphinscheduler.server.master.dispatch.executor;
 
-import org.apache.commons.collections.CollectionUtils;
 import org.apache.dolphinscheduler.remote.NettyRemotingClient;
 import org.apache.dolphinscheduler.remote.command.Command;
 import org.apache.dolphinscheduler.remote.command.CommandType;
@@ -29,18 +28,22 @@ import org.apache.dolphinscheduler.server.master.dispatch.exceptions.ExecuteExce
 import org.apache.dolphinscheduler.server.master.processor.TaskAckProcessor;
 import org.apache.dolphinscheduler.server.master.processor.TaskKillResponseProcessor;
 import org.apache.dolphinscheduler.server.master.processor.TaskResponseProcessor;
-import org.apache.dolphinscheduler.server.registry.ZookeeperNodeManager;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.stereotype.Service;
+import org.apache.dolphinscheduler.server.registry.ServerNodeManager;
+
+import org.apache.commons.collections.CollectionUtils;
 
-import javax.annotation.PostConstruct;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.Set;
 
+import javax.annotation.PostConstruct;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
 /**
  *  netty executor manager
  */
@@ -50,10 +53,10 @@ public class NettyExecutorManager extends AbstractExecutorManager<Boolean>{
     private final Logger logger = LoggerFactory.getLogger(NettyExecutorManager.class);
 
     /**
-     * zookeeper node manager
+     * server node manager
      */
     @Autowired
-    private ZookeeperNodeManager zookeeperNodeManager;
+    private ServerNodeManager serverNodeManager;
 
     /**
      * netty remote client
@@ -183,7 +186,7 @@ public class NettyExecutorManager extends AbstractExecutorManager<Boolean>{
         ExecutorType executorType = context.getExecutorType();
         switch (executorType){
             case WORKER:
-                nodes = zookeeperNodeManager.getWorkerGroupNodes(context.getWorkerGroup());
+                nodes = serverNodeManager.getWorkerGroupNodes(context.getWorkerGroup());
                 break;
             case CLIENT:
                 break;

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/dispatch/host/CommonHostManager.java

@@ -21,28 +21,25 @@ import org.apache.dolphinscheduler.common.utils.CollectionUtils;
 import org.apache.dolphinscheduler.remote.utils.Host;
 import org.apache.dolphinscheduler.server.master.dispatch.context.ExecutionContext;
 import org.apache.dolphinscheduler.server.master.dispatch.enums.ExecutorType;
-import org.apache.dolphinscheduler.server.registry.ZookeeperNodeManager;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.beans.factory.annotation.Autowired;
+import org.apache.dolphinscheduler.server.registry.ServerNodeManager;
 
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
+import java.util.Set;
 
+import org.springframework.beans.factory.annotation.Autowired;
 
 /**
- *  round robin host manager
+ *  common host manager
  */
 public abstract class CommonHostManager implements HostManager {
 
-    private final Logger logger = LoggerFactory.getLogger(CommonHostManager.class);
-
     /**
-     * zookeeperNodeManager
+     * server node manager
      */
     @Autowired
-    protected ZookeeperNodeManager zookeeperNodeManager;
+    protected ServerNodeManager serverNodeManager;
 
     /**
      * select host
@@ -50,39 +47,37 @@ public abstract class CommonHostManager implements HostManager {
      * @return host
      */
     @Override
-    public Host select(ExecutionContext context){
-        Host host = new Host();
-        Collection<String> nodes = null;
-        /**
-         * executor type
-         */
+    public Host select(ExecutionContext context) {
+        List<Host> candidates = null;
+        String workerGroup = context.getWorkerGroup();
         ExecutorType executorType = context.getExecutorType();
-        switch (executorType){
+        switch (executorType) {
             case WORKER:
-                nodes = zookeeperNodeManager.getWorkerGroupNodes(context.getWorkerGroup());
+                candidates = getWorkerCandidates(workerGroup);
                 break;
             case CLIENT:
                 break;
             default:
                 throw new IllegalArgumentException("invalid executorType : " + executorType);
-
         }
-        if(CollectionUtils.isEmpty(nodes)){
-            return host;
-        }
-        List<Host> candidateHosts = new ArrayList<>(nodes.size());
-        nodes.stream().forEach(node -> candidateHosts.add(Host.of(node)));
 
-        return select(candidateHosts);
+        if (CollectionUtils.isEmpty(candidates)) {
+            return new Host();
+        }
+        return select(candidates);
     }
 
     protected abstract Host select(Collection<Host> nodes);
 
-    public void setZookeeperNodeManager(ZookeeperNodeManager zookeeperNodeManager) {
-        this.zookeeperNodeManager = zookeeperNodeManager;
+    protected List<Host> getWorkerCandidates(String workerGroup) {
+        List<Host> hosts = new ArrayList<>();
+        Set<String> nodes = serverNodeManager.getWorkerGroupNodes(workerGroup);
+        if (CollectionUtils.isNotEmpty(nodes)) {
+            for (String node : nodes) {
+                hosts.add(Host.of(node));
+            }
+        }
+        return hosts;
     }
 
-    public ZookeeperNodeManager getZookeeperNodeManager() {
-        return zookeeperNodeManager;
-    }
 }

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/dispatch/host/HostManagerConfig.java

@@ -45,7 +45,7 @@ public class HostManagerConfig {
         String hostSelector = masterConfig.getHostSelector();
         HostSelector selector = HostSelector.of(hostSelector);
         HostManager hostManager;
-        switch (selector){
+        switch (selector) {
             case RANDOM:
                 hostManager = new RandomHostManager();
                 break;

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/dispatch/host/LowerWeightHostManager.java

@@ -19,20 +19,18 @@ package org.apache.dolphinscheduler.server.master.dispatch.host;
 
 import org.apache.dolphinscheduler.common.Constants;
 import org.apache.dolphinscheduler.common.utils.CollectionUtils;
-import org.apache.dolphinscheduler.common.utils.StringUtils;
+import org.apache.dolphinscheduler.common.utils.ResInfo;
 import org.apache.dolphinscheduler.remote.utils.Host;
 import org.apache.dolphinscheduler.remote.utils.NamedThreadFactory;
 import org.apache.dolphinscheduler.server.master.dispatch.context.ExecutionContext;
 import org.apache.dolphinscheduler.server.master.dispatch.host.assign.HostWeight;
 import org.apache.dolphinscheduler.server.master.dispatch.host.assign.LowerWeightRoundRobin;
-import org.apache.dolphinscheduler.server.registry.ZookeeperRegistryCenter;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.beans.factory.annotation.Autowired;
 
-import javax.annotation.PostConstruct;
-import javax.annotation.PreDestroy;
-import java.util.*;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
@@ -40,27 +38,19 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 
-import static org.apache.dolphinscheduler.common.Constants.COMMA;
+import javax.annotation.PostConstruct;
+import javax.annotation.PreDestroy;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
- *  round robin host manager
+ *  lower weight host manager
  */
 public class LowerWeightHostManager extends CommonHostManager {
 
     private final Logger logger = LoggerFactory.getLogger(LowerWeightHostManager.class);
 
-    /**
-     * zookeeper registry center
-     */
-    @Autowired
-    private ZookeeperRegistryCenter registryCenter;
-
-    /**
-     * round robin host manager
-     */
-    private RoundRobinHostManager roundRobinHostManager;
-
     /**
      * selector
      */
@@ -82,18 +72,16 @@ public class LowerWeightHostManager extends CommonHostManager {
     private ScheduledExecutorService executorService;
 
     @PostConstruct
-    public void init(){
+    public void init() {
         this.selector = new LowerWeightRoundRobin();
         this.workerHostWeightsMap = new ConcurrentHashMap<>();
         this.lock = new ReentrantLock();
         this.executorService = Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory("LowerWeightHostManagerExecutor"));
         this.executorService.scheduleWithFixedDelay(new RefreshResourceTask(),0, 5, TimeUnit.SECONDS);
-        this.roundRobinHostManager = new RoundRobinHostManager();
-        this.roundRobinHostManager.setZookeeperNodeManager(getZookeeperNodeManager());
     }
 
     @PreDestroy
-    public void close(){
+    public void close() {
         this.executorService.shutdownNow();
     }
 
@@ -103,9 +91,9 @@ public class LowerWeightHostManager extends CommonHostManager {
      * @return host
      */
     @Override
-    public Host select(ExecutionContext context){
+    public Host select(ExecutionContext context) {
         Set<HostWeight> workerHostWeights = getWorkerHostWeights(context.getWorkerGroup());
-        if(CollectionUtils.isNotEmpty(workerHostWeights)){
+        if (CollectionUtils.isNotEmpty(workerHostWeights)) {
             return selector.select(workerHostWeights).getHost();
         }
         return new Host();
@@ -116,7 +104,7 @@ public class LowerWeightHostManager extends CommonHostManager {
         throw new UnsupportedOperationException("not support");
     }
 
-    private void syncWorkerHostWeight(Map<String, Set<HostWeight>> workerHostWeights){
+    private void syncWorkerHostWeight(Map<String, Set<HostWeight>> workerHostWeights) {
         lock.lock();
         try {
             workerHostWeightsMap.clear();
@@ -126,7 +114,7 @@ public class LowerWeightHostManager extends CommonHostManager {
         }
     }
 
-    private Set<HostWeight> getWorkerHostWeights(String workerGroup){
+    private Set<HostWeight> getWorkerHostWeights(String workerGroup) {
         lock.lock();
         try {
             return workerHostWeightsMap.get(workerGroup);
@@ -135,46 +123,50 @@ public class LowerWeightHostManager extends CommonHostManager {
         }
     }
 
-    class RefreshResourceTask implements Runnable{
+    class RefreshResourceTask implements Runnable {
 
         @Override
         public void run() {
             try {
-                Map<String, Set<String>> workerGroupNodes = zookeeperNodeManager.getWorkerGroupNodes();
-                Set<Map.Entry<String, Set<String>>> entries = workerGroupNodes.entrySet();
                 Map<String, Set<HostWeight>> workerHostWeights = new HashMap<>();
-                for(Map.Entry<String, Set<String>> entry : entries){
+                Map<String, Set<String>> workerGroupNodes = serverNodeManager.getWorkerGroupNodes();
+                for (Map.Entry<String, Set<String>> entry : workerGroupNodes.entrySet()) {
                     String workerGroup = entry.getKey();
                     Set<String> nodes = entry.getValue();
-                    String workerGroupPath = registryCenter.getWorkerGroupPath(workerGroup);
                     Set<HostWeight> hostWeights = new HashSet<>(nodes.size());
-                    for(String node : nodes){
-                        String heartbeat = registryCenter.getRegisterOperator().get(workerGroupPath + "/" + node);
-                        if(StringUtils.isNotEmpty(heartbeat)
-                                && heartbeat.split(COMMA).length == Constants.HEARTBEAT_FOR_ZOOKEEPER_INFO_LENGTH){
-                            String[] parts = heartbeat.split(COMMA);
-
-                            int status = Integer.parseInt(parts[8]);
-                            if (status == Constants.ABNORMAL_NODE_STATUS){
-                                logger.warn("load is too high or availablePhysicalMemorySize(G) is too low, it's availablePhysicalMemorySize(G):{},loadAvg:{}",
-                                        Double.parseDouble(parts[3]) , Double.parseDouble(parts[2]));
-                                continue;
-                            }
-
-                            double cpu = Double.parseDouble(parts[0]);
-                            double memory = Double.parseDouble(parts[1]);
-                            double loadAverage = Double.parseDouble(parts[2]);
-                            HostWeight hostWeight = new HostWeight(Host.of(node), cpu, memory, loadAverage);
+                    for (String node : nodes) {
+                        String heartbeat = serverNodeManager.getWorkerNodeInfo(node);
+                        HostWeight hostWeight = getHostWeight(node, heartbeat);
+                        if (hostWeight != null) {
                             hostWeights.add(hostWeight);
                         }
                     }
-                    workerHostWeights.put(workerGroup, hostWeights);
+                    if (!hostWeights.isEmpty()) {
+                        workerHostWeights.put(workerGroup, hostWeights);
+                    }
                 }
                 syncWorkerHostWeight(workerHostWeights);
-            } catch (Throwable ex){
+            } catch (Throwable ex) {
                 logger.error("RefreshResourceTask error", ex);
             }
         }
+
+        public HostWeight getHostWeight(String addr, String heartbeat) {
+            if (ResInfo.isValidHeartbeatForZKInfo(heartbeat)) {
+                String[] parts = heartbeat.split(Constants.COMMA);
+                int status = Integer.parseInt(parts[8]);
+                if (status == Constants.ABNORMAL_NODE_STATUS) {
+                    logger.warn("load is too high or availablePhysicalMemorySize(G) is too low, it's availablePhysicalMemorySize(G):{},loadAvg:{}",
+                            Double.parseDouble(parts[3]), Double.parseDouble(parts[2]));
+                    return null;
+                }
+                double cpu = Double.parseDouble(parts[0]);
+                double memory = Double.parseDouble(parts[1]);
+                double loadAverage = Double.parseDouble(parts[2]);
+                return new HostWeight(Host.of(addr), cpu, memory, loadAverage);
+            }
+            return null;
+        }
     }
 
 }

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/dispatch/host/RandomHostManager.java

@@ -23,9 +23,8 @@ import org.apache.dolphinscheduler.server.master.dispatch.host.assign.Selector;
 
 import java.util.Collection;
 
-
 /**
- *  round robin host manager
+ *  random host manager
  */
 public class RandomHostManager extends CommonHostManager {
 

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/dispatch/host/RoundRobinHostManager.java

@@ -23,7 +23,6 @@ import org.apache.dolphinscheduler.server.master.dispatch.host.assign.Selector;
 
 import java.util.Collection;
 
-
 /**
  *  round robin host manager
  */
@@ -37,7 +36,7 @@ public class RoundRobinHostManager extends CommonHostManager {
     /**
      * set round robin
      */
-    public RoundRobinHostManager(){
+    public RoundRobinHostManager() {
         this.selector = new RoundRobinSelector<>();
     }
 

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/registry/MasterRegistry.java

@@ -81,7 +81,7 @@ public class MasterRegistry {
      * registry
      */
     public void registry() {
-        String address = NetUtils.getHost();
+        String address = NetUtils.getAddr(masterConfig.getListenPort());
         String localNodePath = getMasterPath();
         zookeeperRegistryCenter.getRegisterOperator().persistEphemeral(localNodePath, "");
         zookeeperRegistryCenter.getRegisterOperator().getZkClient().getConnectionStateListenable().addListener(
@@ -98,10 +98,10 @@ public class MasterRegistry {
             });
         int masterHeartbeatInterval = masterConfig.getMasterHeartbeatInterval();
         HeartBeatTask heartBeatTask = new HeartBeatTask(startTime,
-                masterConfig.getMasterReservedMemory(),
                 masterConfig.getMasterMaxCpuloadAvg(),
+                masterConfig.getMasterReservedMemory(),
                 Sets.newHashSet(getMasterPath()),
-                Constants.MASTER_PREFIX,
+                Constants.MASTER_TYPE,
                 zookeeperRegistryCenter);
 
         this.heartBeatExecutor.scheduleAtFixedRate(heartBeatTask, masterHeartbeatInterval, masterHeartbeatInterval, TimeUnit.SECONDS);

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/registry/HeartBeatTask.java

@@ -48,17 +48,17 @@ public class HeartBeatTask implements Runnable {
     protected IStoppable stoppable = null;
 
     public HeartBeatTask(String startTime,
-                         double reservedMemory,
                          double maxCpuloadAvg,
+                         double reservedMemory,
                          Set<String> heartBeatPaths,
                          String serverType,
                          ZookeeperRegistryCenter zookeeperRegistryCenter) {
         this.startTime = startTime;
-        this.reservedMemory = reservedMemory;
         this.maxCpuloadAvg = maxCpuloadAvg;
+        this.reservedMemory = reservedMemory;
         this.heartBeatPaths = heartBeatPaths;
-        this.zookeeperRegistryCenter = zookeeperRegistryCenter;
         this.serverType = serverType;
+        this.zookeeperRegistryCenter = zookeeperRegistryCenter;
     }
 
     @Override

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/registry/ServerNodeManager.java

@@ -17,63 +17,103 @@
 
 package org.apache.dolphinscheduler.server.registry;
 
-import org.apache.commons.collections.CollectionUtils;
-import org.apache.curator.framework.CuratorFramework;
-
-import org.apache.curator.framework.recipes.cache.TreeCacheEvent;
+import org.apache.dolphinscheduler.common.Constants;
+import org.apache.dolphinscheduler.common.enums.ZKNodeType;
 import org.apache.dolphinscheduler.common.utils.StringUtils;
 import org.apache.dolphinscheduler.dao.AlertDao;
+import org.apache.dolphinscheduler.dao.entity.WorkerGroup;
+import org.apache.dolphinscheduler.dao.mapper.WorkerGroupMapper;
+import org.apache.dolphinscheduler.remote.utils.NamedThreadFactory;
 import org.apache.dolphinscheduler.service.zk.AbstractListener;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.beans.factory.InitializingBean;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.stereotype.Service;
+import org.apache.dolphinscheduler.service.zk.AbstractZKClient;
+
+import org.apache.commons.collections.CollectionUtils;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.recipes.cache.TreeCacheEvent;
 
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.HashSet;
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 
-import static org.apache.dolphinscheduler.common.Constants.DEFAULT_WORKER_GROUP;
+import javax.annotation.PreDestroy;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+import org.springframework.stereotype.Service;
 
 /**
- *  zookeeper node manager
+ *  server node manager
  */
 @Service
-public class ZookeeperNodeManager implements InitializingBean {
+public class ServerNodeManager implements InitializingBean {
 
-    private final Logger logger = LoggerFactory.getLogger(ZookeeperNodeManager.class);
+    private final Logger logger = LoggerFactory.getLogger(ServerNodeManager.class);
 
     /**
-     *  master lock
+     * master lock
      */
     private final Lock masterLock = new ReentrantLock();
 
     /**
-     *  worker group lock
+     * worker group lock
      */
     private final Lock workerGroupLock = new ReentrantLock();
 
     /**
-     *  worker group nodes
+     * worker node info lock
+     */
+    private final Lock workerNodeInfoLock = new ReentrantLock();
+
+    /**
+     * worker group nodes
      */
     private final ConcurrentHashMap<String, Set<String>> workerGroupNodes = new ConcurrentHashMap<>();
 
     /**
-     *  master nodes
+     * master nodes
      */
     private final Set<String> masterNodes = new HashSet<>();
 
+    /**
+     * worker node info
+     */
+    private final Map<String, String> workerNodeInfo = new HashMap<>();
+
+    /**
+     * executor service
+     */
+    private ScheduledExecutorService executorService;
+
+    /**
+     * zk client
+     */
+    @Autowired
+    private ZKClient zkClient;
+
     /**
      * zookeeper registry center
      */
     @Autowired
     private ZookeeperRegistryCenter registryCenter;
 
+    /**
+     * worker group mapper
+     */
+    @Autowired
+    private WorkerGroupMapper workerGroupMapper;
+
     /**
      * alert dao
      */
@@ -87,9 +127,14 @@ public class ZookeeperNodeManager implements InitializingBean {
     @Override
     public void afterPropertiesSet() throws Exception {
         /**
-         *  load nodes from zookeeper
+         * load nodes from zookeeper
          */
         load();
+        /**
+         * init executor service
+         */
+        executorService = Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory("ServerNodeManagerExecutor"));
+        executorService.scheduleWithFixedDelay(new WorkerNodeInfoAndGroupDbSyncTask(), 0, 10, TimeUnit.SECONDS);
         /**
          * init MasterNodeListener listener
          */
@@ -103,22 +148,59 @@ public class ZookeeperNodeManager implements InitializingBean {
     /**
      *  load nodes from zookeeper
      */
-    private void load(){
+    private void load() {
         /**
          * master nodes from zookeeper
          */
-        Set<String> masterNodes = registryCenter.getMasterNodesDirectly();
-        syncMasterNodes(masterNodes);
+        Set<String> initMasterNodes = registryCenter.getMasterNodesDirectly();
+        syncMasterNodes(initMasterNodes);
 
         /**
          * worker group nodes from zookeeper
          */
         Set<String> workerGroups = registryCenter.getWorkerGroupDirectly();
-        for(String workerGroup : workerGroups){
+        for (String workerGroup : workerGroups) {
             syncWorkerGroupNodes(workerGroup, registryCenter.getWorkerGroupNodesDirectly(workerGroup));
         }
     }
 
+    /**
+     * zookeeper client
+     */
+    @Component
+    static class ZKClient extends AbstractZKClient {}
+
+    /**
+     *  worker node info and worker group db sync task
+     */
+    class WorkerNodeInfoAndGroupDbSyncTask implements Runnable {
+
+        @Override
+        public void run() {
+            // sync worker node info
+            Map<String, String> newWorkerNodeInfo = zkClient.getServerMaps(ZKNodeType.WORKER, true);
+            syncWorkerNodeInfo(newWorkerNodeInfo);
+
+            // sync worker group nodes from database
+            List<WorkerGroup> workerGroupList = workerGroupMapper.queryAllWorkerGroup();
+            if (CollectionUtils.isNotEmpty(workerGroupList)) {
+                for (WorkerGroup wg : workerGroupList) {
+                    String workerGroup = wg.getName();
+                    Set<String> nodes = new HashSet<>();
+                    String[] addrs = wg.getAddrList().split(Constants.COMMA);
+                    for (String addr : addrs) {
+                        if (newWorkerNodeInfo.containsKey(addr)) {
+                            nodes.add(addr);
+                        }
+                    }
+                    if (!nodes.isEmpty()) {
+                        syncWorkerGroupNodes(workerGroup, nodes);
+                    }
+                }
+            }
+        }
+    }
+
     /**
      *  worker group node listener
      */
@@ -126,44 +208,38 @@ public class ZookeeperNodeManager implements InitializingBean {
 
         @Override
         protected void dataChanged(CuratorFramework client, TreeCacheEvent event, String path) {
-            if(registryCenter.isWorkerPath(path)){
+            if (registryCenter.isWorkerPath(path)) {
                 try {
                     if (event.getType() == TreeCacheEvent.Type.NODE_ADDED) {
                         logger.info("worker group node : {} added.", path);
                         String group = parseGroup(path);
-                        Set<String> workerNodes = workerGroupNodes.getOrDefault(group, new HashSet<>());
-                        Set<String> previousNodes = new HashSet<>(workerNodes);
                         Set<String> currentNodes = registryCenter.getWorkerGroupNodesDirectly(group);
                         logger.info("currentNodes : {}", currentNodes);
                         syncWorkerGroupNodes(group, currentNodes);
                     } else if (event.getType() == TreeCacheEvent.Type.NODE_REMOVED) {
                         logger.info("worker group node : {} down.", path);
                         String group = parseGroup(path);
-                        Set<String> workerNodes = workerGroupNodes.getOrDefault(group, new HashSet<>());
-                        Set<String> previousNodes = new HashSet<>(workerNodes);
                         Set<String> currentNodes = registryCenter.getWorkerGroupNodesDirectly(group);
                         syncWorkerGroupNodes(group, currentNodes);
                         alertDao.sendServerStopedAlert(1, path, "WORKER");
                     }
-                } catch (IllegalArgumentException ignore) {
-                    logger.warn(ignore.getMessage());
+                } catch (IllegalArgumentException ex) {
+                    logger.warn(ex.getMessage());
                 } catch (Exception ex) {
                     logger.error("WorkerGroupListener capture data change and get data failed", ex);
                 }
             }
         }
 
-        private String parseGroup(String path){
-            String[] parts = path.split("\\/");
-            if(parts.length < 6){
+        private String parseGroup(String path) {
+            String[] parts = path.split("/");
+            if (parts.length < 6) {
                 throw new IllegalArgumentException(String.format("worker group path : %s is not valid, ignore", path));
             }
-            String group = parts[parts.length - 2];
-            return group;
+            return parts[parts.length - 2];
         }
     }
 
-
     /**
      *  master node listener
      */
@@ -175,12 +251,10 @@ public class ZookeeperNodeManager implements InitializingBean {
                 try {
                     if (event.getType() == TreeCacheEvent.Type.NODE_ADDED) {
                         logger.info("master node : {} added.", path);
-                        Set<String> previousNodes = new HashSet<>(masterNodes);
                         Set<String> currentNodes = registryCenter.getMasterNodesDirectly();
                         syncMasterNodes(currentNodes);
                     } else if (event.getType() == TreeCacheEvent.Type.NODE_REMOVED) {
                         logger.info("master node : {} down.", path);
-                        Set<String> previousNodes = new HashSet<>(masterNodes);
                         Set<String> currentNodes = registryCenter.getMasterNodesDirectly();
                         syncMasterNodes(currentNodes);
                         alertDao.sendServerStopedAlert(1, path, "MASTER");
@@ -209,7 +283,7 @@ public class ZookeeperNodeManager implements InitializingBean {
      *  sync master nodes
      * @param nodes master nodes
      */
-    private void syncMasterNodes(Set<String> nodes){
+    private void syncMasterNodes(Set<String> nodes) {
         masterLock.lock();
         try {
             masterNodes.clear();
@@ -224,7 +298,7 @@ public class ZookeeperNodeManager implements InitializingBean {
      * @param workerGroup worker group
      * @param nodes worker nodes
      */
-    private void syncWorkerGroupNodes(String workerGroup, Set<String> nodes){
+    private void syncWorkerGroupNodes(String workerGroup, Set<String> nodes) {
         workerGroupLock.lock();
         try {
             workerGroup = workerGroup.toLowerCase();
@@ -237,7 +311,7 @@ public class ZookeeperNodeManager implements InitializingBean {
         }
     }
 
-    public Map<String, Set<String>> getWorkerGroupNodes(){
+    public Map<String, Set<String>> getWorkerGroupNodes() {
         return Collections.unmodifiableMap(workerGroupNodes);
     }
 
@@ -246,15 +320,15 @@ public class ZookeeperNodeManager implements InitializingBean {
      * @param workerGroup workerGroup
      * @return worker nodes
      */
-    public Set<String> getWorkerGroupNodes(String workerGroup){
+    public Set<String> getWorkerGroupNodes(String workerGroup) {
         workerGroupLock.lock();
         try {
-            if(StringUtils.isEmpty(workerGroup)){
-                workerGroup = DEFAULT_WORKER_GROUP;
+            if (StringUtils.isEmpty(workerGroup)) {
+                workerGroup = Constants.DEFAULT_WORKER_GROUP;
             }
             workerGroup = workerGroup.toLowerCase();
             Set<String> nodes = workerGroupNodes.get(workerGroup);
-            if(CollectionUtils.isNotEmpty(nodes)){
+            if (CollectionUtils.isNotEmpty(nodes)) {
                 return Collections.unmodifiableSet(nodes);
             }
             return nodes;
@@ -264,9 +338,48 @@ public class ZookeeperNodeManager implements InitializingBean {
     }
 
     /**
-     *  close
+     * get worker node info
+     * @return worker node info
+     */
+    public Map<String, String> getWorkerNodeInfo() {
+        return Collections.unmodifiableMap(workerNodeInfo);
+    }
+
+    /**
+     * get worker node info
+     * @param workerNode worker node
+     * @return worker node info
+     */
+    public String getWorkerNodeInfo(String workerNode) {
+        workerNodeInfoLock.lock();
+        try {
+            return workerNodeInfo.getOrDefault(workerNode, null);
+        } finally {
+            workerNodeInfoLock.unlock();
+        }
+    }
+
+    /**
+     * sync worker node info
+     * @param newWorkerNodeInfo new worker node info
      */
-    public void close(){
+    private void syncWorkerNodeInfo(Map<String, String> newWorkerNodeInfo) {
+        workerNodeInfoLock.lock();
+        try {
+            workerNodeInfo.clear();
+            workerNodeInfo.putAll(newWorkerNodeInfo);
+        } finally {
+            workerNodeInfoLock.unlock();
+        }
+    }
+
+    /**
+     *  destroy
+     */
+    @PreDestroy
+    public void destroy() {
+        executorService.shutdownNow();
         registryCenter.close();
     }
+
 }

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/registry/ZookeeperRegistryCenter.java

@@ -17,10 +17,8 @@
 
 package org.apache.dolphinscheduler.server.registry;
 
-import static org.apache.dolphinscheduler.common.Constants.MASTER_PREFIX;
 import static org.apache.dolphinscheduler.common.Constants.SINGLE_SLASH;
 import static org.apache.dolphinscheduler.common.Constants.UNDERLINE;
-import static org.apache.dolphinscheduler.common.Constants.WORKER_PREFIX;
 
 import org.apache.dolphinscheduler.common.Constants;
 import org.apache.dolphinscheduler.common.IStoppable;
@@ -230,9 +228,7 @@ public class ZookeeperRegistryCenter implements InitializingBean {
         // ip_sequence_no
         String[] zNodesPath = zNode.split("\\/");
         String ipSeqNo = zNodesPath[zNodesPath.length - 1];
-
-        String type = serverType.equals(MASTER_PREFIX) ? MASTER_PREFIX : WORKER_PREFIX;
-        String deadServerPath = getDeadZNodeParentPath() + SINGLE_SLASH + type + UNDERLINE + ipSeqNo;
+        String deadServerPath = getDeadZNodeParentPath() + SINGLE_SLASH + serverType + UNDERLINE + ipSeqNo;
 
         return !registerOperator.isExisted(zNode) || registerOperator.isExisted(deadServerPath);
     }

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

@@ -43,11 +43,20 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.WebApplicationType;
 import org.springframework.boot.builder.SpringApplicationBuilder;
 import org.springframework.context.annotation.ComponentScan;
+import org.springframework.context.annotation.FilterType;
 
 /**
  *  worker server
  */
-@ComponentScan("org.apache.dolphinscheduler")
+@ComponentScan(value = "org.apache.dolphinscheduler", excludeFilters = {
+        @ComponentScan.Filter(type = FilterType.REGEX, pattern = {
+                "org.apache.dolphinscheduler.server.master.*",
+                "org.apache.dolphinscheduler.server.monitor.*",
+                "org.apache.dolphinscheduler.server.log.*",
+                "org.apache.dolphinscheduler.server.zk.ZKMasterClient",
+                "org.apache.dolphinscheduler.server.registry.ServerNodeManager"
+        })
+})
 public class WorkerServer implements IStoppable {
 
     /**

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/registry/WorkerRegistry.java

@@ -98,7 +98,7 @@ public class WorkerRegistry {
      * registry
      */
     public void registry() {
-        String address = NetUtils.getHost();
+        String address = NetUtils.getAddr(workerConfig.getListenPort());
         Set<String> workerZkPaths = getWorkerZkPaths();
         int workerHeartbeatInterval = workerConfig.getWorkerHeartbeatInterval();
 
@@ -120,10 +120,10 @@ public class WorkerRegistry {
         }
 
         HeartBeatTask heartBeatTask = new HeartBeatTask(this.startTime,
-                this.workerConfig.getWorkerReservedMemory(),
                 this.workerConfig.getWorkerMaxCpuloadAvg(),
+                this.workerConfig.getWorkerReservedMemory(),
                 workerZkPaths,
-                Constants.WORKER_PREFIX,
+                Constants.WORKER_TYPE,
                 this.zookeeperRegistryCenter);
 
         this.heartBeatExecutor.scheduleAtFixedRate(heartBeatTask, workerHeartbeatInterval, workerHeartbeatInterval, TimeUnit.SECONDS);

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/zk/ZKMasterClient.java

@@ -20,6 +20,7 @@ package org.apache.dolphinscheduler.server.zk;
 import static org.apache.dolphinscheduler.common.Constants.SLEEP_TIME_MILLIS;
 
 import org.apache.dolphinscheduler.common.Constants;
+import org.apache.dolphinscheduler.common.IStoppable;
 import org.apache.dolphinscheduler.common.enums.ExecutionStatus;
 import org.apache.dolphinscheduler.common.enums.ZKNodeType;
 import org.apache.dolphinscheduler.common.model.Server;
@@ -29,7 +30,6 @@ import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
 import org.apache.dolphinscheduler.dao.entity.TaskInstance;
 import org.apache.dolphinscheduler.server.builder.TaskExecutionContextBuilder;
 import org.apache.dolphinscheduler.server.entity.TaskExecutionContext;
-import org.apache.dolphinscheduler.server.master.MasterServer;
 import org.apache.dolphinscheduler.server.master.registry.MasterRegistry;
 import org.apache.dolphinscheduler.server.utils.ProcessUtils;
 import org.apache.dolphinscheduler.service.process.ProcessService;
@@ -73,7 +73,7 @@ public class ZKMasterClient extends AbstractZKClient {
     @Autowired
     private MasterRegistry masterRegistry;
 
-    public void start(MasterServer masterServer) {
+    public void start(IStoppable stoppable) {
         InterProcessMutex mutex = null;
         try {
             // create distributed lock with the root node path of the lock space as /dolphinscheduler/lock/failover/master
@@ -83,7 +83,7 @@ public class ZKMasterClient extends AbstractZKClient {
 
             // master registry
             masterRegistry.registry();
-            masterRegistry.getZookeeperRegistryCenter().setStoppable(masterServer);
+            masterRegistry.getZookeeperRegistryCenter().setStoppable(stoppable);
             String registryPath = this.masterRegistry.getMasterPath();
             masterRegistry.getZookeeperRegistryCenter().getRegisterOperator().handleDeadServer(registryPath, ZKNodeType.MASTER, Constants.DELETE_ZK_OP);
 
@@ -280,7 +280,7 @@ public class ZKMasterClient extends AbstractZKClient {
             return false;
         }
         Date workerServerStartDate = null;
-        List<Server> workerServers = getServersList(ZKNodeType.WORKER);
+        List<Server> workerServers = getServerList(ZKNodeType.WORKER);
         for (Server workerServer : workerServers) {
             if (taskInstance.getHost().equals(workerServer.getHost() + Constants.COLON + workerServer.getPort())) {
                 workerServerStartDate = workerServer.getCreateTime();

dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/master/consumer/TaskPriorityQueueConsumerTest.java

@@ -25,6 +25,7 @@ import org.apache.dolphinscheduler.common.enums.ResourceType;
 import org.apache.dolphinscheduler.common.model.TaskNode;
 import org.apache.dolphinscheduler.common.thread.Stopper;
 import org.apache.dolphinscheduler.common.utils.JSONUtils;
+import org.apache.dolphinscheduler.dao.datasource.SpringConnectionFactory;
 import org.apache.dolphinscheduler.dao.entity.DataSource;
 import org.apache.dolphinscheduler.dao.entity.ProcessDefinition;
 import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
@@ -36,10 +37,12 @@ import org.apache.dolphinscheduler.server.entity.TaskExecutionContext;
 import org.apache.dolphinscheduler.server.master.config.MasterConfig;
 import org.apache.dolphinscheduler.server.master.dispatch.ExecutorDispatcher;
 import org.apache.dolphinscheduler.server.master.dispatch.executor.NettyExecutorManager;
+import org.apache.dolphinscheduler.server.master.registry.MasterRegistry;
 import org.apache.dolphinscheduler.server.registry.DependencyConfig;
-import org.apache.dolphinscheduler.server.registry.ZookeeperNodeManager;
+import org.apache.dolphinscheduler.server.registry.ServerNodeManager;
 import org.apache.dolphinscheduler.server.registry.ZookeeperRegistryCenter;
 import org.apache.dolphinscheduler.server.zk.SpringZKServer;
+import org.apache.dolphinscheduler.server.zk.ZKMasterClient;
 import org.apache.dolphinscheduler.service.bean.SpringApplicationContext;
 import org.apache.dolphinscheduler.service.process.ProcessService;
 import org.apache.dolphinscheduler.service.queue.TaskPriority;
@@ -47,8 +50,6 @@ import org.apache.dolphinscheduler.service.queue.TaskPriorityQueue;
 import org.apache.dolphinscheduler.service.zk.RegisterOperator;
 import org.apache.dolphinscheduler.service.zk.ZookeeperConfig;
 
-import org.apache.curator.CuratorZookeeperClient;
-
 import java.util.ArrayList;
 import java.util.Date;
 import java.util.List;
@@ -67,11 +68,10 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
 @RunWith(SpringJUnit4ClassRunner.class)
 @ContextConfiguration(classes={DependencyConfig.class, SpringApplicationContext.class, SpringZKServer.class,
-        NettyExecutorManager.class, ExecutorDispatcher.class, ZookeeperRegistryCenter.class, TaskPriorityQueueConsumer.class,
-        ZookeeperNodeManager.class, RegisterOperator.class, ZookeeperConfig.class, MasterConfig.class})
+        NettyExecutorManager.class, ExecutorDispatcher.class, ZookeeperRegistryCenter.class, ZKMasterClient.class, TaskPriorityQueueConsumer.class,
+        ServerNodeManager.class, RegisterOperator.class, ZookeeperConfig.class, MasterConfig.class, MasterRegistry.class, SpringConnectionFactory.class})
 public class TaskPriorityQueueConsumerTest {
 
-
     @Autowired
     private TaskPriorityQueue<TaskPriority> taskPriorityQueue;
 

dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/master/dispatch/ExecutorDispatcherTest.java

@@ -17,13 +17,14 @@
 package org.apache.dolphinscheduler.server.master.dispatch;
 
 
+import org.apache.dolphinscheduler.dao.datasource.SpringConnectionFactory;
 import org.apache.dolphinscheduler.remote.NettyRemotingServer;
 import org.apache.dolphinscheduler.remote.config.NettyServerConfig;
 import org.apache.dolphinscheduler.server.master.dispatch.context.ExecutionContext;
 import org.apache.dolphinscheduler.server.master.dispatch.exceptions.ExecuteException;
 import org.apache.dolphinscheduler.server.master.dispatch.executor.NettyExecutorManager;
 import org.apache.dolphinscheduler.server.registry.DependencyConfig;
-import org.apache.dolphinscheduler.server.registry.ZookeeperNodeManager;
+import org.apache.dolphinscheduler.server.registry.ServerNodeManager;
 import org.apache.dolphinscheduler.server.registry.ZookeeperRegistryCenter;
 import org.apache.dolphinscheduler.server.utils.ExecutionContextTestUtils;
 import org.apache.dolphinscheduler.server.worker.config.WorkerConfig;
@@ -33,6 +34,7 @@ import org.apache.dolphinscheduler.server.zk.SpringZKServer;
 import org.apache.dolphinscheduler.service.bean.SpringApplicationContext;
 import org.apache.dolphinscheduler.service.zk.ZookeeperCachedOperator;
 import org.apache.dolphinscheduler.service.zk.ZookeeperConfig;
+
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.mockito.Mockito;
@@ -46,7 +48,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 @RunWith(SpringJUnit4ClassRunner.class)
 @ContextConfiguration(classes={DependencyConfig.class, SpringApplicationContext.class, SpringZKServer.class, WorkerRegistry.class,
         NettyExecutorManager.class, ExecutorDispatcher.class, ZookeeperRegistryCenter.class, WorkerConfig.class,
-        ZookeeperNodeManager.class, ZookeeperCachedOperator.class, ZookeeperConfig.class})
+        ServerNodeManager.class, ZookeeperCachedOperator.class, ZookeeperConfig.class, SpringConnectionFactory.class})
 public class ExecutorDispatcherTest {
 
     @Autowired

dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/master/dispatch/executor/NettyExecutorManagerTest.java

@@ -18,6 +18,7 @@ package org.apache.dolphinscheduler.server.master.dispatch.executor;
 
 import org.apache.dolphinscheduler.common.enums.CommandType;
 import org.apache.dolphinscheduler.common.utils.NetUtils;
+import org.apache.dolphinscheduler.dao.datasource.SpringConnectionFactory;
 import org.apache.dolphinscheduler.dao.entity.ProcessDefinition;
 import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
 import org.apache.dolphinscheduler.dao.entity.TaskInstance;
@@ -30,7 +31,7 @@ import org.apache.dolphinscheduler.server.master.dispatch.context.ExecutionConte
 import org.apache.dolphinscheduler.server.master.dispatch.enums.ExecutorType;
 import org.apache.dolphinscheduler.server.master.dispatch.exceptions.ExecuteException;
 import org.apache.dolphinscheduler.server.registry.DependencyConfig;
-import org.apache.dolphinscheduler.server.registry.ZookeeperNodeManager;
+import org.apache.dolphinscheduler.server.registry.ServerNodeManager;
 import org.apache.dolphinscheduler.server.registry.ZookeeperRegistryCenter;
 import org.apache.dolphinscheduler.server.worker.config.WorkerConfig;
 import org.apache.dolphinscheduler.server.worker.processor.TaskExecuteProcessor;
@@ -53,7 +54,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
  */
 @RunWith(SpringJUnit4ClassRunner.class)
 @ContextConfiguration(classes={DependencyConfig.class, SpringZKServer.class, WorkerRegistry.class,
-        ZookeeperNodeManager.class, ZookeeperRegistryCenter.class, WorkerConfig.class,
+        ServerNodeManager.class, ZookeeperRegistryCenter.class, WorkerConfig.class, SpringConnectionFactory.class,
         ZookeeperCachedOperator.class, ZookeeperConfig.class, SpringApplicationContext.class, NettyExecutorManager.class})
 public class NettyExecutorManagerTest {
 

dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/master/dispatch/host/RoundRobinHostManagerTest.java

@@ -14,66 +14,51 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.dolphinscheduler.server.master.dispatch.host;
 
+package org.apache.dolphinscheduler.server.master.dispatch.host;
 
-import org.apache.dolphinscheduler.common.utils.NetUtils;
 import org.apache.dolphinscheduler.common.utils.StringUtils;
 import org.apache.dolphinscheduler.remote.utils.Host;
 import org.apache.dolphinscheduler.server.master.dispatch.context.ExecutionContext;
-import org.apache.dolphinscheduler.server.registry.DependencyConfig;
-import org.apache.dolphinscheduler.server.registry.ZookeeperNodeManager;
-import org.apache.dolphinscheduler.server.registry.ZookeeperRegistryCenter;
+import org.apache.dolphinscheduler.server.registry.ServerNodeManager;
 import org.apache.dolphinscheduler.server.utils.ExecutionContextTestUtils;
-import org.apache.dolphinscheduler.server.worker.config.WorkerConfig;
-import org.apache.dolphinscheduler.server.worker.registry.WorkerRegistry;
-import org.apache.dolphinscheduler.server.zk.SpringZKServer;
-import org.apache.dolphinscheduler.service.zk.ZookeeperCachedOperator;
-import org.apache.dolphinscheduler.service.zk.ZookeeperConfig;
 
 import org.junit.Assert;
 import org.junit.Test;
 import org.junit.runner.RunWith;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.test.context.ContextConfiguration;
-import org.springframework.test.context.junit4.SpringRunner;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.junit.MockitoJUnitRunner;
 
+import com.google.common.collect.Sets;
 
 /**
  * round robin host manager test
  */
-@RunWith(SpringRunner.class)
-@ContextConfiguration(classes={DependencyConfig.class, SpringZKServer.class, WorkerRegistry.class, ZookeeperRegistryCenter.class, WorkerConfig.class,
-        ZookeeperNodeManager.class, ZookeeperCachedOperator.class, ZookeeperConfig.class})
+@RunWith(MockitoJUnitRunner.class)
 public class RoundRobinHostManagerTest {
 
+    @Mock
+    private ServerNodeManager serverNodeManager;
 
-    @Autowired
-    private ZookeeperNodeManager zookeeperNodeManager;
-
-    @Autowired
-    private WorkerRegistry workerRegistry;
-
-    @Autowired
-    private WorkerConfig workerConfig;
+    @InjectMocks
+    RoundRobinHostManager roundRobinHostManager;
 
     @Test
-    public void testSelectWithEmptyResult(){
-        RoundRobinHostManager roundRobinHostManager = new RoundRobinHostManager();
-        roundRobinHostManager.setZookeeperNodeManager(zookeeperNodeManager);
+    public void testSelectWithEmptyResult() {
+        Mockito.when(serverNodeManager.getWorkerGroupNodes("default")).thenReturn(null);
         ExecutionContext context = ExecutionContextTestUtils.getExecutionContext(10000);
         Host emptyHost = roundRobinHostManager.select(context);
         Assert.assertTrue(StringUtils.isEmpty(emptyHost.getAddress()));
     }
 
     @Test
-    public void testSelectWithResult(){
-        workerRegistry.registry();
-        RoundRobinHostManager roundRobinHostManager = new RoundRobinHostManager();
-        roundRobinHostManager.setZookeeperNodeManager(zookeeperNodeManager);
+    public void testSelectWithResult() {
+        Mockito.when(serverNodeManager.getWorkerGroupNodes("default")).thenReturn(Sets.newHashSet("192.168.1.1:22"));
         ExecutionContext context = ExecutionContextTestUtils.getExecutionContext(10000);
         Host host = roundRobinHostManager.select(context);
         Assert.assertTrue(StringUtils.isNotEmpty(host.getAddress()));
-        Assert.assertTrue(host.getAddress().equalsIgnoreCase(NetUtils.getAddr(workerConfig.getListenPort())));
+        Assert.assertTrue(host.getAddress().equalsIgnoreCase("192.168.1.1:22"));
     }
 }

dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/master/processor/queue/TaskResponseServiceTest.java

@@ -18,12 +18,16 @@ package org.apache.dolphinscheduler.server.master.processor.queue;
 
 
 import org.apache.dolphinscheduler.common.enums.ExecutionStatus;
+import org.apache.dolphinscheduler.dao.datasource.SpringConnectionFactory;
 import org.apache.dolphinscheduler.server.registry.DependencyConfig;
-import org.apache.dolphinscheduler.server.registry.ZookeeperNodeManager;
+import org.apache.dolphinscheduler.server.registry.ServerNodeManager;
 import org.apache.dolphinscheduler.server.registry.ZookeeperRegistryCenter;
 import org.apache.dolphinscheduler.server.zk.SpringZKServer;
 import org.apache.dolphinscheduler.service.zk.ZookeeperCachedOperator;
 import org.apache.dolphinscheduler.service.zk.ZookeeperConfig;
+
+import java.util.Date;
+
 import org.junit.Assert;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -31,11 +35,10 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
-import java.util.Date;
-
 @RunWith(SpringJUnit4ClassRunner.class)
 @ContextConfiguration(classes={DependencyConfig.class, SpringZKServer.class, TaskResponseService.class, ZookeeperRegistryCenter.class,
-        ZookeeperCachedOperator.class, ZookeeperConfig.class, ZookeeperNodeManager.class, TaskResponseService.class})
+        ZookeeperCachedOperator.class, ZookeeperConfig.class, ServerNodeManager.class, TaskResponseService.class,
+        SpringConnectionFactory.class})
 public class TaskResponseServiceTest {
 
     @Autowired

dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/master/runner/MasterTaskExecThreadTest.java

@@ -18,18 +18,23 @@
 package org.apache.dolphinscheduler.server.master.runner;
 
 import org.apache.dolphinscheduler.common.enums.ExecutionStatus;
+import org.apache.dolphinscheduler.dao.datasource.SpringConnectionFactory;
 import org.apache.dolphinscheduler.dao.entity.TaskInstance;
 import org.apache.dolphinscheduler.server.master.consumer.TaskPriorityQueueConsumer;
 import org.apache.dolphinscheduler.server.master.dispatch.ExecutorDispatcher;
 import org.apache.dolphinscheduler.server.master.dispatch.executor.NettyExecutorManager;
 import org.apache.dolphinscheduler.server.registry.DependencyConfig;
-import org.apache.dolphinscheduler.server.registry.ZookeeperNodeManager;
+import org.apache.dolphinscheduler.server.registry.ServerNodeManager;
 import org.apache.dolphinscheduler.server.registry.ZookeeperRegistryCenter;
 import org.apache.dolphinscheduler.server.zk.SpringZKServer;
 import org.apache.dolphinscheduler.service.bean.SpringApplicationContext;
 import org.apache.dolphinscheduler.service.process.ProcessService;
 import org.apache.dolphinscheduler.service.zk.ZookeeperCachedOperator;
 import org.apache.dolphinscheduler.service.zk.ZookeeperConfig;
+
+import java.util.HashSet;
+import java.util.Set;
+
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.mockito.Mockito;
@@ -37,13 +42,10 @@ import org.springframework.context.ApplicationContext;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
-import java.util.HashSet;
-import java.util.Set;
-
 @RunWith(SpringJUnit4ClassRunner.class)
 @ContextConfiguration(classes={DependencyConfig.class, SpringApplicationContext.class, SpringZKServer.class,
         NettyExecutorManager.class, ExecutorDispatcher.class, ZookeeperRegistryCenter.class, TaskPriorityQueueConsumer.class,
-        ZookeeperNodeManager.class, ZookeeperCachedOperator.class, ZookeeperConfig.class})
+        ServerNodeManager.class, ZookeeperCachedOperator.class, ZookeeperConfig.class, SpringConnectionFactory.class})
 public class MasterTaskExecThreadTest {
 
 

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

@@ -14,11 +14,12 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.dolphinscheduler.server.registry;
 
+package org.apache.dolphinscheduler.server.registry;
 
 import org.apache.dolphinscheduler.common.utils.CollectionUtils;
 import org.apache.dolphinscheduler.common.utils.NetUtils;
+import org.apache.dolphinscheduler.dao.datasource.SpringConnectionFactory;
 import org.apache.dolphinscheduler.server.master.config.MasterConfig;
 import org.apache.dolphinscheduler.server.master.registry.MasterRegistry;
 import org.apache.dolphinscheduler.server.worker.config.WorkerConfig;
@@ -38,16 +39,16 @@ import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
 /**
- * zookeeper node manager test
+ * server node manager test
  */
 @RunWith(SpringJUnit4ClassRunner.class)
 @ContextConfiguration(classes={DependencyConfig.class, SpringZKServer.class, MasterRegistry.class,WorkerRegistry.class,
-        ZookeeperRegistryCenter.class, MasterConfig.class, WorkerConfig.class,
-        ZookeeperCachedOperator.class, ZookeeperConfig.class, ZookeeperNodeManager.class})
-public class ZookeeperNodeManagerTest {
+        ZookeeperRegistryCenter.class, MasterConfig.class, WorkerConfig.class, SpringConnectionFactory.class,
+        ZookeeperCachedOperator.class, ZookeeperConfig.class, ServerNodeManager.class})
+public class ServerNodeManagerTest {
 
     @Autowired
-    private ZookeeperNodeManager zookeeperNodeManager;
+    private ServerNodeManager serverNodeManager;
 
     @Autowired
     private MasterRegistry masterRegistry;
@@ -68,11 +69,11 @@ public class ZookeeperNodeManagerTest {
     public void testGetMasterNodes(){
         masterRegistry.registry();
         try {
-            //let the zookeeperNodeManager catch the registry event
+            //let the serverNodeManager catch the registry event
             Thread.sleep(2000);
         } catch (InterruptedException ignore) {
         }
-        Set<String> masterNodes = zookeeperNodeManager.getMasterNodes();
+        Set<String> masterNodes = serverNodeManager.getMasterNodes();
         Assert.assertTrue(CollectionUtils.isNotEmpty(masterNodes));
         Assert.assertEquals(1, masterNodes.size());
         Assert.assertEquals(NetUtils.getAddr(masterConfig.getListenPort()), masterNodes.iterator().next());
@@ -82,11 +83,11 @@ public class ZookeeperNodeManagerTest {
     public void testGetWorkerGroupNodes(){
         workerRegistry.registry();
         try {
-            //let the zookeeperNodeManager catch the registry event
+            //let the serverNodeManager catch the registry event
             Thread.sleep(2000);
         } catch (InterruptedException ignore) {
         }
-        Map<String, Set<String>> workerGroupNodes = zookeeperNodeManager.getWorkerGroupNodes();
+        Map<String, Set<String>> workerGroupNodes = serverNodeManager.getWorkerGroupNodes();
         Assert.assertEquals(1, workerGroupNodes.size());
         Assert.assertEquals("default".trim(), workerGroupNodes.keySet().iterator().next());
     }
@@ -95,12 +96,11 @@ public class ZookeeperNodeManagerTest {
     public void testGetWorkerGroupNodesWithParam(){
         workerRegistry.registry();
         try {
-            //let the zookeeperNodeManager catch the registry event
+            //let the serverNodeManager catch the registry event
             Thread.sleep(3000);
         } catch (InterruptedException ignore) {
         }
-        Map<String, Set<String>> workerGroupNodes = zookeeperNodeManager.getWorkerGroupNodes();
-        Set<String> workerNodes = zookeeperNodeManager.getWorkerGroupNodes("default");
+        Set<String> workerNodes = serverNodeManager.getWorkerGroupNodes("default");
         Assert.assertTrue(CollectionUtils.isNotEmpty(workerNodes));
         Assert.assertEquals(1, workerNodes.size());
         Assert.assertEquals(NetUtils.getAddr(workerConfig.getListenPort()), workerNodes.iterator().next());

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

@@ -18,6 +18,7 @@
 package org.apache.dolphinscheduler.server.worker.processor;
 
 import org.apache.dolphinscheduler.common.thread.Stopper;
+import org.apache.dolphinscheduler.dao.datasource.SpringConnectionFactory;
 import org.apache.dolphinscheduler.remote.NettyRemotingClient;
 import org.apache.dolphinscheduler.remote.NettyRemotingServer;
 import org.apache.dolphinscheduler.remote.command.CommandType;
@@ -31,7 +32,7 @@ import org.apache.dolphinscheduler.server.master.processor.TaskAckProcessor;
 import org.apache.dolphinscheduler.server.master.processor.TaskResponseProcessor;
 import org.apache.dolphinscheduler.server.master.processor.queue.TaskResponseService;
 import org.apache.dolphinscheduler.server.master.registry.MasterRegistry;
-import org.apache.dolphinscheduler.server.registry.ZookeeperNodeManager;
+import org.apache.dolphinscheduler.server.registry.ServerNodeManager;
 import org.apache.dolphinscheduler.server.registry.ZookeeperRegistryCenter;
 import org.apache.dolphinscheduler.server.worker.cache.impl.TaskExecutionContextCacheManagerImpl;
 import org.apache.dolphinscheduler.server.worker.config.WorkerConfig;
@@ -61,24 +62,11 @@ import io.netty.channel.Channel;
  */
 @RunWith(SpringJUnit4ClassRunner.class)
 @ContextConfiguration(classes = {
-    TaskCallbackServiceTestConfig.class,
-    SpringZKServer.class,
-    SpringApplicationContext.class,
-    MasterRegistry.class,
-    WorkerRegistry.class,
-    ZookeeperRegistryCenter.class,
-    MasterConfig.class,
-    WorkerConfig.class,
-    RegisterOperator.class,
-    ZookeeperConfig.class,
-    ZookeeperNodeManager.class,
-    TaskCallbackService.class,
-    TaskResponseService.class,
-    TaskAckProcessor.class,
-    TaskResponseProcessor.class,
-    TaskExecuteProcessor.class,
-    CuratorZookeeperClient.class,
-    TaskExecutionContextCacheManagerImpl.class})
+        TaskCallbackServiceTestConfig.class, SpringZKServer.class, SpringApplicationContext.class,
+        SpringConnectionFactory.class, MasterRegistry.class, WorkerRegistry.class, ZookeeperRegistryCenter.class,
+        MasterConfig.class, WorkerConfig.class, RegisterOperator.class, ZookeeperConfig.class, ServerNodeManager.class,
+        TaskCallbackService.class, TaskResponseService.class, TaskAckProcessor.class, TaskResponseProcessor.class,
+        TaskExecuteProcessor.class, CuratorZookeeperClient.class, TaskExecutionContextCacheManagerImpl.class})
 public class TaskCallbackServiceTest {
 
     @Autowired

dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/zk/AbstractZKClient.java

@@ -17,14 +17,8 @@
 
 package org.apache.dolphinscheduler.service.zk;
 
-import static org.apache.dolphinscheduler.common.Constants.ADD_ZK_OP;
 import static org.apache.dolphinscheduler.common.Constants.COLON;
-import static org.apache.dolphinscheduler.common.Constants.DELETE_ZK_OP;
 import static org.apache.dolphinscheduler.common.Constants.DIVISION_STRING;
-import static org.apache.dolphinscheduler.common.Constants.MASTER_PREFIX;
-import static org.apache.dolphinscheduler.common.Constants.SINGLE_SLASH;
-import static org.apache.dolphinscheduler.common.Constants.UNDERLINE;
-import static org.apache.dolphinscheduler.common.Constants.WORKER_PREFIX;
 
 import org.apache.dolphinscheduler.common.Constants;
 import org.apache.dolphinscheduler.common.enums.ZKNodeType;
@@ -35,9 +29,12 @@ import org.apache.dolphinscheduler.common.utils.StringUtils;
 import org.apache.curator.framework.recipes.locks.InterProcessMutex;
 
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -82,59 +79,123 @@ public abstract class AbstractZKClient extends RegisterOperator {
      * @param zkNodeType zookeeper node type
      * @return server list
      */
-    public List<Server> getServersList(ZKNodeType zkNodeType) {
-        Map<String, String> masterMap = getServerMaps(zkNodeType);
+    public List<Server> getServerList(ZKNodeType zkNodeType) {
+        Map<String, String> serverMaps = getServerMaps(zkNodeType);
         String parentPath = getZNodeParentPath(zkNodeType);
 
-        List<Server> masterServers = new ArrayList<>();
-        for (Map.Entry<String, String> entry : masterMap.entrySet()) {
-            Server masterServer = ResInfo.parseHeartbeatForZKInfo(entry.getValue());
-            if (masterServer == null) {
+        List<Server> serverList = new ArrayList<>();
+        for (Map.Entry<String, String> entry : serverMaps.entrySet()) {
+            Server server = ResInfo.parseHeartbeatForZKInfo(entry.getValue());
+            if (server == null) {
                 continue;
             }
             String key = entry.getKey();
-            masterServer.setZkDirectory(parentPath + "/" + key);
-            //set host and port
+            server.setZkDirectory(parentPath + "/" + key);
+            // set host and port
             String[] hostAndPort = key.split(COLON);
             String[] hosts = hostAndPort[0].split(DIVISION_STRING);
             // fetch the last one
-            masterServer.setHost(hosts[hosts.length - 1]);
-            masterServer.setPort(Integer.parseInt(hostAndPort[1]));
-            masterServers.add(masterServer);
+            server.setHost(hosts[hosts.length - 1]);
+            server.setPort(Integer.parseInt(hostAndPort[1]));
+            serverList.add(server);
         }
-        return masterServers;
+        return serverList;
     }
 
     /**
-     * get master server list map.
+     * get server zk nodes.
      *
      * @param zkNodeType zookeeper node type
-     * @return result : {host : resource info}
+     * @return result : list<zknode>
      */
-    public Map<String, String> getServerMaps(ZKNodeType zkNodeType) {
+    public List<String> getServerZkNodes(ZKNodeType zkNodeType) {
+        String path = getZNodeParentPath(zkNodeType);
+        List<String> serverList = super.getChildrenKeys(path);
+        if (zkNodeType == ZKNodeType.WORKER) {
+            List<String> workerList = new ArrayList<>();
+            for (String group : serverList) {
+                List<String> groupServers = super.getChildrenKeys(path + Constants.SLASH + group);
+                for (String groupServer : groupServers) {
+                    workerList.add(group + Constants.SLASH + groupServer);
+                }
+            }
+            serverList = workerList;
+        }
+        return serverList;
+    }
 
-        Map<String, String> masterMap = new HashMap<>();
+    /**
+     * get server list map.
+     *
+     * @param zkNodeType zookeeper node type
+     * @param hostOnly host only
+     * @return result : {host : resource info}
+     */
+    public Map<String, String> getServerMaps(ZKNodeType zkNodeType, boolean hostOnly) {
+        Map<String, String> serverMap = new HashMap<>();
         try {
             String path = getZNodeParentPath(zkNodeType);
-            List<String> serverList = super.getChildrenKeys(path);
-            if (zkNodeType == ZKNodeType.WORKER) {
-                List<String> workerList = new ArrayList<>();
-                for (String group : serverList) {
-                    List<String> groupServers = super.getChildrenKeys(path + Constants.SLASH + group);
-                    for (String groupServer : groupServers) {
-                        workerList.add(group + Constants.SLASH + groupServer);
-                    }
+            List<String> serverList = getServerZkNodes(zkNodeType);
+            for (String server : serverList) {
+                String host = server;
+                if (zkNodeType == ZKNodeType.WORKER && hostOnly) {
+                    host = server.split(Constants.SLASH)[1];
                 }
-                serverList = workerList;
+                serverMap.putIfAbsent(host, super.get(path + Constants.SLASH + server));
             }
+        } catch (Exception e) {
+            logger.error("get server list failed", e);
+        }
+
+        return serverMap;
+    }
+
+    /**
+     * get server list map.
+     *
+     * @param zkNodeType zookeeper node type
+     * @return result : {host : resource info}
+     */
+    public Map<String, String> getServerMaps(ZKNodeType zkNodeType) {
+        return getServerMaps(zkNodeType, false);
+    }
+
+    /**
+     * get server node set.
+     *
+     * @param zkNodeType zookeeper node type
+     * @param hostOnly host only
+     * @return result : set<host>
+     */
+    public Set<String> getServerNodeSet(ZKNodeType zkNodeType, boolean hostOnly) {
+        Set<String> serverSet = new HashSet<>();
+        try {
+            List<String> serverList = getServerZkNodes(zkNodeType);
             for (String server : serverList) {
-                masterMap.putIfAbsent(server, super.get(path + Constants.SLASH + server));
+                String host = server;
+                if (zkNodeType == ZKNodeType.WORKER && hostOnly) {
+                    host = server.split(Constants.SLASH)[1];
+                }
+                serverSet.add(host);
             }
         } catch (Exception e) {
-            logger.error("get server list failed", e);
+            logger.error("get server node set failed", e);
         }
+        return serverSet;
+    }
 
-        return masterMap;
+    /**
+     * get server node list.
+     *
+     * @param zkNodeType zookeeper node type
+     * @param hostOnly host only
+     * @return result : list<host>
+     */
+    public List<String> getServerNodeList(ZKNodeType zkNodeType, boolean hostOnly) {
+        Set<String> serverSet = getServerNodeSet(zkNodeType, hostOnly);
+        List<String> serverList = new ArrayList<>(serverSet);
+        Collections.sort(serverList);
+        return serverList;
     }
 
     /**
@@ -151,7 +212,7 @@ public abstract class AbstractZKClient extends RegisterOperator {
                     host, zkNodeType);
             return false;
         }
-        Map<String, String> serverMaps = getServerMaps(zkNodeType);
+        Map<String, String> serverMaps = getServerMaps(zkNodeType, true);
         for (String hostKey : serverMaps.keySet()) {
             if (hostKey.contains(host)) {
                 return true;

dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/zk/RegisterOperator.java

@@ -19,10 +19,10 @@ package org.apache.dolphinscheduler.service.zk;
 
 import static org.apache.dolphinscheduler.common.Constants.ADD_ZK_OP;
 import static org.apache.dolphinscheduler.common.Constants.DELETE_ZK_OP;
-import static org.apache.dolphinscheduler.common.Constants.MASTER_PREFIX;
+import static org.apache.dolphinscheduler.common.Constants.MASTER_TYPE;
 import static org.apache.dolphinscheduler.common.Constants.SINGLE_SLASH;
 import static org.apache.dolphinscheduler.common.Constants.UNDERLINE;
-import static org.apache.dolphinscheduler.common.Constants.WORKER_PREFIX;
+import static org.apache.dolphinscheduler.common.Constants.WORKER_TYPE;
 
 import org.apache.dolphinscheduler.common.Constants;
 import org.apache.dolphinscheduler.common.enums.ZKNodeType;
@@ -99,7 +99,7 @@ public class RegisterOperator extends ZookeeperCachedOperator {
      */
     public void handleDeadServer(String zNode, ZKNodeType zkNodeType, String opType) throws Exception {
         String host = getHostByEventDataPath(zNode);
-        String type = (zkNodeType == ZKNodeType.MASTER) ? MASTER_PREFIX : WORKER_PREFIX;
+        String type = (zkNodeType == ZKNodeType.MASTER) ? MASTER_TYPE : WORKER_TYPE;
 
         //check server restart, if restart , dead server path in zk should be delete
         if (opType.equals(DELETE_ZK_OP)) {
@@ -130,7 +130,7 @@ public class RegisterOperator extends ZookeeperCachedOperator {
      */
     public void handleDeadServer(Set<String> zNodeSet, ZKNodeType zkNodeType, String opType) throws Exception {
 
-        String type = (zkNodeType == ZKNodeType.MASTER) ? MASTER_PREFIX : WORKER_PREFIX;
+        String type = (zkNodeType == ZKNodeType.MASTER) ? MASTER_TYPE : WORKER_TYPE;
         for (String zNode : zNodeSet) {
             String host = getHostByEventDataPath(zNode);
             //check server restart, if restart , dead server path in zk should be delete

dolphinscheduler-service/src/test/java/org/apache/dolphinscheduler/service/zk/RegisterOperatorTest.java

@@ -97,7 +97,7 @@ public class RegisterOperatorTest {
         testAfterPropertiesSet();
         registerOperator.handleDeadServer(MASTER_NODE, ZKNodeType.MASTER,Constants.ADD_ZK_OP);
         String path = registerOperator.getDeadZNodeParentPath();
-        Assert.assertTrue(registerOperator.getChildrenKeys(path).contains(String.format("%s_%s",Constants.MASTER_PREFIX,MASTER_NODE)));
+        Assert.assertTrue(registerOperator.getChildrenKeys(path).contains(String.format("%s_%s",Constants.MASTER_TYPE,MASTER_NODE)));
 
     }
 
@@ -107,10 +107,10 @@ public class RegisterOperatorTest {
         String path = registerOperator.getDeadZNodeParentPath();
 
         registerOperator.handleDeadServer(MASTER_NODE, ZKNodeType.MASTER,Constants.ADD_ZK_OP);
-        Assert.assertTrue(registerOperator.getChildrenKeys(path).contains(String.format("%s_%s",Constants.MASTER_PREFIX,MASTER_NODE)));
+        Assert.assertTrue(registerOperator.getChildrenKeys(path).contains(String.format("%s_%s",Constants.MASTER_TYPE,MASTER_NODE)));
 
-        registerOperator.removeDeadServerByHost(MASTER_NODE,Constants.MASTER_PREFIX);
-        Assert.assertFalse(registerOperator.getChildrenKeys(path).contains(String.format("%s_%s",Constants.MASTER_PREFIX,MASTER_NODE)));
+        registerOperator.removeDeadServerByHost(MASTER_NODE,Constants.MASTER_TYPE);
+        Assert.assertFalse(registerOperator.getChildrenKeys(path).contains(String.format("%s_%s",Constants.MASTER_TYPE,MASTER_NODE)));
     }
 
     @Test

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

@@ -34,7 +34,7 @@
           <th>
             <span>{{$t('Update Time')}}</span>
           </th>
-          <th width="70">
+          <th scope="col" width="70">
             <span>{{$t('Operation')}}</span>
           </th>
         </tr>

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

@@ -40,7 +40,7 @@
           <th>
             <span>{{$t('Update Time')}}</span>
           </th>
-          <th width="70">
+          <th scope="col" width="70">
             <span>{{$t('Operation')}}</span>
           </th>
         </tr>

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

@@ -47,7 +47,7 @@
           <th>
             <span>{{$t('Update Time')}}</span>
           </th>
-          <th width="120">
+          <th scope="col" width="100">
             <span>{{$t('Operation')}}</span>
           </th>
         </tr>

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

@@ -37,7 +37,7 @@
           <th>
             <span>{{$t('Update Time')}}</span>
           </th>
-          <th width="120">
+          <th scope="col" width="100">
             <span>{{$t('Operation')}}</span>
           </th>
         </tr>

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

@@ -34,17 +34,9 @@
           </template>
         </m-list-box-f>
         <m-list-box-f>
-          <template slot="name"><strong>*</strong>IP</template>
+          <template slot="name"><strong>*</strong>{{$t('Worker Addresses')}}</template>
           <template slot="content">
-            <x-input
-                    :autosize="{ minRows: 4, maxRows: 6 }"
-                    type="textarea"
-                    v-model="ipList"
-                    :placeholder="$t('Please enter the IP address separated by commas')">
-            </x-input>
-            <div class="ipt-tip">
-              <span>{{$t('Note: Multiple IP addresses have been comma separated')}}</span>
-            </div>
+            <treeselect :options="this.workerAddressList" v-model="addrList" :multiple="true" :placeholder="$t('Please select the worker addresses')"></treeselect>
           </template>
         </m-list-box-f>
       </div>
@@ -56,6 +48,8 @@
   import store from '@/conf/home/store'
   import mPopup from '@/module/components/popup/popup'
   import mListBoxF from '@/module/components/listBoxF/listBoxF'
+  import Treeselect from '@riophae/vue-treeselect'
+  import '@riophae/vue-treeselect/dist/vue-treeselect.css'
 
   export default {
     name: 'create-warning',
@@ -64,11 +58,12 @@
         store,
         id: 0,
         name: '',
-        ipList: ''
+        addrList: []
       }
     },
     props: {
-      item: Object
+      item: Object,
+      workerAddressList: Object
     },
     methods: {
       _ok () {
@@ -77,28 +72,14 @@
           this._submit()
         }
       },
-      checkIsIps(ips) {
-        let reg = /^(\d{1,2}|1\d\d|2[0-4]\d|25[0-5])\.(\d{1,2}|1\d\d|2[0-4]\d|25[0-5])\.(\d{1,2}|1\d\d|2[0-4]\d|25[0-5])\.(\d{1,2}|1\d\d|2[0-4]\d|25[0-5])$/
-        let valdata = ips.split(',');
-        for(let i=0;i<valdata.length;i++){
-            if(reg.test(valdata[i])== false){
-                return false;
-            }
-        }
-        return true
-      },
       _verification () {
         // group name
         if (!this.name) {
           this.$message.warning(`${i18n.$t('Please enter group name')}`)
           return false
         }
-        if (!this.ipList) {
-          this.$message.warning(`${i18n.$t('IP address cannot be empty')}`)
-          return false
-        }
-        if(!this.checkIsIps(this.ipList)) {
-          this.$message.warning(`${i18n.$t('Please enter the correct IP')}`)
+        if (!this.addrList.length) {
+          this.$message.warning(`${i18n.$t('Worker addresses cannot be empty')}`)
           return false
         }
         return true
@@ -107,7 +88,7 @@
         let param = {
           id: this.id,
           name: this.name,
-          ipList: this.ipList
+          addrList: this.addrList.join(',')
         }
         if (this.item) {
           param.id = this.item.id
@@ -130,20 +111,11 @@
       if (this.item) {
         this.id = this.item.id
         this.name = this.item.name
-        this.ipList = this.item.ipList
+        this.addrList = this.item.addrList.split(',')
       }
     },
     mounted () {
     },
-    components: { mPopup, mListBoxF }
+    components: { mPopup, mListBoxF, Treeselect }
   }
 </script>
-<style lang="scss" rel="stylesheet/scss">
-  .create-worker-model {
-    .ipt-tip {
-      color: #999;
-      padding-top: 4px;
-      display: block;
-    }
-  }
-</style>

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

@@ -26,7 +26,7 @@
             <span>{{$t('Group')}}</span>
           </th>
           <th>
-            <span>IPList</span>
+            <span>{{$t('Addresses')}}</span>
           </th>
           <th>
             <span>{{$t('Create Time')}}</span>
@@ -34,18 +34,19 @@
           <th>
             <span>{{$t('Update Time')}}</span>
           </th>
+          <th scope="col" width="70">
+            <span>{{$t('Operation')}}</span>
+          </th>
         </tr>
         <tr v-for="(item, $index) in list" :key="$index">
           <td>
             <span>{{parseInt(pageNo === 1 ? ($index + 1) : (($index + 1) + (pageSize * (pageNo - 1))))}}</span>
           </td>
           <td>
-            <span>
-              {{item.name}}
-            </span>
+            <span>{{item.name}}</span>
           </td>
           <td>
-            <span>{{item.ipList.join(',')}}</span>
+            <span style="display: inline-block; margin-left: 10px; margin-right: 10px">{{item.addrList}}</span>
           </td>
           <td>
             <span v-if="item.createTime">{{item.createTime | formatDate}}</span>
@@ -55,6 +56,24 @@
             <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)" v-if="!item.systemDefault"></x-button>
+            <x-poptip
+                    :ref="'poptip-delete-' + $index"
+                    placement="bottom-end"
+                    width="90"
+                    v-if="!item.systemDefault">
+              <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>

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

@@ -17,7 +17,11 @@
 <template>
   <m-list-construction :title="$t('Worker group manage')">
     <template slot="conditions">
-      <m-conditions @on-conditions="_onConditions"></m-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>
     </template>
     <template slot="content">
       <template v-if="workerGroupList.length || total>0">
@@ -57,6 +61,7 @@
         total: null,
         isLoading: false,
         workerGroupList: [],
+        workerAddressList: [],
         searchParams: {
           pageSize: 10,
           pageNo: 1,
@@ -68,7 +73,7 @@
     mixins: [listUrlParamHandle],
     props: {},
     methods: {
-      ...mapActions('security', ['getWorkerGroups']),
+      ...mapActions('security', ['getWorkerGroups', 'getWorkerAddresses']),
       /**
        * Inquire
        */
@@ -108,7 +113,8 @@
                 }
               },
               props: {
-                item: item
+                item: item,
+                workerAddressList: self.workerAddressList
               }
             })
           }
@@ -128,6 +134,11 @@
         }).catch(e => {
           this.isLoading = false
         })
+      },
+      _getWorkerAddressList () {
+        this.getWorkerAddresses().then(res => {
+          this.workerAddressList = res.data.map(x => ({ id: x, label: x }))
+        })
       }
     },
     watch: {
@@ -137,7 +148,9 @@
         this.searchParams.pageNo = _.isEmpty(a.query) ? 1 : a.query.pageNo
       }
     },
-    created () {},
+    created () {
+      this._getWorkerAddressList()
+    },
     mounted () {
       this.$modal.destroy()
     },

dolphinscheduler-ui/src/js/conf/home/store/security/actions.js

@@ -521,7 +521,16 @@ export default {
   },
   deleteWorkerGroups ({ state }, payload) {
     return new Promise((resolve, reject) => {
-      io.get('worker-group/delete-by-id', payload, res => {
+      io.post('worker-group/delete-by-id', payload, res => {
+        resolve(res)
+      }).catch(e => {
+        reject(e)
+      })
+    })
+  },
+  getWorkerAddresses ({ state }, payload) {
+    return new Promise((resolve, reject) => {
+      io.get('worker-group/worker-address-list', payload, res => {
         resolve(res)
       }).catch(e => {
         reject(e)

dolphinscheduler-ui/src/js/module/i18n/locale/en_US.js

@@ -353,7 +353,7 @@ export default {
   'Drag the file into the current upload window': 'Drag the file into the current upload window',
   'Drag area upload': 'Drag area upload',
   Upload: 'Upload',
-  'ReUpload File': 'Re-upload file',
+  'ReUpload File': 'ReUpload File',
   'Please enter file name': 'Please enter file name',
   'Please select the file to upload': 'Please select the file to upload',
   'Resources manage': 'Resources',
@@ -479,8 +479,9 @@ export default {
   'Token manage': 'Token manage',
   'Create token': 'Create token',
   'Edit token': 'Edit token',
-  'Please enter the IP address separated by commas': 'Please enter the IP address separated by commas',
-  'Note: Multiple IP addresses have been comma separated': 'Note: Multiple IP addresses have been comma separated',
+  Addresses: 'Addresses',
+  'Worker Addresses': 'Worker Addresses',
+  'Please select the worker addresses': 'Please select the worker addresses',
   'Expiration time': 'Expiration time',
   User: 'User',
   'Please enter token': 'Please enter token',
@@ -541,8 +542,7 @@ export default {
   'Please Enter Http Url': 'Please Enter Http Url(required)',
   'Please Enter Http Condition': 'Please Enter Http Condition',
   'There is no data for this period of time': 'There is no data for this period of time',
-  'IP address cannot be empty': 'IP address cannot be empty',
-  'Please enter the correct IP': 'Please enter the correct IP',
+  'Worker addresses cannot be empty': 'Worker addresses cannot be empty',
   'Please generate token': 'Please generate token',
   'Spark Version': 'Spark Version',
   TargetDataBase: 'target database',

dolphinscheduler-ui/src/js/module/i18n/locale/zh_CN.js

@@ -479,8 +479,9 @@ export default {
   'Token manage': '令牌管理',
   'Create token': '创建令牌',
   'Edit token': '编辑令牌',
-  'Please enter the IP address separated by commas': '请输入IP地址多个用英文逗号隔开',
-  'Note: Multiple IP addresses have been comma separated': '注意：多个IP地址以英文逗号分割',
+  Addresses: '地址',
+  'Worker Addresses': 'Worker地址',
+  'Please select the worker addresses': '请选择Worker地址',
   'Expiration time': '失效时间',
   User: '用户',
   'Please enter token': '请输入令牌',
@@ -541,8 +542,7 @@ export default {
   'Please Enter Http Url': '请填写请求地址(必填)',
   'Please Enter Http Condition': '请填写校验内容',
   'There is no data for this period of time': '该时间段无数据',
-  'IP address cannot be empty': 'IP地址不能为空',
-  'Please enter the correct IP': '请输入正确的IP',
+  'Worker addresses cannot be empty': 'Worker地址不能为空',
   'Please generate token': '请生成Token',
   'Spark Version': 'Spark版本',
   TargetDataBase: '目标库',

pom.xml

@@ -817,7 +817,7 @@
                         <include>**/server/master/ConditionsTaskTest.java</include>
                         <include>**/server/master/MasterExecThreadTest.java</include> -->
                         <include>**/server/master/ParamsTest.java</include>
-                        <include>**/server/register/ZookeeperNodeManagerTest.java</include>
+                        <include>**/server/register/ServerNodeManagerTest.java</include>
                         <include>**/server/register/ZookeeperRegistryCenterTest.java</include>
                         <include>**/server/utils/DataxUtilsTest.java</include>
                         <include>**/server/utils/ExecutionContextTestUtils.java</include>

sql/dolphinscheduler_mysql.sql

@@ -784,6 +784,23 @@ CREATE TABLE `t_ds_user` (
 -- Records of t_ds_user
 -- ----------------------------
 
+-- ----------------------------
+-- Table structure for t_ds_worker_group
+-- ----------------------------
+DROP TABLE IF EXISTS `t_ds_worker_group`;
+CREATE TABLE `t_ds_worker_group` (
+  `id` bigint(11) NOT NULL AUTO_INCREMENT COMMENT 'id',
+  `name` varchar(256) NOT NULL COMMENT 'worker group name',
+  `addr_list` text NULL DEFAULT NULL COMMENT 'worker addr list. split by [,]',
+  `create_time` datetime NULL DEFAULT NULL COMMENT 'create time',
+  `update_time` datetime NULL DEFAULT NULL COMMENT 'update time',
+  PRIMARY KEY (`id`),
+  UNIQUE KEY `name_unique` (`name`)
+) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8;
+
+-- ----------------------------
+-- Records of t_ds_worker_group
+-- ----------------------------
 
 -- ----------------------------
 -- Table structure for t_ds_version

sql/dolphinscheduler_postgre.sql

@@ -647,11 +647,12 @@ create index version_index on t_ds_version(version);
 DROP TABLE IF EXISTS t_ds_worker_group;
 CREATE TABLE t_ds_worker_group (
   id bigint NOT NULL  ,
-  name varchar(256) DEFAULT NULL ,
-  ip_list varchar(256) DEFAULT NULL ,
+  name varchar(256) NOT NULL ,
+  addr_list text DEFAULT NULL ,
   create_time timestamp DEFAULT NULL ,
   update_time timestamp DEFAULT NULL ,
-  PRIMARY KEY (id)
+  PRIMARY KEY (id) ,
+  CONSTRAINT name_unique UNIQUE (name)
 ) ;
 
 --

sql/upgrade/1.3.6_schema/mysql/dolphinscheduler_ddl.sql

@@ -0,0 +1,40 @@
+/*
+ * 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.
+*/
+
+SET sql_mode=(SELECT REPLACE(@@sql_mode,'ONLY_FULL_GROUP_BY',''));
+
+-- uc_dolphin_T_t_ds_worker_group_R_ip_list
+drop PROCEDURE if EXISTS uc_dolphin_T_t_ds_worker_group_R_ip_list;
+delimiter d//
+CREATE PROCEDURE uc_dolphin_T_t_ds_worker_group_R_ip_list()
+BEGIN
+    IF EXISTS (SELECT 1 FROM information_schema.COLUMNS
+        WHERE TABLE_NAME='t_ds_worker_group'
+        AND TABLE_SCHEMA=(SELECT DATABASE())
+        AND COLUMN_NAME ='ip_list')
+    THEN
+        ALTER TABLE t_ds_worker_group CHANGE COLUMN `ip_list` `addr_list` text;
+        ALTER TABLE t_ds_worker_group MODIFY COLUMN `name` varchar(256) NOT NULL;
+        ALTER TABLE t_ds_worker_group ADD UNIQUE KEY `name_unique` (`name`);
+    END IF;
+END;
+
+d//
+
+delimiter ;
+CALL uc_dolphin_T_t_ds_worker_group_R_ip_list;
+DROP PROCEDURE uc_dolphin_T_t_ds_worker_group_R_ip_list;

sql/upgrade/1.3.6_schema/mysql/dolphinscheduler_dml.sql

@@ -0,0 +1,16 @@
+/*
+ * 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.
+*/

sql/upgrade/1.3.6_schema/postgresql/dolphinscheduler_ddl.sql

@@ -0,0 +1,37 @@
+/*
+ * 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.
+*/
+
+-- uc_dolphin_T_t_ds_worker_group_A_ip_list
+delimiter d//
+CREATE OR REPLACE FUNCTION uc_dolphin_T_t_ds_worker_group_A_ip_list() RETURNS void AS $$
+BEGIN
+    IF EXISTS (SELECT 1 FROM information_schema.COLUMNS
+        WHERE TABLE_NAME='t_ds_worker_group'
+        AND COLUMN_NAME ='ip_list')
+    THEN
+        ALTER TABLE t_ds_worker_group RENAME ip_list TO addr_list;
+        ALTER TABLE t_ds_worker_group ALTER COLUMN addr_list type text;
+        ALTER TABLE t_ds_worker_group ALTER COLUMN name type varchar(256), ALTER COLUMN name SET NOT NULL;
+        ALTER TABLE t_ds_worker_group ADD CONSTRAINT name_unique UNIQUE (name);
+    END IF;
+END;
+$$ LANGUAGE plpgsql;
+d//
+
+delimiter ;
+SELECT uc_dolphin_T_t_ds_worker_group_A_ip_list();
+DROP FUNCTION IF EXISTS uc_dolphin_T_t_ds_worker_group_A_ip_list();

sql/upgrade/1.3.6_schema/postgresql/dolphinscheduler_dml.sql

@@ -0,0 +1,16 @@
+/*
+ * 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.
+*/