From 493e4ac912cb7f94fa57ed01f0589ab8b5fe5eda Mon Sep 17 00:00:00 2001 From: qiaozhanwei Date: Wed, 15 May 2019 16:05:10 +0800 Subject: [PATCH] Delete the process instance and delete the corresponding queue information. --- .../controller/ProcessInstanceController.java | 6 ++- .../api/service/ProcessInstanceService.java | 39 ++++++++++++++----- 2 files changed, 35 insertions(+), 10 deletions(-) diff --git a/escheduler-api/src/main/java/cn/escheduler/api/controller/ProcessInstanceController.java b/escheduler-api/src/main/java/cn/escheduler/api/controller/ProcessInstanceController.java index b05aa656f1..cea47be2c6 100644 --- a/escheduler-api/src/main/java/cn/escheduler/api/controller/ProcessInstanceController.java +++ b/escheduler-api/src/main/java/cn/escheduler/api/controller/ProcessInstanceController.java @@ -22,6 +22,8 @@ import cn.escheduler.api.utils.Constants; import cn.escheduler.api.utils.Result; import cn.escheduler.common.enums.ExecutionStatus; import cn.escheduler.common.enums.Flag; +import cn.escheduler.common.queue.ITaskQueue; +import cn.escheduler.common.queue.TaskQueueFactory; import cn.escheduler.common.utils.ParameterUtils; import cn.escheduler.dao.model.User; import org.slf4j.Logger; @@ -189,7 +191,9 @@ public class ProcessInstanceController extends BaseController{ try{ logger.info("delete process instance by id, login user:{}, project name:{}, process instance id:{}", loginUser.getUserName(), projectName, processInstanceId); - Map result = processInstanceService.deleteProcessInstanceById(loginUser, projectName, processInstanceId); + // task queue + ITaskQueue tasksQueue = TaskQueueFactory.getTaskQueueInstance(); + Map result = processInstanceService.deleteProcessInstanceById(loginUser, projectName, processInstanceId,tasksQueue); return returnDataList(result); }catch (Exception e){ logger.error(DELETE_PROCESS_INSTANCE_BY_ID_ERROR.getMsg(),e); diff --git a/escheduler-api/src/main/java/cn/escheduler/api/service/ProcessInstanceService.java b/escheduler-api/src/main/java/cn/escheduler/api/service/ProcessInstanceService.java index 70abaf07eb..53d83536a0 100644 --- a/escheduler-api/src/main/java/cn/escheduler/api/service/ProcessInstanceService.java +++ b/escheduler-api/src/main/java/cn/escheduler/api/service/ProcessInstanceService.java @@ -30,6 +30,8 @@ import cn.escheduler.common.graph.DAG; import cn.escheduler.common.model.TaskNode; import cn.escheduler.common.model.TaskNodeRelation; import cn.escheduler.common.process.Property; +import cn.escheduler.common.queue.ITaskQueue; +import cn.escheduler.common.queue.TaskQueueFactory; import cn.escheduler.common.utils.CollectionUtils; import cn.escheduler.common.utils.DateUtils; import cn.escheduler.common.utils.JSONUtils; @@ -446,13 +448,13 @@ public class ProcessInstanceService extends BaseDAGService { /** * delete process instance by id, at the same timeļ¼Œdelete task instance and their mapping relation data - * * @param loginUser * @param projectName - * @param workflowId + * @param processInstanceId + * @param tasksQueue * @return */ - public Map deleteProcessInstanceById(User loginUser, String projectName, Integer workflowId) { + public Map deleteProcessInstanceById(User loginUser, String projectName, Integer processInstanceId,ITaskQueue tasksQueue) { Map result = new HashMap<>(5); Project project = projectMapper.queryByName(projectName); @@ -462,17 +464,34 @@ public class ProcessInstanceService extends BaseDAGService { if (resultEnum != Status.SUCCESS) { return checkResult; } - ProcessInstance processInstance = processDao.findProcessInstanceDetailById(workflowId); + ProcessInstance processInstance = processDao.findProcessInstanceDetailById(processInstanceId); + //process instance priority + int processInstancePriority = processInstance.getProcessInstancePriority().ordinal(); if (processInstance == null) { - putMsg(result, Status.PROCESS_INSTANCE_NOT_EXIST, workflowId); + putMsg(result, Status.PROCESS_INSTANCE_NOT_EXIST, processInstanceId); return result; } - int delete = processDao.deleteWorkProcessInstanceById(workflowId); - processDao.deleteAllSubWorkProcessByParentId(workflowId); - processDao.deleteWorkProcessMapByParentId(workflowId); + int delete = processDao.deleteWorkProcessInstanceById(processInstanceId); + processDao.deleteAllSubWorkProcessByParentId(processInstanceId); + processDao.deleteWorkProcessMapByParentId(processInstanceId); if (delete > 0) { + List taskInstanceList = processDao.findValidTaskListByProcessId(processInstanceId); + if (CollectionUtils.isNotEmpty(taskInstanceList)){ + for (TaskInstance taskInstance : taskInstanceList){ + // task instance priority + int taskInstancePriority = taskInstance.getTaskInstancePriority().ordinal(); + String nodeValue=processInstancePriority + "_" + processInstanceId + "_" +taskInstancePriority + "_" + taskInstance.getId(); + try { + logger.info("delete task queue node : {}",nodeValue); + tasksQueue.removeNode(cn.escheduler.common.Constants.SCHEDULER_TASKS_QUEUE, nodeValue); + }catch (Exception e){ + logger.error("delete task queue node : {}",nodeValue); + } + } + } + putMsg(result, Status.SUCCESS); } else { putMsg(result, Status.DELETE_PROCESS_INSTANCE_BY_ID_ERROR); @@ -489,6 +508,8 @@ public class ProcessInstanceService extends BaseDAGService { * @return */ public Map batchDeleteProcessInstanceByIds(User loginUser, String projectName, String processInstanceIds) { + // task queue + ITaskQueue tasksQueue = TaskQueueFactory.getTaskQueueInstance(); Map result = new HashMap<>(5); List deleteFailedIdList = new ArrayList(); @@ -507,7 +528,7 @@ public class ProcessInstanceService extends BaseDAGService { for (String strProcessInstanceId:processInstanceIdArray) { int processInstanceId = Integer.parseInt(strProcessInstanceId); try { - deleteProcessInstanceById(loginUser, projectName, processInstanceId); + deleteProcessInstanceById(loginUser, projectName, processInstanceId,tasksQueue); } catch (Exception e) { deleteFailedIdList.add(processInstanceId); }