diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/TaskGroupQueueService.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/TaskGroupQueueService.java index a0d05c5b8d..72aa498241 100644 --- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/TaskGroupQueueService.java +++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/TaskGroupQueueService.java @@ -19,6 +19,7 @@ package org.apache.dolphinscheduler.api.service; import org.apache.dolphinscheduler.dao.entity.User; +import java.util.List; import java.util.Map; /** @@ -68,7 +69,13 @@ public interface TaskGroupQueueService { */ boolean deleteByTaskId(int taskId); + void deleteByTaskInstanceIds(List taskInstanceIds); + + void deleteByWorkflowInstanceId(Integer workflowInstanceId); + void forceStartTask(int queueId, int forceStart); void modifyPriority(Integer queueId, Integer priority); + + void deleteByTaskGroupIds(List taskGroupIds); } diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/TaskGroupService.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/TaskGroupService.java index 3987109733..db909da158 100644 --- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/TaskGroupService.java +++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/TaskGroupService.java @@ -139,4 +139,6 @@ public interface TaskGroupService { Map forceStartTask(User loginUser, int taskId); Map modifyPriority(User loginUser, Integer queueId, Integer priority); + + void deleteTaskGroupByProjectCode(long projectCode); } diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ExecutorServiceImpl.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ExecutorServiceImpl.java index d254b31481..2761b67ca3 100644 --- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ExecutorServiceImpl.java +++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ExecutorServiceImpl.java @@ -104,6 +104,7 @@ import java.util.stream.Collectors; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Lazy; import org.springframework.stereotype.Service; import com.fasterxml.jackson.core.type.TypeReference; @@ -135,6 +136,7 @@ public class ExecutorServiceImpl extends BaseServiceImpl implements ExecutorServ @Autowired private ProcessInstanceMapper processInstanceMapper; + @Lazy() @Autowired private ProcessService processService; diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ProjectServiceImpl.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ProjectServiceImpl.java index 7450679a72..2279edff6e 100644 --- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ProjectServiceImpl.java +++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ProjectServiceImpl.java @@ -24,6 +24,7 @@ import static org.apache.dolphinscheduler.api.constants.ApiFuncIdentificationCon import org.apache.dolphinscheduler.api.enums.Status; import org.apache.dolphinscheduler.api.exceptions.ServiceException; import org.apache.dolphinscheduler.api.service.ProjectService; +import org.apache.dolphinscheduler.api.service.TaskGroupService; import org.apache.dolphinscheduler.api.utils.PageInfo; import org.apache.dolphinscheduler.api.utils.Result; import org.apache.dolphinscheduler.common.constants.Constants; @@ -60,6 +61,7 @@ import lombok.NonNull; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Lazy; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; @@ -74,6 +76,10 @@ public class ProjectServiceImpl extends BaseServiceImpl implements ProjectServic private static final Logger logger = LoggerFactory.getLogger(ProjectServiceImpl.class); + @Lazy + @Autowired + private TaskGroupService taskGroupService; + @Autowired private ProjectMapper projectMapper; @@ -446,6 +452,9 @@ public class ProjectServiceImpl extends BaseServiceImpl implements ProjectServic putMsg(result, Status.DELETE_PROJECT_ERROR_DEFINES_NOT_NULL); return result; } + // delete the task group + taskGroupService.deleteTaskGroupByProjectCode(project.getCode()); + int delete = projectMapper.deleteById(project.getId()); if (delete > 0) { logger.info("Project is deleted and id is :{}.", project.getId()); diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskGroupQueueServiceImpl.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskGroupQueueServiceImpl.java index d2f990fb3b..89f2c4f133 100644 --- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskGroupQueueServiceImpl.java +++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskGroupQueueServiceImpl.java @@ -28,6 +28,8 @@ import org.apache.dolphinscheduler.dao.entity.User; import org.apache.dolphinscheduler.dao.mapper.ProjectMapper; import org.apache.dolphinscheduler.dao.mapper.TaskGroupQueueMapper; +import org.apache.commons.collections4.CollectionUtils; + import java.util.HashMap; import java.util.List; import java.util.Map; @@ -145,6 +147,19 @@ public class TaskGroupQueueServiceImpl extends BaseServiceImpl implements TaskGr return taskGroupQueueMapper.deleteByTaskId(taskId) == 1; } + @Override + public void deleteByTaskInstanceIds(List taskInstanceIds) { + if (CollectionUtils.isEmpty(taskInstanceIds)) { + return; + } + taskGroupQueueMapper.deleteByTaskInstanceIds(taskInstanceIds); + } + + @Override + public void deleteByWorkflowInstanceId(Integer workflowInstanceId) { + taskGroupQueueMapper.deleteByWorkflowInstanceId(workflowInstanceId); + } + @Override public void forceStartTask(int queueId, int forceStart) { taskGroupQueueMapper.updateForceStart(queueId, forceStart); @@ -154,4 +169,12 @@ public class TaskGroupQueueServiceImpl extends BaseServiceImpl implements TaskGr public void modifyPriority(Integer queueId, Integer priority) { taskGroupQueueMapper.modifyPriority(queueId, priority); } + + @Override + public void deleteByTaskGroupIds(List taskGroupIds) { + if (CollectionUtils.isEmpty(taskGroupIds)) { + return; + } + taskGroupQueueMapper.deleteByTaskGroupIds(taskGroupIds); + } } diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskGroupServiceImpl.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskGroupServiceImpl.java index 1d722e1fd9..232268970a 100644 --- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskGroupServiceImpl.java +++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskGroupServiceImpl.java @@ -29,8 +29,8 @@ import org.apache.dolphinscheduler.common.enums.Flag; import org.apache.dolphinscheduler.dao.entity.TaskGroup; import org.apache.dolphinscheduler.dao.entity.User; import org.apache.dolphinscheduler.dao.mapper.TaskGroupMapper; -import org.apache.dolphinscheduler.service.process.ProcessService; +import org.apache.commons.collections4.CollectionUtils; import org.apache.commons.lang3.StringUtils; import java.util.ArrayList; @@ -40,6 +40,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Set; +import java.util.stream.Collectors; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -63,9 +64,6 @@ public class TaskGroupServiceImpl extends BaseServiceImpl implements TaskGroupSe @Autowired private TaskGroupQueueService taskGroupQueueService; - @Autowired - private ProcessService processService; - @Autowired private ExecutorService executorService; @@ -429,4 +427,17 @@ public class TaskGroupServiceImpl extends BaseServiceImpl implements TaskGroupSe putMsg(result, Status.SUCCESS); return result; } + + @Override + public void deleteTaskGroupByProjectCode(long projectCode) { + List taskGroups = taskGroupMapper.selectByProjectCode(projectCode); + if (CollectionUtils.isEmpty(taskGroups)) { + return; + } + List taskGroupIds = taskGroups.stream() + .map(TaskGroup::getId) + .collect(Collectors.toList()); + taskGroupQueueService.deleteByTaskGroupIds(taskGroupIds); + taskGroupMapper.deleteBatchIds(taskGroupIds); + } } diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskInstanceServiceImpl.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskInstanceServiceImpl.java index d8e8de5b90..0d74489f4b 100644 --- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskInstanceServiceImpl.java +++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskInstanceServiceImpl.java @@ -24,6 +24,7 @@ import static org.apache.dolphinscheduler.api.constants.ApiFuncIdentificationCon import org.apache.dolphinscheduler.api.dto.taskInstance.TaskInstanceRemoveCacheResponse; import org.apache.dolphinscheduler.api.enums.Status; import org.apache.dolphinscheduler.api.service.ProjectService; +import org.apache.dolphinscheduler.api.service.TaskGroupQueueService; import org.apache.dolphinscheduler.api.service.TaskInstanceService; import org.apache.dolphinscheduler.api.service.UsersService; import org.apache.dolphinscheduler.api.utils.PageInfo; @@ -107,6 +108,9 @@ public class TaskInstanceServiceImpl extends BaseServiceImpl implements TaskInst @Autowired private DqExecuteResultDao dqExecuteResultDao; + @Autowired + private TaskGroupQueueService taskGroupQueueService; + /** * query task list by project, process instance, task name, task start time, task end time, task status, keyword paging * @@ -384,7 +388,9 @@ public class TaskInstanceServiceImpl extends BaseServiceImpl implements TaskInst } } } + dqExecuteResultDao.deleteByWorkflowInstanceId(workflowInstanceId); + taskGroupQueueService.deleteByWorkflowInstanceId(workflowInstanceId); taskInstanceDao.deleteByWorkflowInstanceId(workflowInstanceId); } diff --git a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/ProjectServiceTest.java b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/ProjectServiceTest.java index 2eaeebc7a5..6be492e060 100644 --- a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/ProjectServiceTest.java +++ b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/ProjectServiceTest.java @@ -96,6 +96,9 @@ public class ProjectServiceTest { @Mock private ResourcePermissionCheckService resourcePermissionCheckService; + @Mock + private TaskGroupService taskGroupService; + private String projectName = "ProjectServiceTest"; private String userName = "ProjectServiceTest"; diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/TaskGroup.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/TaskGroup.java index 0d42d8480f..3aa49f8e17 100644 --- a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/TaskGroup.java +++ b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/TaskGroup.java @@ -72,7 +72,7 @@ public class TaskGroup implements Serializable { */ private Date updateTime; /** - * project Id + * project code */ private long projectCode; diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/TaskGroupMapper.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/TaskGroupMapper.java index 9893726873..7cb79c9e58 100644 --- a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/TaskGroupMapper.java +++ b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/TaskGroupMapper.java @@ -82,8 +82,11 @@ public interface TaskGroupMapper extends BaseMapper { /** * listAuthorizedResource + * * @param userId * @return */ List listAuthorizedResource(@Param("userId") int userId); + + List selectByProjectCode(@Param("projectCode") long projectCode); } diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/TaskGroupQueueMapper.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/TaskGroupQueueMapper.java index 9826b3e6d5..437a22efb7 100644 --- a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/TaskGroupQueueMapper.java +++ b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/TaskGroupQueueMapper.java @@ -99,4 +99,10 @@ public interface TaskGroupQueueMapper extends BaseMapper { @Param("status") Integer status, @Param("groupId") int groupId, @Param("projects") List projects); + + void deleteByTaskInstanceIds(@Param("taskInstanceIds") List taskInstanceIds); + + void deleteByWorkflowInstanceId(@Param("workflowInstanceId") Integer workflowInstanceId); + + void deleteByTaskGroupIds(@Param("taskGroupIds") List taskGroupIds); } diff --git a/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/TaskGroupMapper.xml b/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/TaskGroupMapper.xml index 969077309e..6e387afb60 100644 --- a/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/TaskGroupMapper.xml +++ b/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/TaskGroupMapper.xml @@ -69,10 +69,17 @@ #{i} - and project_code in ( #{projectCode} , 0) + and project_code in ( #{projectCode} , 0) order by update_time desc + + update t_ds_task_group diff --git a/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/TaskGroupQueueMapper.xml b/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/TaskGroupQueueMapper.xml index 5eacbea9b8..d2c43f18a7 100644 --- a/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/TaskGroupQueueMapper.xml +++ b/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/TaskGroupQueueMapper.xml @@ -157,4 +157,27 @@ order by queue.update_time desc + + delete from t_ds_task_group_queue + where task_id in + + #{i} + + + + + delete + from t_ds_task_group_queue + where process_id = #{workflowInstanceId} + + + + delete + from t_ds_task_group_queue + where group_id in + + #{i} + + + diff --git a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessServiceImpl.java b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessServiceImpl.java index 865692a38e..359a36cde5 100644 --- a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessServiceImpl.java +++ b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessServiceImpl.java @@ -2433,8 +2433,12 @@ public class ProcessServiceImpl implements ProcessService { * @param taskId task id */ @Override - public boolean acquireTaskGroup(int taskId, String taskName, int groupId, int processId, int priority) { - TaskGroup taskGroup = taskGroupMapper.selectById(groupId); + public boolean acquireTaskGroup(int taskInstanceId, + String taskName, + int taskGroupId, + int workflowInstanceId, + int taskGroupPriority) { + TaskGroup taskGroup = taskGroupMapper.selectById(taskGroupId); if (taskGroup == null) { // we don't throw exception here, to avoid the task group has been deleted during workflow running return true; @@ -2444,17 +2448,17 @@ public class ProcessServiceImpl implements ProcessService { return true; } // Create a waiting taskGroupQueue, after acquire resource, we can update the status to ACQUIRE_SUCCESS - TaskGroupQueue taskGroupQueue = this.taskGroupQueueMapper.queryByTaskId(taskId); + TaskGroupQueue taskGroupQueue = this.taskGroupQueueMapper.queryByTaskId(taskInstanceId); if (taskGroupQueue == null) { taskGroupQueue = insertIntoTaskGroupQueue( - taskId, + taskInstanceId, taskName, - groupId, - processId, - priority, + taskGroupId, + workflowInstanceId, + taskGroupPriority, TaskGroupQueueStatus.WAIT_QUEUE); } else { - logger.info("The task queue is already exist, taskId: {}", taskId); + logger.info("The task queue is already exist, taskId: {}", taskInstanceId); if (taskGroupQueue.getStatus() == TaskGroupQueueStatus.ACQUIRE_SUCCESS) { return true; } @@ -2464,19 +2468,19 @@ public class ProcessServiceImpl implements ProcessService { } // check if there already exist higher priority tasks List highPriorityTasks = taskGroupQueueMapper.queryHighPriorityTasks( - groupId, - priority, + taskGroupId, + taskGroupPriority, TaskGroupQueueStatus.WAIT_QUEUE.getCode()); if (CollectionUtils.isNotEmpty(highPriorityTasks)) { return false; } // try to get taskGroup - int count = taskGroupMapper.selectAvailableCountById(groupId); + int count = taskGroupMapper.selectAvailableCountById(taskGroupId); if (count == 1 && robTaskGroupResource(taskGroupQueue)) { - logger.info("Success acquire taskGroup, taskInstanceId: {}, taskGroupId: {}", taskId, groupId); + logger.info("Success acquire taskGroup, taskInstanceId: {}, taskGroupId: {}", taskInstanceId, taskGroupId); return true; } - logger.info("Failed to acquire taskGroup, taskInstanceId: {}, taskGroupId: {}", taskId, groupId); + logger.info("Failed to acquire taskGroup, taskInstanceId: {}, taskGroupId: {}", taskInstanceId, taskGroupId); this.taskGroupQueueMapper.updateInQueue(Flag.NO.getCode(), taskGroupQueue.getId()); return false; } @@ -2608,6 +2612,7 @@ public class ProcessServiceImpl implements ProcessService { .processId(workflowInstanceId) .priority(taskGroupPriority) .status(status) + .inQueue(Flag.NO.getCode()) .createTime(now) .updateTime(now) .build();