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

[Optimization] Config module resource relation fix. (#10718) 

* Config module resource relation fix.

* Boolean judge fix

* e2e rerun

* e2e rerun

* e2e rerun

* ArrayList fix

* change variable name.

* Add transaction processing on create data resource method
This commit is contained in:
WangJPLeo 2022-07-05 15:51:23 +08:00 committed by GitHub
parent 91283e59cf
commit d2fe16d252
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
15 changed files with 63 additions and 40 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/AlertGroupServiceImpl.java
View File

@ -160,6 +160,7 @@ public class AlertGroupServiceImpl extends BaseServiceImpl implements AlertGroup
* @return create result code
*/
@Override
@Transactional(rollbackFor = Exception.class)
public Map<String, Object> createAlertgroup(User loginUser, String groupName, String desc, String alertInstanceIds) {
Map<String, Object> result = new HashMap<>();
//only admin can operate
@ -191,6 +192,8 @@ public class AlertGroupServiceImpl extends BaseServiceImpl implements AlertGroup
} catch (DuplicateKeyException ex) {
logger.error("Create alert group error.", ex);
putMsg(result, Status.ALERT_GROUP_EXIST);
} catch (RuntimeException e) {
throw new RuntimeException(e.getMessage());
}
return result;

3
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/DataSourceServiceImpl.java
View File

@ -100,6 +100,7 @@ public class DataSourceServiceImpl extends BaseServiceImpl implements DataSource
* @return create result code
*/
@Override
@Transactional(rollbackFor = Exception.class)
public Result<Object> createDataSource(User loginUser, BaseDataSourceParamDTO datasourceParam) {
DataSourceUtils.checkDatasourceParam(datasourceParam);
Result<Object> result = new Result<>();
@ -139,6 +140,8 @@ public class DataSourceServiceImpl extends BaseServiceImpl implements DataSource
} catch (DuplicateKeyException ex) {
logger.error("Create datasource error.", ex);
putMsg(result, Status.DATASOURCE_EXIST);
} catch (RuntimeException e) {
throw new RuntimeException(e.getMessage());
}
return result;

2
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/EnvironmentServiceImpl.java
View File

@ -92,8 +92,8 @@ public class EnvironmentServiceImpl extends BaseServiceImpl implements Environme
* @param desc environment desc
* @param workerGroups worker groups
*/
@Transactional(rollbackFor = RuntimeException.class)
@Override
@Transactional(rollbackFor = Exception.class)
public Map<String, Object> createEnvironment(User loginUser, String name, String config, String desc, String workerGroups) {
Map<String, Object> result = new HashMap<>();
if (!canOperatorPermissions(loginUser, null, AuthorizationType.ENVIRONMENT, ENVIRONMENT_CREATE)) {

2
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ProjectServiceImpl.java
View File

@ -41,6 +41,7 @@ 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;
import java.util.ArrayList;
import java.util.Collections;
@ -86,6 +87,7 @@ public class ProjectServiceImpl extends BaseServiceImpl implements ProjectServic
* @return returns an error if it exists
*/
@Override
@Transactional(rollbackFor = Exception.class)
public Map<String, Object> createProject(User loginUser, String name, String desc) {
Map<String, Object> result = checkDesc(desc);

28
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/QueueServiceImpl.java
View File

@ -17,22 +17,26 @@
package org.apache.dolphinscheduler.api.service.impl;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import org.apache.commons.lang3.StringUtils;
import org.apache.dolphinscheduler.api.enums.Status;
import org.apache.dolphinscheduler.api.service.QueueService;
import org.apache.dolphinscheduler.api.utils.PageInfo;
import org.apache.dolphinscheduler.api.utils.Result;
import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.common.enums.AuthorizationType;
import org.apache.dolphinscheduler.common.enums.UserType;
import org.apache.dolphinscheduler.dao.entity.Queue;
import org.apache.dolphinscheduler.dao.entity.User;
import org.apache.dolphinscheduler.dao.mapper.QueueMapper;
import org.apache.dolphinscheduler.dao.mapper.UserMapper;
import org.apache.commons.lang3.StringUtils;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
@ -41,11 +45,10 @@ 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;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import static org.apache.dolphinscheduler.api.constants.ApiFuncIdentificationConstant.*;
import static org.apache.dolphinscheduler.api.constants.ApiFuncIdentificationConstant.YARN_QUEUE_CREATE;
import static org.apache.dolphinscheduler.api.constants.ApiFuncIdentificationConstant.YARN_QUEUE_UPDATE;
/**
* queue service impl
@ -71,15 +74,13 @@ public class QueueServiceImpl extends BaseServiceImpl implements QueueService {
public Map<String, Object> queryList(User loginUser) {
Map<String, Object> result = new HashMap<>();
Set<Integer> ids = resourcePermissionCheckService.userOwnedResourceIdsAcquisition(AuthorizationType.QUEUE, loginUser.getId(), logger);
if (ids.isEmpty()) {
result.put(Constants.DATA_LIST, Collections.emptyList());
putMsg(result, Status.SUCCESS);
return result;
if (loginUser.getUserType().equals(UserType.GENERAL_USER)) {
ids = ids.isEmpty() ? new HashSet<>() : ids;
ids.add(Constants.DEFAULT_QUEUE_ID);
}
List<Queue> queueList = queueMapper.selectBatchIds(ids);
result.put(Constants.DATA_LIST, queueList);
putMsg(result, Status.SUCCESS);
return result;
}
@ -103,7 +104,7 @@ public class QueueServiceImpl extends BaseServiceImpl implements QueueService {
return result;
}
Page<Queue> page = new Page<>(pageNo, pageSize);
IPage<Queue> queueList = queueMapper.queryQueuePaging(page, searchVal);
IPage<Queue> queueList = queueMapper.queryQueuePaging(page, new ArrayList<>(ids), searchVal);
Integer count = (int) queueList.getTotal();
pageInfo.setTotal(count);
pageInfo.setTotalList(queueList.getRecords());
@ -122,6 +123,7 @@ public class QueueServiceImpl extends BaseServiceImpl implements QueueService {
* @return create result
*/
@Override
@Transactional(rollbackFor = Exception.class)
public Map<String, Object> createQueue(User loginUser, String queue, String queueName) {
Map<String, Object> result = new HashMap<>();
if (!canOperatorPermissions(loginUser,null, AuthorizationType.QUEUE,YARN_QUEUE_CREATE)) {
@ -160,7 +162,7 @@ public class QueueServiceImpl extends BaseServiceImpl implements QueueService {
queueMapper.insert(queueObj);
result.put(Constants.DATA_LIST, queueObj);
putMsg(result, Status.SUCCESS);
permissionPostHandle(AuthorizationType.QUEUE, loginUser.getId(), Collections.singletonList(queueObj.getId()), logger);
return result;
}

2
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskGroupServiceImpl.java
View File

@ -39,6 +39,7 @@ 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;
import java.util.ArrayList;
import java.util.Collections;
@ -78,6 +79,7 @@ public class TaskGroupServiceImpl extends BaseServiceImpl implements TaskGroupSe
* @return the result code and msg
*/
@Override
@Transactional(rollbackFor = Exception.class)
public Map<String, Object> createTaskGroup(User loginUser, Long projectCode, String name, String description, int groupSize) {
Map<String, Object> result = new HashMap<>();

2
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TenantServiceImpl.java
View File

@ -46,6 +46,7 @@ import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
@ -132,6 +133,7 @@ public class TenantServiceImpl extends BaseServiceImpl implements TenantService
if (PropertyUtils.getResUploadStartupState()) {
storageOperate.createTenantDirIfNotExists(tenantCode);
}
permissionPostHandle(AuthorizationType.TENANT, loginUser.getId(), Collections.singletonList(tenant.getId()), logger);
result.put(Constants.DATA_LIST, tenant);
putMsg(result, Status.SUCCESS);
return result;

1
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/UdfFuncServiceImpl.java
View File

@ -77,6 +77,7 @@ public class UdfFuncServiceImpl extends BaseServiceImpl implements UdfFuncServic
* @return create result code
*/
@Override
@Transactional(rollbackFor = Exception.class)
public Result<Object> createUdfFunction(User loginUser,
String funcName,
String className,

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

@ -84,6 +84,7 @@ public class WorkerGroupServiceImpl extends BaseServiceImpl implements WorkerGro
* @return create or update result code
*/
@Override
@Transactional(rollbackFor = Exception.class)
public Map<String, Object> saveWorkerGroup(User loginUser, int id, String name, String addrList) {
Map<String, Object> result = new HashMap<>();
if (!canOperatorPermissions(loginUser,null, AuthorizationType.WORKER_GROUP, WORKER_GROUP_CREATE)) {
@ -191,14 +192,12 @@ public class WorkerGroupServiceImpl extends BaseServiceImpl implements WorkerGro
int toIndex = (pageNo - 1) * pageSize + pageSize;
Result result = new Result();
List<WorkerGroup> workerGroups = new ArrayList<>();
List<WorkerGroup> workerGroups;
if (loginUser.getUserType().equals(UserType.ADMIN_USER)) {
workerGroups = getWorkerGroups(true);
workerGroups = getWorkerGroups(true, null);
} else {
Set<Integer> ids = resourcePermissionCheckService.userOwnedResourceIdsAcquisition(AuthorizationType.WORKER_GROUP, loginUser.getId(), logger);
if (!ids.isEmpty()) {
workerGroups = workerGroupMapper.selectBatchIds(ids);
}
workerGroups = getWorkerGroups(true, ids.isEmpty() ? Collections.emptyList() : new ArrayList<>(ids));
}
List<WorkerGroup> resultDataList = new ArrayList<>();
int total = 0;
@ -244,15 +243,10 @@ public class WorkerGroupServiceImpl extends BaseServiceImpl implements WorkerGro
Map<String, Object> result = new HashMap<>();
List<WorkerGroup> workerGroups;
if (loginUser.getUserType().equals(UserType.ADMIN_USER)) {
workerGroups = getWorkerGroups(false);
workerGroups = getWorkerGroups(false, null);
} else {
Set<Integer> ids = resourcePermissionCheckService.userOwnedResourceIdsAcquisition(AuthorizationType.WORKER_GROUP, loginUser.getId(), logger);
if (ids.isEmpty()) {
result.put(Constants.DATA_LIST, Collections.emptyList());
putMsg(result, Status.SUCCESS);
return result;
}
workerGroups = workerGroupMapper.selectBatchIds(ids);
workerGroups = getWorkerGroups(false, ids.isEmpty() ? Collections.emptyList() : new ArrayList<>(ids));
}
List<String> availableWorkerGroupList = workerGroups.stream()
.map(WorkerGroup::getName)
@ -273,9 +267,14 @@ public class WorkerGroupServiceImpl extends BaseServiceImpl implements WorkerGro
* @param isPaging whether paging
* @return WorkerGroup list
*/
private List<WorkerGroup> getWorkerGroups(boolean isPaging) {
private List<WorkerGroup> getWorkerGroups(boolean isPaging, List<Integer> ids) {
// worker groups from database
List<WorkerGroup> workerGroups = workerGroupMapper.queryAllWorkerGroup();
List<WorkerGroup> workerGroups;
if (ids != null) {
workerGroups = ids.isEmpty() ? new ArrayList<>() : workerGroupMapper.selectBatchIds(ids);
} else {
workerGroups = workerGroupMapper.queryAllWorkerGroup();
}
// worker groups from zookeeper
String workerPath = Constants.REGISTRY_DOLPHINSCHEDULER_WORKERS;
Collection<String> workerGroupList = null;

14
dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/permission/ResourcePermissionCheckServiceTest.java
View File

@ -17,13 +17,19 @@
package org.apache.dolphinscheduler.api.permission;
import com.google.common.collect.Lists;
import org.apache.dolphinscheduler.common.enums.AuthorizationType;
import org.apache.dolphinscheduler.common.enums.UserType;
import org.apache.dolphinscheduler.dao.entity.Project;
import org.apache.dolphinscheduler.dao.entity.User;
import org.apache.dolphinscheduler.dao.mapper.ProjectMapper;
import org.apache.dolphinscheduler.service.process.ProcessService;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import org.junit.Assert;
import org.junit.Test;
import org.junit.runner.RunWith;
@ -34,11 +40,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationContext;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import com.google.common.collect.Lists;
/**
* permission service test

2
dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/QueueServiceTest.java
View File

@ -110,7 +110,7 @@ public class QueueServiceTest {
Set<Integer> ids = new HashSet<>();
ids.add(1);
Mockito.when(resourcePermissionCheckService.userOwnedResourceIdsAcquisition(AuthorizationType.QUEUE, getLoginUser().getId(), queueServiceImplLogger)).thenReturn(ids);
Mockito.when(queueMapper.queryQueuePaging(Mockito.any(Page.class), Mockito.eq(queueName))).thenReturn(page);
Mockito.when(queueMapper.queryQueuePaging(Mockito.any(Page.class), Mockito.anyList(), Mockito.eq(queueName))).thenReturn(page);
Result result = queueService.queryList(getLoginUser(), queueName, 1, 10);
logger.info(result.toString());
PageInfo<Queue> pageInfo = (PageInfo<Queue>) result.getData();

2
dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/Constants.java
View File

@ -833,4 +833,6 @@ public final class Constants {
public static final int USER_PASSWORD_MIN_LENGTH = 2;
public static final String FUNCTION_START_WITH = "$";
public static final Integer DEFAULT_QUEUE_ID = 1;
}

2
dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/QueueMapper.java
View File

@ -34,7 +34,7 @@ public interface QueueMapper extends BaseMapper<Queue> {
* @param searchVal searchVal
* @return queue IPage
*/
IPage<Queue> queryQueuePaging(IPage<Queue> page,
IPage<Queue> queryQueuePaging(IPage<Queue> page, @Param("ids")List<Integer> ids,
@Param("searchVal") String searchVal);
/**

6
dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/QueueMapper.xml
View File

@ -29,6 +29,12 @@
<if test="searchVal != null and searchVal != ''">
and queue_name like concat('%', #{searchVal}, '%')
</if>
<if test="ids != null and ids.size() > 0">
and id in
<foreach collection="ids" item="i" open="(" close=")" separator=",">
#{i}
</foreach>
</if>
order by update_time desc
</select>
<select id="queryAllQueueList" resultType="org.apache.dolphinscheduler.dao.entity.Queue">

7
dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/QueueMapperTest.java
View File

@ -20,6 +20,7 @@ package org.apache.dolphinscheduler.dao.mapper;
import org.apache.dolphinscheduler.dao.BaseDaoTest;
import org.apache.dolphinscheduler.dao.entity.Queue;
import java.util.Collections;
import java.util.Date;
import java.util.List;
@ -94,12 +95,10 @@ public class QueueMapperTest extends BaseDaoTest {
Queue queue = insertOne();
Page<Queue> page = new Page(1,3);
IPage<Queue> queueIPage= queueMapper.queryQueuePaging(page,
null);
IPage<Queue> queueIPage= queueMapper.queryQueuePaging(page, Collections.singletonList(queue.getId()), null);
Assert.assertNotEquals(queueIPage.getTotal(), 0);
queueIPage= queueMapper.queryQueuePaging(page,
queue.getQueueName());
queueIPage= queueMapper.queryQueuePaging(page, Collections.singletonList(queue.getId()), queue.getQueueName());
Assert.assertNotEquals(queueIPage.getTotal(), 0);
}