diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/TaskDefinitionService.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/TaskDefinitionService.java index 00ebbc64e5..087b433189 100644 --- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/TaskDefinitionService.java +++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/TaskDefinitionService.java @@ -22,11 +22,13 @@ 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; /** @@ -255,4 +257,13 @@ 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 queryAllTaskDefinitionByWorkFlowReleaseStateAndTaskFlag(ReleaseState releaseState, Flag flag); } diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ProcessDefinitionServiceImpl.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ProcessDefinitionServiceImpl.java index d98d7b75fc..af8459fe41 100644 --- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ProcessDefinitionServiceImpl.java +++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ProcessDefinitionServiceImpl.java @@ -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. - * + *

* 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 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. - * + *

* 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,9 +2062,10 @@ 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 @@ -2238,8 +2239,9 @@ 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) { @@ -2261,10 +2263,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 @@ -2315,11 +2317,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 result, List failedProcessList, boolean isCopy) { @@ -2346,11 +2348,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 @@ -2382,10 +2384,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 @@ -2564,16 +2566,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 @@ -2679,9 +2681,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 @@ -2925,6 +2927,7 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro /** * save other relation + * * @param loginUser * @param processDefinition * @param result @@ -2938,6 +2941,7 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro /** * get Json String + * * @param loginUser * @param processDefinition * @return Json String @@ -2949,7 +2953,8 @@ 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 @@ -3032,6 +3037,7 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro /** * delete other relation + * * @param project * @param result * @param processDefinition diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ResourcesServiceImpl.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ResourcesServiceImpl.java index b4a9bba80e..8831798d37 100644 --- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ResourcesServiceImpl.java +++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ResourcesServiceImpl.java @@ -35,22 +35,24 @@ 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; @@ -118,10 +120,7 @@ public class ResourcesServiceImpl extends BaseServiceImpl implements ResourcesSe private ResourceUserMapper resourceUserMapper; @Autowired - private ProcessDefinitionLogMapper processDefinitionLogMapper; - - @Autowired - private ProcessTaskRelationMapper processTaskRelationMapper; + private TaskDefinitionService taskDefinitionService; @Autowired(required = false) private StorageOperate storageOperate; @@ -129,11 +128,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 @@ -271,7 +270,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) { @@ -316,13 +315,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 @@ -410,6 +409,11 @@ 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) { @@ -509,14 +513,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 @@ -818,7 +822,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) { @@ -837,8 +841,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 @@ -899,14 +903,41 @@ 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 result = new Result<>(); + putMsg(result, Status.SUCCESS); + // query all online work flow task + List 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 taskParameters = JSONUtils.parseObject( taskParameter, @@ -959,8 +990,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 @@ -1010,7 +1041,8 @@ 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 @@ -1057,10 +1089,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 @@ -1253,10 +1285,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 @@ -1317,9 +1349,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 uploadContentToStorage(User loginUser, String fullName, String tenantCode, String content) { @@ -1365,6 +1397,7 @@ public class ResourcesServiceImpl extends BaseServiceImpl implements ResourcesSe /** * download file + * * @return resource content */ @Override @@ -1751,10 +1784,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, diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskDefinitionServiceImpl.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskDefinitionServiceImpl.java index 838744767a..6e5afa45a0 100644 --- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskDefinitionServiceImpl.java +++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskDefinitionServiceImpl.java @@ -1326,4 +1326,10 @@ public class TaskDefinitionServiceImpl extends BaseServiceImpl implements TaskDe // delete task workflow relation processTaskRelationService.deleteByWorkflowDefinitionCode(workflowDefinitionCode, workflowDefinitionVersion); } + + @Override + public List queryAllTaskDefinitionByWorkFlowReleaseStateAndTaskFlag(ReleaseState releaseState, + Flag flag) { + return taskDefinitionMapper.queryAllTaskDefinitionByWorkFlowReleaseStateAndTaskFlag(releaseState, flag); + } } diff --git a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/ResourcesServiceTest.java b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/ResourcesServiceTest.java index 1295a8f3f3..0e0036182e 100644 --- a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/ResourcesServiceTest.java +++ b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/ResourcesServiceTest.java @@ -28,14 +28,16 @@ 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; @@ -109,7 +111,7 @@ public class ResourcesServiceTest { private UdfFuncMapper udfFunctionMapper; @Mock - private ProcessDefinitionMapper processDefinitionMapper; + private TaskDefinitionService taskDefinitionService; @Mock private ResourceUserMapper resourceUserMapper; @@ -407,6 +409,9 @@ 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()); diff --git a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/TaskDefinitionServiceImplTest.java b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/TaskDefinitionServiceImplTest.java index 26091e7529..da33fbfe90 100644 --- a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/TaskDefinitionServiceImplTest.java +++ b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/TaskDefinitionServiceImplTest.java @@ -762,4 +762,15 @@ public class TaskDefinitionServiceImplTest { return processTaskRelationList; } + + @Test + public void testQueryAllTaskDefinitionByWorkFlowReleaseStateAndTaskFlag() { + List tasks = Arrays.asList(new TaskDefinition(), new TaskDefinition()); + Mockito.when(taskDefinitionMapper.queryAllTaskDefinitionByWorkFlowReleaseStateAndTaskFlag(ReleaseState.OFFLINE, + null)).thenReturn(tasks); + + List taskDefinitions = taskDefinitionService + .queryAllTaskDefinitionByWorkFlowReleaseStateAndTaskFlag(ReleaseState.OFFLINE, null); + Assertions.assertEquals(tasks, taskDefinitions); + } } diff --git a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/constants/Constants.java b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/constants/Constants.java index 9ec208e44c..8952ddd975 100644 --- a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/constants/Constants.java +++ b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/constants/Constants.java @@ -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 = 128; + public static final int RESOURCE_FULL_NAME_MAX_LENGTH = 255; /** * tenant diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/TaskDefinition.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/TaskDefinition.java index 18c3a405a7..551439adf6 100644 --- a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/TaskDefinition.java +++ b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/TaskDefinition.java @@ -27,7 +27,9 @@ 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; @@ -259,6 +261,21 @@ public class TaskDefinition { return taskParamMap; } + /** + * get task param resource names + * + * @return {@link List}<{@link String}> + */ + public List 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); } diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/TaskDefinitionMapper.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/TaskDefinitionMapper.java index 105cf04700..4b9e8e84a3 100644 --- a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/TaskDefinitionMapper.java +++ b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/TaskDefinitionMapper.java @@ -17,6 +17,8 @@ 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; @@ -158,4 +160,14 @@ public interface TaskDefinitionMapper extends BaseMapper { int deleteByBatchCodes(@Param("taskCodeList") List taskCodeList); void deleteByWorkflowDefinitionCodeAndVersion(long workflowDefinitionCode, int workflowDefinitionVersion); + + /** + * queryAll TaskDefinition + * + * @param releaseState release state + * @param flag task flag + * @return task list + */ + List queryAllTaskDefinitionByWorkFlowReleaseStateAndTaskFlag(@Param("releaseState") ReleaseState releaseState, + @Param("taskFlag") Flag flag); } diff --git a/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/TaskDefinitionMapper.xml b/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/TaskDefinitionMapper.xml index fbc9c793df..71ba473e88 100644 --- a/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/TaskDefinitionMapper.xml +++ b/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/TaskDefinitionMapper.xml @@ -170,4 +170,24 @@ order by update_time desc, id asc + + +