natas_hw
/
dolphinscheduler
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

[Improvement][API] ignore noNodeException when get worker groups (#4120) 

* ignore noNodeException when get worker groups

* add ut
This commit is contained in:
felix.wang 2020-11-30 10:39:13 +08:00 committed by GitHub
parent 763c65938d
commit b46044d592
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 71 additions and 47 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -14,8 +14,11 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dolphinscheduler.api.service;
import static org.apache.dolphinscheduler.common.Constants.DEFAULT_WORKER_GROUP;
import org.apache.dolphinscheduler.api.enums.Status;
import org.apache.dolphinscheduler.api.utils.PageInfo;
import org.apache.dolphinscheduler.common.Constants;
@ -26,11 +29,17 @@ 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.service.zk.ZookeeperCachedOperator;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.util.*;
import java.util.stream.Collectors;
/**
* work group service
@ -38,14 +47,11 @@ import java.util.stream.Collectors;
@Service
public class WorkerGroupService extends BaseService {
@Autowired
ProcessInstanceMapper processInstanceMapper;
private static final String NO_NODE_EXCEPTION_REGEX = "KeeperException$NoNodeException";
@Autowired
protected ZookeeperCachedOperator zookeeperCachedOperator;
@Autowired
ProcessInstanceMapper processInstanceMapper;
/**
* query worker group paging
@ -56,7 +62,7 @@ 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;
@ -72,20 +78,20 @@ public class WorkerGroupService extends BaseService {
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){
if (searchValDataList.size() < pageSize) {
toIndex = (pageNo - 1) * pageSize + searchValDataList.size();
}
resultDataList = searchValDataList.subList(fromIndex, toIndex);
@ -100,14 +106,12 @@ 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);
@ -120,30 +124,46 @@ public class WorkerGroupService extends BaseService {
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);
String workerPath = zookeeperCachedOperator.getZookeeperConfig().getDsRoot() + Constants.ZOOKEEPER_DOLPHINSCHEDULER_WORKERS;
List<WorkerGroup> workerGroups = new ArrayList<>();
List<String> workerGroupList;
try {
workerGroupList = zookeeperCachedOperator.getChildrenKeys(workerPath);
} catch (Exception e) {
if (e.getMessage().contains(NO_NODE_EXCEPTION_REGEX)) {
if (isPaging) {
return workerGroups;
} else {
//ignore noNodeException return Default
WorkerGroup wg = new WorkerGroup();
wg.setName(DEFAULT_WORKER_GROUP);
workerGroups.add(wg);
return workerGroups;
}
} else {
throw e;
}
}
// available workerGroup list
List<String> availableWorkerGroupList = new ArrayList<>();
List<WorkerGroup> workerGroups = new ArrayList<>();
for (String workerGroup : workerGroupList){
String workerGroupPath= workerPath + "/" + workerGroup;
for (String workerGroup : workerGroupList) {
String workerGroupPath = workerPath + "/" + workerGroup;
List<String> childrenNodes = zookeeperCachedOperator.getChildrenKeys(workerGroupPath);
if (CollectionUtils.isNotEmpty(childrenNodes)){
if (CollectionUtils.isNotEmpty(childrenNodes)) {
availableWorkerGroupList.add(workerGroup);
WorkerGroup wg = new WorkerGroup();
wg.setName(workerGroup);
if (isPaging){
if (isPaging) {
wg.setIpList(childrenNodes);
String registeredIpValue = zookeeperCachedOperator.get(workerGroupPath + "/" + childrenNodes.get(0));
wg.setCreateTime(DateUtils.stringToDate(registeredIpValue.split(",")[6]));

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

@ -14,20 +14,24 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
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.common.Constants;
import org.apache.dolphinscheduler.common.enums.UserType;
import org.apache.dolphinscheduler.common.utils.CollectionUtils;
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.service.zk.ZookeeperCachedOperator;
import org.apache.dolphinscheduler.service.zk.ZookeeperConfig;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
@ -35,16 +39,10 @@ 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 {
@ -59,14 +57,13 @@ public class WorkerGroupServiceTest {
@Mock
private ZookeeperCachedOperator zookeeperCachedOperator;
@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");
@ -83,19 +80,18 @@ public class WorkerGroupServiceTest {
}
/**
* 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 {
Map<String, Object> result = workerGroupService.queryAllGroup();
@ -103,16 +99,24 @@ public class WorkerGroupServiceTest {
Assert.assertEquals(workerGroups.size(), 1);
}
/**
* 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 testQueryAllGroupWithNoNodeException() {
String workerPath = zookeeperCachedOperator.getZookeeperConfig().getDsRoot() + Constants.ZOOKEEPER_DOLPHINSCHEDULER_WORKERS;
Mockito.when(zookeeperCachedOperator.getChildrenKeys(workerPath)).thenThrow(new RuntimeException("KeeperException$NoNodeException"));
Map<String, Object> result = workerGroupService.queryAllGroup();
Set<String> workerGroups = (Set<String>) result.get(Constants.DATA_LIST);
Assert.assertEquals(1, workerGroups.size());
Assert.assertEquals("default", workerGroups.toArray()[0]);
}
}