Browse Source

[fix] update queue and tenant with same code error (#10877)

in ae6aa53f I use the same logic test to create or update
tenant and queue which failed create with the same value,
this patch will fix it. and also I use the constructor to reduce
the new object logic here

(cherry picked from commit a6154220dc)
3.0.0/version-upgrade
Jiajie Zhong 2 years ago
parent
commit
c67f9ee85d
  1. 113
      dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/QueueServiceImpl.java
  2. 92
      dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TenantServiceImpl.java
  3. 11
      dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/QueueServiceTest.java
  4. 6
      dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/TenantServiceTest.java
  5. 20
      dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/Queue.java
  6. 21
      dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/Tenant.java

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

@ -17,9 +17,9 @@
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 static org.apache.dolphinscheduler.api.constants.ApiFuncIdentificationConstant.YARN_QUEUE_CREATE;
import static org.apache.dolphinscheduler.api.constants.ApiFuncIdentificationConstant.YARN_QUEUE_UPDATE;
import org.apache.dolphinscheduler.api.enums.Status;
import org.apache.dolphinscheduler.api.exceptions.ServiceException;
import org.apache.dolphinscheduler.api.service.QueueService;
@ -33,9 +33,10 @@ 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;
@ -49,8 +50,8 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import static org.apache.dolphinscheduler.api.constants.ApiFuncIdentificationConstant.YARN_QUEUE_CREATE;
import static org.apache.dolphinscheduler.api.constants.ApiFuncIdentificationConstant.YARN_QUEUE_UPDATE;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
/**
* queue service impl
@ -67,41 +68,45 @@ public class QueueServiceImpl extends BaseServiceImpl implements QueueService {
private UserMapper userMapper;
/**
* Valid both queue and queueName when we want to create or update queue object
* Check the queue new object valid or not
*
* @param queue queue value
* @param queueName queue name
* @param queue The queue object want to create
*/
private void queueValid(String queue, String queueName) throws ServiceException {
if (StringUtils.isEmpty(queue)) {
private void createQueueValid(Queue queue) throws ServiceException {
if (StringUtils.isEmpty(queue.getQueue())) {
throw new ServiceException(Status.REQUEST_PARAMS_NOT_VALID_ERROR, Constants.QUEUE);
} else if (StringUtils.isEmpty(queueName)) {
} else if (StringUtils.isEmpty(queue.getQueueName())) {
throw new ServiceException(Status.REQUEST_PARAMS_NOT_VALID_ERROR, Constants.QUEUE_NAME);
} else if (checkQueueExist(queue)) {
throw new ServiceException(Status.QUEUE_VALUE_EXIST, queue);
} else if (checkQueueNameExist(queueName)) {
throw new ServiceException(Status.QUEUE_NAME_EXIST, queueName);
} else if (checkQueueExist(queue.getQueue())) {
throw new ServiceException(Status.QUEUE_VALUE_EXIST, queue.getQueue());
} else if (checkQueueNameExist(queue.getQueueName())) {
throw new ServiceException(Status.QUEUE_NAME_EXIST, queue.getQueueName());
}
}
/**
* Insert one single new Queue record to database
* Check queue update object valid or not
*
* @param queue queue value
* @param queueName queue name
* @return Queue
* @param existsQueue The exists queue object
* @param updateQueue The queue object want to update
*/
private Queue createObjToDB(String queue, String queueName) {
Queue queueObj = new Queue();
Date now = new Date();
queueObj.setQueue(queue);
queueObj.setQueueName(queueName);
queueObj.setCreateTime(now);
queueObj.setUpdateTime(now);
// save
queueMapper.insert(queueObj);
return queueObj;
private void updateQueueValid(Queue existsQueue, Queue updateQueue) throws ServiceException {
// Check the exists queue and the necessary of update operation, in not exist checker have to use updateQueue to avoid NPE
if (Objects.isNull(existsQueue)) {
throw new ServiceException(Status.QUEUE_NOT_EXIST, updateQueue.getQueue());
} else if (Objects.equals(existsQueue, updateQueue)) {
throw new ServiceException(Status.NEED_NOT_UPDATE_QUEUE);
}
// Check the update queue parameters
else if (StringUtils.isEmpty(updateQueue.getQueue())) {
throw new ServiceException(Status.REQUEST_PARAMS_NOT_VALID_ERROR, Constants.QUEUE);
} else if (StringUtils.isEmpty(updateQueue.getQueueName())) {
throw new ServiceException(Status.REQUEST_PARAMS_NOT_VALID_ERROR, Constants.QUEUE_NAME);
} else if (!Objects.equals(updateQueue.getQueue(), existsQueue.getQueue()) && checkQueueExist(updateQueue.getQueue())) {
throw new ServiceException(Status.QUEUE_VALUE_EXIST, updateQueue.getQueue());
} else if (!Objects.equals(updateQueue.getQueueName(), existsQueue.getQueueName()) && checkQueueNameExist(updateQueue.getQueueName())) {
throw new ServiceException(Status.QUEUE_NAME_EXIST, updateQueue.getQueueName());
}
}
/**
@ -169,11 +174,14 @@ public class QueueServiceImpl extends BaseServiceImpl implements QueueService {
if (isNotAdmin(loginUser, result)) {
return result;
}
queueValid(queue, queueName);
Queue newQueue = createObjToDB(queue, queueName);
result.put(Constants.DATA_LIST, newQueue);
Queue queueObj = new Queue(queueName, queue);
createQueueValid(queueObj);
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;
}
@ -193,35 +201,20 @@ public class QueueServiceImpl extends BaseServiceImpl implements QueueService {
return result;
}
queueValid(queue, queueName);
Queue queueObj = queueMapper.selectById(id);
if (Objects.isNull(queueObj)) {
throw new ServiceException(Status.QUEUE_NOT_EXIST, queue);
}
// whether queue value or queueName is changed
if (queue.equals(queueObj.getQueue()) && queueName.equals(queueObj.getQueueName())) {
throw new ServiceException(Status.NEED_NOT_UPDATE_QUEUE);
}
Queue updateQueue = new Queue(id, queueName, queue);
Queue existsQueue = queueMapper.selectById(id);
updateQueueValid(existsQueue, updateQueue);
// check old queue using by any user
if (checkIfQueueIsInUsing(queueObj.getQueueName(), queueName)) {
if (checkIfQueueIsInUsing(existsQueue.getQueueName(), updateQueue.getQueueName())) {
//update user related old queue
Integer relatedUserNums = userMapper.updateUserQueue(queueObj.getQueueName(), queueName);
Integer relatedUserNums = userMapper.updateUserQueue(existsQueue.getQueueName(), updateQueue.getQueueName());
logger.info("old queue have related {} user, exec update user success.", relatedUserNums);
}
// update queue
Date now = new Date();
queueObj.setQueue(queue);
queueObj.setQueueName(queueName);
queueObj.setUpdateTime(now);
queueMapper.updateById(queueObj);
queueMapper.updateById(updateQueue);
putMsg(result, Status.SUCCESS);
return result;
}
@ -235,7 +228,9 @@ public class QueueServiceImpl extends BaseServiceImpl implements QueueService {
@Override
public Result<Object> verifyQueue(String queue, String queueName) {
Result<Object> result = new Result<>();
queueValid(queue, queueName);
Queue queueValidator = new Queue(queueName, queue);
createQueueValid(queueValidator);
putMsg(result, Status.SUCCESS);
return result;
@ -287,10 +282,12 @@ public class QueueServiceImpl extends BaseServiceImpl implements QueueService {
*/
@Override
public Queue createQueueIfNotExists(String queue, String queueName) {
queueValid(queue, queueName);
Queue queueObj = new Queue(queueName, queue);
createQueueValid(queueObj);
Queue existsQueue = queueMapper.queryQueueName(queue, queueName);
if (Objects.isNull(existsQueue)) {
return createObjToDB(queue, queueName);
queueMapper.insert(queueObj);
return queueObj;
}
return existsQueue;
}

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

@ -86,44 +86,43 @@ public class TenantServiceImpl extends BaseServiceImpl implements TenantService
private StorageOperate storageOperate;
/**
* Valid tenantCode when we want to create or update tenant object
* Check the tenant new object valid or not
*
* @param tenantCode Tenant code of tenant object
* @return Optional of Status map
* @param tenant The tenant object want to create
*/
private void tenantCodeValid(String tenantCode) throws ServiceException {
Map<String, Object> result = new HashMap<>();
if (StringUtils.isEmpty(tenantCode)) {
throw new ServiceException(Status.REQUEST_PARAMS_NOT_VALID_ERROR, tenantCode);
} else if (StringUtils.length(tenantCode) > TENANT_FULL_NAME_MAX_LENGTH) {
private void createTenantValid(Tenant tenant) throws ServiceException {
if (StringUtils.isEmpty(tenant.getTenantCode())) {
throw new ServiceException(Status.REQUEST_PARAMS_NOT_VALID_ERROR, tenant.getTenantCode());
} else if (StringUtils.length(tenant.getTenantCode()) > TENANT_FULL_NAME_MAX_LENGTH) {
throw new ServiceException(Status.TENANT_FULL_NAME_TOO_LONG_ERROR);
} else if (!RegexUtils.isValidLinuxUserName(tenantCode)) {
} else if (!RegexUtils.isValidLinuxUserName(tenant.getTenantCode())) {
throw new ServiceException(Status.CHECK_OS_TENANT_CODE_ERROR);
} else if (checkTenantExists(tenantCode)) {
throw new ServiceException(Status.OS_TENANT_CODE_EXIST, tenantCode);
} else if (checkTenantExists(tenant.getTenantCode())) {
throw new ServiceException(Status.OS_TENANT_CODE_EXIST, tenant.getTenantCode());
}
}
/**
* Insert one single new Tenant record to database
* Check tenant update object valid or not
*
* @param tenantCode new Tenant object tenant code
* @param desc new Tenant object description
* @param queueId The Queue id of new Tenant object
* @return Tenant
* @param existsTenant The exists queue object
* @param updateTenant The queue object want to update
*/
private Tenant createObjToDB(String tenantCode, String desc, int queueId) {
Tenant tenant = new Tenant();
Date now = new Date();
tenant.setTenantCode(tenantCode);
tenant.setQueueId(queueId);
tenant.setDescription(desc);
tenant.setCreateTime(now);
tenant.setUpdateTime(now);
// save
tenantMapper.insert(tenant);
return tenant;
private void updateTenantValid(Tenant existsTenant, Tenant updateTenant) throws ServiceException {
// Check the exists tenant
if (Objects.isNull(existsTenant)) {
throw new ServiceException(Status.TENANT_NOT_EXIST);
}
// Check the update tenant parameters
else if (StringUtils.isEmpty(updateTenant.getTenantCode())) {
throw new ServiceException(Status.REQUEST_PARAMS_NOT_VALID_ERROR, updateTenant.getTenantCode());
} else if (StringUtils.length(updateTenant.getTenantCode()) > TENANT_FULL_NAME_MAX_LENGTH) {
throw new ServiceException(Status.TENANT_FULL_NAME_TOO_LONG_ERROR);
} else if (!RegexUtils.isValidLinuxUserName(updateTenant.getTenantCode())) {
throw new ServiceException(Status.CHECK_OS_TENANT_CODE_ERROR);
} else if (!Objects.equals(existsTenant.getTenantCode(), updateTenant.getTenantCode()) && checkTenantExists(updateTenant.getTenantCode())) {
throw new ServiceException(Status.OS_TENANT_CODE_EXIST, updateTenant.getTenantCode());
}
}
/**
@ -148,15 +147,16 @@ public class TenantServiceImpl extends BaseServiceImpl implements TenantService
throw new ServiceException(Status.USER_NO_OPERATION_PERM);
}
tenantCodeValid(tenantCode);
Tenant tenant = new Tenant(tenantCode, desc, queueId);
createTenantValid(tenant);
tenantMapper.insert(tenant);
Tenant newTenant = createObjToDB(tenantCode, desc, queueId);
// if storage startup
if (PropertyUtils.getResUploadStartupState()) {
storageOperate.createTenantDirIfNotExists(tenantCode);
}
permissionPostHandle(AuthorizationType.TENANT, loginUser.getId(), Collections.singletonList(newTenant.getId()), logger);
result.put(Constants.DATA_LIST, newTenant);
permissionPostHandle(AuthorizationType.TENANT, loginUser.getId(), Collections.singletonList(tenant.getId()), logger);
result.put(Constants.DATA_LIST, tenant);
putMsg(result, Status.SUCCESS);
return result;
}
@ -212,30 +212,18 @@ public class TenantServiceImpl extends BaseServiceImpl implements TenantService
throw new ServiceException(Status.USER_NO_OPERATION_PERM);
}
Tenant tenant = tenantMapper.queryById(id);
if (Objects.isNull(tenant)) {
throw new ServiceException(Status.TENANT_NOT_EXIST);
}
tenantCodeValid(tenantCode);
Tenant updateTenant = new Tenant(id, tenantCode, desc, queueId);
Tenant existsTenant = tenantMapper.queryById(id);
updateTenantValid(existsTenant, updateTenant);
// updateProcessInstance tenant
/**
* if the tenant code is modified, the original resource needs to be copied to the new tenant.
*/
if (!tenant.getTenantCode().equals(tenantCode) && PropertyUtils.getResUploadStartupState()) {
if (!Objects.equals(existsTenant.getTenantCode(), updateTenant.getTenantCode()) && PropertyUtils.getResUploadStartupState()) {
storageOperate.createTenantDirIfNotExists(tenantCode);
}
Date now = new Date();
if (queueId != 0) {
tenant.setQueueId(queueId);
}
tenant.setDescription(desc);
tenant.setUpdateTime(now);
tenantMapper.updateById(tenant);
tenantMapper.updateById(updateTenant);
putMsg(result, Status.SUCCESS);
return result;
@ -378,8 +366,10 @@ public class TenantServiceImpl extends BaseServiceImpl implements TenantService
return tenantMapper.queryByTenantCode(tenantCode);
}
tenantCodeValid(tenantCode);
Queue newQueue = queueService.createQueueIfNotExists(queue, queueName);
return createObjToDB(tenantCode, desc, newQueue.getId());
Tenant tenant = new Tenant(tenantCode, desc, newQueue.getId());
createTenantValid(tenant);
tenantMapper.insert(tenant);
return tenant;
}
}

11
dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/QueueServiceTest.java

@ -82,6 +82,7 @@ public class QueueServiceTest {
private static final String QUEUE_NAME = "queueName";
private static final String EXISTS = "exists";
private static final String NOT_EXISTS = "not_exists";
private static final String NOT_EXISTS_FINAL = "not_exists_final";
@Before
public void setUp() {
@ -170,6 +171,16 @@ public class QueueServiceTest {
Mockito.when(userMapper.existUser(Mockito.anyString())).thenReturn(false);
Map<String, Object> result = queueService.updateQueue(getLoginUser(), 1, NOT_EXISTS, NOT_EXISTS);
Assert.assertEquals(Status.SUCCESS.getCode(), ((Status) result.get(Constants.STATUS)).getCode());
// success update with same queue name
Mockito.when(queueMapper.existQueue(NOT_EXISTS_FINAL, null)).thenReturn(false);
result = queueService.updateQueue(getLoginUser(), 1, NOT_EXISTS_FINAL, NOT_EXISTS);
Assert.assertEquals(Status.SUCCESS.getCode(), ((Status) result.get(Constants.STATUS)).getCode());
// success update with same queue value
Mockito.when(queueMapper.existQueue(null, NOT_EXISTS_FINAL)).thenReturn(false);
result = queueService.updateQueue(getLoginUser(), 1, NOT_EXISTS, NOT_EXISTS_FINAL);
Assert.assertEquals(Status.SUCCESS.getCode(), ((Status) result.get(Constants.STATUS)).getCode());
}
@Test

6
dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/TenantServiceTest.java

@ -166,11 +166,17 @@ public class TenantServiceTest {
Mockito.when(resourcePermissionCheckService.operationPermissionCheck(AuthorizationType.TENANT, getLoginUser().getId(), TENANT_UPDATE, baseServiceLogger)).thenReturn(true);
Mockito.when(resourcePermissionCheckService.resourcePermissionCheck(AuthorizationType.TENANT, null, 0, baseServiceLogger)).thenReturn(true);
// update not exists tenant
Throwable exception = Assertions.assertThrows(ServiceException.class, () -> tenantService.updateTenant(getLoginUser(), 912222, tenantCode, 1, tenantDesc));
Assertions.assertEquals(Status.TENANT_NOT_EXIST.getMsg(), exception.getMessage());
// success
Map<String, Object> result = tenantService.updateTenant(getLoginUser(), 1, tenantCode, 1, tenantDesc);
Assert.assertEquals(Status.SUCCESS, result.get(Constants.STATUS));
// success update with same tenant code
result = tenantService.updateTenant(getLoginUser(), 1, tenantCode, 1, tenantDesc);
Assert.assertEquals(Status.SUCCESS, result.get(Constants.STATUS));
}
@Test

20
dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/Queue.java

@ -51,6 +51,26 @@ public class Queue {
*/
private Date updateTime;
public Queue() {
}
public Queue(String queueName, String queue) {
Date now = new Date();
this.queueName = queueName;
this.queue = queue;
this.createTime = now;
this.updateTime = now;
}
public Queue(int id, String queueName, String queue) {
Date now = new Date();
this.id = id;
this.queueName = queueName;
this.queue = queue;
this.createTime = now;
this.updateTime = now;
}
public int getId() {
return id;
}

21
dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/Tenant.java

@ -72,6 +72,27 @@ public class Tenant {
*/
private Date updateTime;
public Tenant() {
}
public Tenant(String tenantCode, String description, int queueId) {
Date now = new Date();
this.tenantCode = tenantCode;
this.description = description;
this.queueId = queueId;
this.createTime = now;
this.updateTime = now;
}
public Tenant(int id, String tenantCode, String description, int queueId) {
Date now = new Date();
this.id = id;
this.tenantCode = tenantCode;
this.description = description;
this.queueId = queueId;
this.createTime = now;
this.updateTime = now;
}
public int getId() {
return id;

Loading…
Cancel
Save