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

Revert "[Bug] [Resource] fix resource delete bug (#15003)" (#15020) 

This reverts commit d6a900667a.
This commit is contained in:
Kerwin 2023-10-12 17:28:11 +08:00 committed by GitHub
parent d6a900667a
commit 1f67e68400
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
10 changed files with 127 additions and 248 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/TaskDefinitionService.java
View File

@ -22,13 +22,11 @@ import org.apache.dolphinscheduler.api.dto.task.TaskFilterRequest;
import org.apache.dolphinscheduler.api.dto.task.TaskUpdateRequest;
import org.apache.dolphinscheduler.api.utils.PageInfo;
import org.apache.dolphinscheduler.api.utils.Result;
import org.apache.dolphinscheduler.common.enums.Flag;
import org.apache.dolphinscheduler.common.enums.ReleaseState;
import org.apache.dolphinscheduler.common.enums.TaskExecuteType;
import org.apache.dolphinscheduler.dao.entity.TaskDefinition;
import org.apache.dolphinscheduler.dao.entity.User;
import java.util.List;
import java.util.Map;
/**
@ -257,13 +255,4 @@ public interface TaskDefinitionService {
ReleaseState releaseState);
void deleteTaskByWorkflowDefinitionCode(long workflowDefinitionCode, int workflowDefinitionVersion);
/**
* queryAll TaskDefinition
*
* @param releaseState workFlow releaseState
* @param flag task flag
* @return TaskDefinition list
*/
List<TaskDefinition> queryAllTaskDefinitionByWorkFlowReleaseStateAndTaskFlag(ReleaseState releaseState, Flag flag);
}

168
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ProcessDefinitionServiceImpl.java
View File

@ -254,14 +254,14 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
/**
* create process definition
*
* @param loginUser login user
* @param projectCode project code
* @param name process definition name
* @param description description
* @param globalParams global params
* @param locations locations for nodes
* @param timeout timeout
* @param taskRelationJson relation json for nodes
* @param loginUser login user
* @param projectCode project code
* @param name process definition name
* @param description description
* @param globalParams global params
* @param locations locations for nodes
* @param timeout timeout
* @param taskRelationJson relation json for nodes
* @param taskDefinitionJson taskDefinitionJson
* @return create result code
*/
@ -342,7 +342,7 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
/**
* create single process definition
*
* @param loginUser login user
* @param loginUser login user
* @param workflowCreateRequest the new workflow object will be created
* @return New ProcessDefinition object created just now
*/
@ -493,7 +493,7 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
/**
* query process definition list
*
* @param loginUser login user
* @param loginUser login user
* @param projectCode project code
* @return definition list
*/
@ -516,7 +516,7 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
/**
* query process definition simple list
*
* @param loginUser login user
* @param loginUser login user
* @param projectCode project code
* @return definition simple list
*/
@ -547,12 +547,12 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
/**
* query process definition list paging
*
* @param loginUser login user
* @param loginUser login user
* @param projectCode project code
* @param searchVal search value
* @param userId user id
* @param pageNo page number
* @param pageSize page size
* @param searchVal search value
* @param userId user id
* @param pageNo page number
* @param pageSize page size
* @return process definition page
*/
@Override
@ -602,7 +602,7 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
/**
* Filter resource process definitions
*
* @param loginUser login user
* @param loginUser login user
* @param workflowFilterRequest workflow filter requests
* @return List process definition
*/
@ -642,9 +642,9 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
/**
* query detail of process definition
*
* @param loginUser login user
* @param loginUser login user
* @param projectCode project code
* @param code process definition code
* @param code process definition code
* @return process definition detail
*/
@Override
@ -673,7 +673,7 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
* query detail of process definition
*
* @param loginUser login user
* @param code process definition code
* @param code process definition code
* @return process definition detail
*/
@Override
@ -734,17 +734,17 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
/**
* update process definition
*
* @param loginUser login user
* @param projectCode project code
* @param name process definition name
* @param code process definition code
* @param description description
* @param globalParams global params
* @param locations locations for nodes
* @param timeout timeout
* @param taskRelationJson relation json for nodes
* @param loginUser login user
* @param projectCode project code
* @param name process definition name
* @param code process definition code
* @param description description
* @param globalParams global params
* @param locations locations for nodes
* @param timeout timeout
* @param taskRelationJson relation json for nodes
* @param taskDefinitionJson taskDefinitionJson
* @param otherParamsJson otherParamsJson handle other params
* @param otherParamsJson otherParamsJson handle other params
* @return update result code
*/
@Override
@ -811,11 +811,11 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
/**
* Task want to delete whether used in other task, should throw exception when have be used.
* <p>
*
* This function avoid delete task already dependencies by other tasks by accident.
*
* @param processDefinition ProcessDefinition you change task definition and task relation
* @param taskRelationList All the latest task relation list from process definition
* @param taskRelationList All the latest task relation list from process definition
*/
private void taskUsedInOtherTaskValid(ProcessDefinition processDefinition,
List<ProcessTaskRelationLog> taskRelationList) {
@ -923,9 +923,9 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
/**
* verify process definition name unique
*
* @param loginUser login user
* @param loginUser login user
* @param projectCode project code
* @param name name
* @param name name
* @return true if process definition name not exists, otherwise false
*/
@Override
@ -995,7 +995,7 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
/**
* Process definition want to delete whether used in other task, should throw exception when have be used.
* <p>
*
* This function avoid delete process definition already dependencies by other tasks by accident.
*
* @param processDefinition ProcessDefinition you change task definition and task relation
@ -1076,9 +1076,9 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
/**
* release process definition: online / offline
*
* @param loginUser login user
* @param projectCode project code
* @param code process definition code
* @param loginUser login user
* @param projectCode project code
* @param code process definition code
* @param releaseState release state
* @return release result code
*/
@ -1243,9 +1243,9 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
/**
* import process definition
*
* @param loginUser login user
* @param loginUser login user
* @param projectCode project code
* @param file process metadata json file
* @param file process metadata json file
* @return import process
*/
@Override
@ -1717,9 +1717,9 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
/**
* get task node details based on process definition
*
* @param loginUser loginUser
* @param loginUser loginUser
* @param projectCode project code
* @param code process definition code
* @param code process definition code
* @return task node list
*/
@Override
@ -1746,9 +1746,9 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
/**
* get task node details map based on process definition
*
* @param loginUser loginUser
* @param loginUser loginUser
* @param projectCode project code
* @param codes define codes
* @param codes define codes
* @return task node list
*/
@Override
@ -1799,7 +1799,7 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
/**
* query process definition all by project code
*
* @param loginUser loginUser
* @param loginUser loginUser
* @param projectCode project code
* @return process definitions in the project
*/
@ -1839,7 +1839,7 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
/**
* query process definition list by process definition code
*
* @param projectCode project code
* @param projectCode project code
* @param processDefinitionCode process definition code
* @return task definition list in the process definition
*/
@ -1875,8 +1875,8 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
* Encapsulates the TreeView structure
*
* @param projectCode project code
* @param code process definition code
* @param limit limit
* @param code process definition code
* @param limit limit
* @return tree view json data
*/
@Override
@ -2038,9 +2038,9 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
/**
* batch copy process definition
*
* @param loginUser loginUser
* @param projectCode projectCode
* @param codes processDefinitionCodes
* @param loginUser loginUser
* @param projectCode projectCode
* @param codes processDefinitionCodes
* @param targetProjectCode targetProjectCode
*/
@Override
@ -2062,10 +2062,9 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
/**
* batch move process definition
* Will be deleted
*
* @param loginUser loginUser
* @param projectCode projectCode
* @param codes processDefinitionCodes
* @param loginUser loginUser
* @param projectCode projectCode
* @param codes processDefinitionCodes
* @param targetProjectCode targetProjectCode
*/
@Override
@ -2239,9 +2238,8 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
/**
* get new Task name or Process name when copy or import operate
*
* @param originalName Task or Process original name
* @param suffix "_copy_" or "_import_"
* @param suffix "_copy_" or "_import_"
* @return new name
*/
public String getNewName(String originalName, String suffix) {
@ -2263,10 +2261,10 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
/**
* switch the defined process definition version
*
* @param loginUser login user
* @param loginUser login user
* @param projectCode project code
* @param code process definition code
* @param version the version user want to switch
* @param code process definition code
* @param version the version user want to switch
* @return switch process definition version result code
*/
@Override
@ -2317,11 +2315,11 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
/**
* check batch operate result
*
* @param srcProjectCode srcProjectCode
* @param srcProjectCode srcProjectCode
* @param targetProjectCode targetProjectCode
* @param result result
* @param result result
* @param failedProcessList failedProcessList
* @param isCopy isCopy
* @param isCopy isCopy
*/
private void checkBatchOperateResult(long srcProjectCode, long targetProjectCode,
Map<String, Object> result, List<String> failedProcessList, boolean isCopy) {
@ -2348,11 +2346,11 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
/**
* query the pagination versions info by one certain process definition code
*
* @param loginUser login user info to check auth
* @param loginUser login user info to check auth
* @param projectCode project code
* @param pageNo page number
* @param pageSize page size
* @param code process definition code
* @param pageNo page number
* @param pageSize page size
* @param code process definition code
* @return the pagination process definition versions info of the certain process definition
*/
@Override
@ -2384,10 +2382,10 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
/**
* delete one certain process definition by version number and process definition code
*
* @param loginUser login user info to check auth
* @param loginUser login user info to check auth
* @param projectCode project code
* @param code process definition code
* @param version version number
* @param code process definition code
* @param version version number
* @return delete result code
*/
@Override
@ -2566,16 +2564,16 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
/**
* update process definition basic info
*
* @param loginUser login user
* @param projectCode project code
* @param name process definition name
* @param code process definition code
* @param description description
* @param globalParams globalParams
* @param timeout timeout
* @param scheduleJson scheduleJson
* @param loginUser login user
* @param projectCode project code
* @param name process definition name
* @param code process definition code
* @param description description
* @param globalParams globalParams
* @param timeout timeout
* @param scheduleJson scheduleJson
* @param otherParamsJson otherParamsJson handle other params
* @param executionType executionType
* @param executionType executionType
* @return update result code
*/
@Override
@ -2681,9 +2679,9 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
/**
* update single resource workflow
*
* @param loginUser login user
* @param workflowCode workflow resource code want to update
* @param workflowUpdateRequest workflow update resource object
* @param loginUser login user
* @param workflowCode workflow resource code want to update
* @param workflowUpdateRequest workflow update resource object
* @return Process definition
*/
@Override
@ -2927,7 +2925,6 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
/**
* save other relation
*
* @param loginUser
* @param processDefinition
* @param result
@ -2941,7 +2938,6 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
/**
* get Json String
*
* @param loginUser
* @param processDefinition
* @return Json String
@ -2953,8 +2949,7 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
/**
* view process variables
*
* @param loginUser login user
* @param loginUser login user
* @param projectCode project code
* @param code process definition code
* @return variables data
@ -3037,7 +3032,6 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
/**
* delete other relation
*
* @param project
* @param result
* @param processDefinition

119
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ResourcesServiceImpl.java
View File

@ -35,24 +35,22 @@ import org.apache.dolphinscheduler.api.enums.Status;
import org.apache.dolphinscheduler.api.exceptions.ServiceException;
import org.apache.dolphinscheduler.api.metrics.ApiServerMetrics;
import org.apache.dolphinscheduler.api.service.ResourcesService;
import org.apache.dolphinscheduler.api.service.TaskDefinitionService;
import org.apache.dolphinscheduler.api.utils.PageInfo;
import org.apache.dolphinscheduler.api.utils.RegexUtils;
import org.apache.dolphinscheduler.api.utils.Result;
import org.apache.dolphinscheduler.common.constants.Constants;
import org.apache.dolphinscheduler.common.enums.AuthorizationType;
import org.apache.dolphinscheduler.common.enums.Flag;
import org.apache.dolphinscheduler.common.enums.ProgramType;
import org.apache.dolphinscheduler.common.enums.ReleaseState;
import org.apache.dolphinscheduler.common.enums.ResUploadType;
import org.apache.dolphinscheduler.common.utils.FileUtils;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.common.utils.PropertyUtils;
import org.apache.dolphinscheduler.dao.entity.Resource;
import org.apache.dolphinscheduler.dao.entity.TaskDefinition;
import org.apache.dolphinscheduler.dao.entity.Tenant;
import org.apache.dolphinscheduler.dao.entity.UdfFunc;
import org.apache.dolphinscheduler.dao.entity.User;
import org.apache.dolphinscheduler.dao.mapper.ProcessDefinitionLogMapper;
import org.apache.dolphinscheduler.dao.mapper.ProcessTaskRelationMapper;
import org.apache.dolphinscheduler.dao.mapper.ResourceMapper;
import org.apache.dolphinscheduler.dao.mapper.ResourceUserMapper;
import org.apache.dolphinscheduler.dao.mapper.TenantMapper;
@ -120,7 +118,10 @@ public class ResourcesServiceImpl extends BaseServiceImpl implements ResourcesSe
private ResourceUserMapper resourceUserMapper;
@Autowired
private TaskDefinitionService taskDefinitionService;
private ProcessDefinitionLogMapper processDefinitionLogMapper;
@Autowired
private ProcessTaskRelationMapper processTaskRelationMapper;
@Autowired(required = false)
private StorageOperate storageOperate;
@ -128,11 +129,11 @@ public class ResourcesServiceImpl extends BaseServiceImpl implements ResourcesSe
/**
* create directory
*
* @param loginUser login user
* @param name alias
* @param type type
* @param pid parent id
* @param currentDir current directory
* @param loginUser login user
* @param name alias
* @param type type
* @param pid parent id
* @param currentDir current directory
* @return create directory result
*/
@Override
@ -270,7 +271,7 @@ public class ResourcesServiceImpl extends BaseServiceImpl implements ResourcesSe
* update the folder's size of the resource
*
* @param resource the current resource
* @param size size
* @param size size
*/
private void updateParentResourceSize(Resource resource, long size) {
if (resource.getSize() > 0) {
@ -315,13 +316,13 @@ public class ResourcesServiceImpl extends BaseServiceImpl implements ResourcesSe
/**
* update resource
*
* @param loginUser login user
* @param loginUser login user
* @param resourceFullName resource full name
* @param resTenantCode tenantCode in the request field "resTenantCode" for tenant code owning the resource,
* can be different from the login user in the case of logging in as admin users.
* @param name name
* @param type resource type
* @param file resource file
* @param resTenantCode tenantCode in the request field "resTenantCode" for tenant code owning the resource,
* can be different from the login user in the case of logging in as admin users.
* @param name name
* @param type resource type
* @param file resource file
* @return update result code
*/
@Override
@ -409,11 +410,6 @@ public class ResourcesServiceImpl extends BaseServiceImpl implements ResourcesSe
// if name unchanged, return directly without moving on HDFS
if (originResourceName.equals(name) && file == null) {
return result;
} else {
result = verifyResourceExistsOnlineUsage(resourceFullName);
if (result.isFailed()) {
return result;
}
}
if (file != null) {
@ -513,14 +509,14 @@ public class ResourcesServiceImpl extends BaseServiceImpl implements ResourcesSe
/**
* query resources list paging
*
* @param loginUser login user
* @param fullName resource full name
* @param loginUser login user
* @param fullName resource full name
* @param resTenantCode tenantCode in the request field "resTenantCode" for tenant code owning the resource,
* can be different from the login user in the case of logging in as admin users.
* @param type resource type
* @param searchVal search value
* @param pageNo page number
* @param pageSize page size
* @param type resource type
* @param searchVal search value
* @param pageNo page number
* @param pageSize page size
* @return resource list page
*/
@Override
@ -822,7 +818,7 @@ public class ResourcesServiceImpl extends BaseServiceImpl implements ResourcesSe
/**
* transform resource object into StorageEntity object
*
* @param resource a resource object
* @param resource a resource object
* @return a storageEntity object
*/
private StorageEntity createStorageEntityBasedOnResource(Resource resource) {
@ -841,8 +837,8 @@ public class ResourcesServiceImpl extends BaseServiceImpl implements ResourcesSe
/**
* delete resource
*
* @param loginUser login user
* @param fullName resource full name
* @param loginUser login user
* @param fullName resource full name
* @param resTenantCode tenantCode in the request field "resTenantCode" for tenant code owning the resource,
* can be different from the login user in the case of logging in as admin users.
* @return delete result code
@ -903,41 +899,14 @@ public class ResourcesServiceImpl extends BaseServiceImpl implements ResourcesSe
}
}
result = verifyResourceExistsOnlineUsage(fullName);
if (result.isFailed()) {
return result;
}
// delete file on hdfs,S3
storageOperate.delete(fullName, allChildren, true);
putMsg(result, Status.SUCCESS);
return result;
}
/**
* validate whether usage resource
*
* @param fullName fullName
* @return {@link Result}
*/
private Result verifyResourceExistsOnlineUsage(String fullName) {
Result<Object> result = new Result<>();
putMsg(result, Status.SUCCESS);
// query all online work flow task
List<TaskDefinition> taskDefinitions = taskDefinitionService
.queryAllTaskDefinitionByWorkFlowReleaseStateAndTaskFlag(ReleaseState.ONLINE, Flag.YES);
if (CollectionUtils.isNotEmpty(taskDefinitions)) {
for (TaskDefinition taskDefinition : taskDefinitions) {
if (taskDefinition.getTaskParamResourceNames().contains(fullName)) {
putMsg(result, Status.RESOURCE_IS_USED);
break;
}
}
}
return result;
}
private String RemoveResourceFromResourceList(String stringToDelete, String taskParameter, boolean isDir) {
Map<String, Object> taskParameters = JSONUtils.parseObject(
taskParameter,
@ -990,8 +959,8 @@ public class ResourcesServiceImpl extends BaseServiceImpl implements ResourcesSe
/**
* verify resource by full name or pid and type
*
* @param fileName resource file name
* @param type resource type
* @param fileName resource file name
* @param type resource type
* @param resTenantCode tenantCode in the request field "resTenantCode" for tenant code owning the resource,
* can be different from the login user in the case of logging in as admin users.
* @return true if the resource full name or pid not exists, otherwise return false
@ -1041,8 +1010,7 @@ public class ResourcesServiceImpl extends BaseServiceImpl implements ResourcesSe
/**
* get resource by id
*
* @param fullName resource full name
* @param fullName resource full name
* @param resTenantCode tenantCode in the request field "resTenantCode" for tenant code owning the resource,
* can be different from the login user in the case of logging in as admin users.
* @return resource
@ -1089,10 +1057,10 @@ public class ResourcesServiceImpl extends BaseServiceImpl implements ResourcesSe
/**
* view resource file online
*
* @param fullName resource fullName
* @param resTenantCode owner's tenant code of the resource
* @param skipLineNum skip line number
* @param limit limit
* @param fullName resource fullName
* @param resTenantCode owner's tenant code of the resource
* @param skipLineNum skip line number
* @param limit limit
* @return resource content
*/
@Override
@ -1285,10 +1253,10 @@ public class ResourcesServiceImpl extends BaseServiceImpl implements ResourcesSe
/**
* updateProcessInstance resource
*
* @param fullName resource full name
* @param fullName resource full name
* @param resTenantCode tenantCode in the request field "resTenantCode" for tenant code owning the resource,
* can be different from the login user in the case of logging in as admin users.
* @param content content
* @param content content
* @return update result cod
*/
@Override
@ -1349,9 +1317,9 @@ public class ResourcesServiceImpl extends BaseServiceImpl implements ResourcesSe
}
/**
* @param fullName resource full name
* @param tenantCode tenant code
* @param content content
* @param fullName resource full name
* @param tenantCode tenant code
* @param content content
* @return result
*/
private Result<Object> uploadContentToStorage(User loginUser, String fullName, String tenantCode, String content) {
@ -1397,7 +1365,6 @@ public class ResourcesServiceImpl extends BaseServiceImpl implements ResourcesSe
/**
* download file
*
* @return resource content
*/
@Override
@ -1784,10 +1751,10 @@ public class ResourcesServiceImpl extends BaseServiceImpl implements ResourcesSe
/**
* check permission by comparing login user's tenantCode with tenantCode in the request
*
* @param isAdmin is the login user admin
* @param isAdmin is the login user admin
* @param userTenantCode loginUser's tenantCode
* @param resTenantCode tenantCode in the request field "resTenantCode" for tenant code owning the resource,
* can be different from the login user in the case of logging in as admin users.
* @param resTenantCode tenantCode in the request field "resTenantCode" for tenant code owning the resource,
* can be different from the login user in the case of logging in as admin users.
* @return isValid
*/
private boolean isUserTenantValid(boolean isAdmin, String userTenantCode,

6
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskDefinitionServiceImpl.java
View File

@ -1326,10 +1326,4 @@ public class TaskDefinitionServiceImpl extends BaseServiceImpl implements TaskDe
// delete task workflow relation
processTaskRelationService.deleteByWorkflowDefinitionCode(workflowDefinitionCode, workflowDefinitionVersion);
}
@Override
public List<TaskDefinition> queryAllTaskDefinitionByWorkFlowReleaseStateAndTaskFlag(ReleaseState releaseState,
Flag flag) {
return taskDefinitionMapper.queryAllTaskDefinitionByWorkFlowReleaseStateAndTaskFlag(releaseState, flag);
}
}

9
dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/ResourcesServiceTest.java
View File

@ -28,16 +28,14 @@ import org.apache.dolphinscheduler.api.service.impl.ResourcesServiceImpl;
import org.apache.dolphinscheduler.api.utils.PageInfo;
import org.apache.dolphinscheduler.api.utils.Result;
import org.apache.dolphinscheduler.common.constants.Constants;
import org.apache.dolphinscheduler.common.enums.Flag;
import org.apache.dolphinscheduler.common.enums.ReleaseState;
import org.apache.dolphinscheduler.common.enums.UserType;
import org.apache.dolphinscheduler.common.utils.FileUtils;
import org.apache.dolphinscheduler.common.utils.PropertyUtils;
import org.apache.dolphinscheduler.dao.entity.Resource;
import org.apache.dolphinscheduler.dao.entity.TaskDefinition;
import org.apache.dolphinscheduler.dao.entity.Tenant;
import org.apache.dolphinscheduler.dao.entity.UdfFunc;
import org.apache.dolphinscheduler.dao.entity.User;
import org.apache.dolphinscheduler.dao.mapper.ProcessDefinitionMapper;
import org.apache.dolphinscheduler.dao.mapper.ResourceMapper;
import org.apache.dolphinscheduler.dao.mapper.ResourceUserMapper;
import org.apache.dolphinscheduler.dao.mapper.TenantMapper;
@ -111,7 +109,7 @@ public class ResourcesServiceTest {
private UdfFuncMapper udfFunctionMapper;
@Mock
private TaskDefinitionService taskDefinitionService;
private ProcessDefinitionMapper processDefinitionMapper;
@Mock
private ResourceUserMapper resourceUserMapper;
@ -409,9 +407,6 @@ public class ResourcesServiceTest {
Mockito.when(storageOperate.getFileStatus("/dolphinscheduler/123/resources/ResourcesServiceTest",
null, "123", null))
.thenReturn(getStorageEntityResource());
Mockito.when(taskDefinitionService.queryAllTaskDefinitionByWorkFlowReleaseStateAndTaskFlag(ReleaseState.OFFLINE,
Flag.YES)).thenReturn(Arrays.asList(new TaskDefinition()));
Result result = resourcesService.delete(loginUser, "/dolphinscheduler/123/resources/ResNotExist", "123");
logger.info(result.toString());
Assertions.assertEquals(Status.RESOURCE_NOT_EXIST.getMsg(), result.getMsg());

11
dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/TaskDefinitionServiceImplTest.java
View File

@ -762,15 +762,4 @@ public class TaskDefinitionServiceImplTest {
return processTaskRelationList;
}
@Test
public void testQueryAllTaskDefinitionByWorkFlowReleaseStateAndTaskFlag() {
List<TaskDefinition> tasks = Arrays.asList(new TaskDefinition(), new TaskDefinition());
Mockito.when(taskDefinitionMapper.queryAllTaskDefinitionByWorkFlowReleaseStateAndTaskFlag(ReleaseState.OFFLINE,
null)).thenReturn(tasks);
List<TaskDefinition> taskDefinitions = taskDefinitionService
.queryAllTaskDefinitionByWorkFlowReleaseStateAndTaskFlag(ReleaseState.OFFLINE, null);
Assertions.assertEquals(tasks, taskDefinitions);
}
}

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

@ -697,7 +697,7 @@ public final class Constants {
* schedule timezone
*/
public static final String SCHEDULE_TIMEZONE = "schedule_timezone";
public static final int RESOURCE_FULL_NAME_MAX_LENGTH = 255;
public static final int RESOURCE_FULL_NAME_MAX_LENGTH = 128;
/**
* tenant

17
dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/TaskDefinition.java
View File

@ -27,9 +27,7 @@ import org.apache.dolphinscheduler.plugin.task.api.enums.TaskTimeoutStrategy;
import org.apache.dolphinscheduler.plugin.task.api.model.Property;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
@ -261,21 +259,6 @@ public class TaskDefinition {
return taskParamMap;
}
/**
* get task param resource names
*
* @return {@link List}<{@link String}>
*/
public List<String> getTaskParamResourceNames() {
if (StringUtils.isNotEmpty(taskParams)) {
JsonNode resourceList = JSONUtils.parseObject(taskParams).findValue("resourceList");
if (resourceList != null && !resourceList.isNull()) {
return resourceList.findValuesAsText("resourceName");
}
}
return Collections.EMPTY_LIST;
}
public String getDependence() {
return JSONUtils.getNodeString(this.taskParams, Constants.DEPENDENCE);
}

12
dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/TaskDefinitionMapper.java
View File

@ -17,8 +17,6 @@
package org.apache.dolphinscheduler.dao.mapper;
import org.apache.dolphinscheduler.common.enums.Flag;
import org.apache.dolphinscheduler.common.enums.ReleaseState;
import org.apache.dolphinscheduler.common.enums.TaskExecuteType;
import org.apache.dolphinscheduler.dao.entity.DefinitionGroupByUser;
import org.apache.dolphinscheduler.dao.entity.TaskDefinition;
@ -160,14 +158,4 @@ public interface TaskDefinitionMapper extends BaseMapper<TaskDefinition> {
int deleteByBatchCodes(@Param("taskCodeList") List<Long> taskCodeList);
void deleteByWorkflowDefinitionCodeAndVersion(long workflowDefinitionCode, int workflowDefinitionVersion);
/**
* queryAll TaskDefinition
*
* @param releaseState release state
* @param flag task flag
* @return task list
*/
List<TaskDefinition> queryAllTaskDefinitionByWorkFlowReleaseStateAndTaskFlag(@Param("releaseState") ReleaseState releaseState,
@Param("taskFlag") Flag flag);
}

20
dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/TaskDefinitionMapper.xml
View File

@ -170,24 +170,4 @@
</where>
order by update_time desc, id asc
</select>
<select id="queryAllTaskDefinitionByWorkFlowReleaseStateAndTaskFlag" resultType="org.apache.dolphinscheduler.dao.entity.TaskDefinition">
SELECT
<include refid="baseSqlV2">
<property name="alias" value="td"/>
</include>
FROM t_ds_process_definition as pd
JOIN t_ds_process_task_relation as ptr on pd.code = ptr.process_definition_code and pd.version = ptr.process_definition_version
JOIN t_ds_task_definition as td on ptr.post_task_code = td.code and ptr.post_task_version = td.version
<where>
<if test="releaseState != null">
and pd.release_state = #{releaseState}
</if>
<if test="taskFlag != null">
and td.flag = #{taskFlag}
</if>
</where>
</select>
</mapper>