From d7eab830a74f5db35a50d6ae1ffd23cb4320fe9f Mon Sep 17 00:00:00 2001 From: JinYong Li <42576980+JinyLeeChina@users.noreply.github.com> Date: Tue, 7 Dec 2021 16:12:59 +0800 Subject: [PATCH] [cherry-pick-7221/7232] pick some api bug (#7234) * fix worker group display (#7208) * [Bug] [API] queryProcessDefinitionByCode bug (#7221) * pick-7232/7221 --- .../api/controller/SchedulerController.java | 6 ++-- .../impl/TaskDefinitionServiceImpl.java | 31 ++++++++++++++++--- .../impl/WorkFlowLineageServiceImpl.java | 3 ++ .../TaskDefinitionServiceImplTest.java | 16 ++++++++-- .../service/process/ProcessService.java | 12 +++++-- 5 files changed, 56 insertions(+), 12 deletions(-) diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/SchedulerController.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/SchedulerController.java index e1edfea077..0e1ca4d3ab 100644 --- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/SchedulerController.java +++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/SchedulerController.java @@ -148,6 +148,7 @@ public class SchedulerController extends BaseController { @ApiImplicitParam(name = "environmentCode", value = "ENVIRONMENT_CODE", dataType = "Long"), }) @PutMapping("/{id}") + @ResponseStatus(HttpStatus.OK) @ApiException(UPDATE_SCHEDULE_ERROR) @AccessLogAnnotation(ignoreRequestArgs = "loginUser") public Result updateSchedule(@ApiIgnore @RequestAttribute(value = SESSION_USER) User loginUser, @@ -330,11 +331,12 @@ public class SchedulerController extends BaseController { @ApiImplicitParam(name = "warningType", value = "WARNING_TYPE", type = "WarningType"), @ApiImplicitParam(name = "warningGroupId", value = "WARNING_GROUP_ID", dataType = "Int", example = "100"), @ApiImplicitParam(name = "failureStrategy", value = "FAILURE_STRATEGY", type = "FailureStrategy"), - @ApiImplicitParam(name = "workerGroupId", value = "WORKER_GROUP_ID", dataType = "Int", example = "100"), + @ApiImplicitParam(name = "workerGroup", value = "WORKER_GROUP", dataType = "String", example = "default"), @ApiImplicitParam(name = "processInstancePriority", value = "PROCESS_INSTANCE_PRIORITY", type = "Priority"), @ApiImplicitParam(name = "environmentCode", value = "ENVIRONMENT_CODE", dataType = "Long"), }) - @PutMapping("/{code}") + @PutMapping("/update/{code}") + @ResponseStatus(HttpStatus.OK) @ApiException(UPDATE_SCHEDULE_ERROR) @AccessLogAnnotation(ignoreRequestArgs = "loginUser") public Result updateScheduleByProcessDefinitionCode(@ApiIgnore @RequestAttribute(value = SESSION_USER) User loginUser, 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 c7c5953bf1..c5f40c5cbe 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 @@ -206,6 +206,19 @@ public class TaskDefinitionServiceImpl extends BaseServiceImpl implements TaskDe } int delete = taskDefinitionMapper.deleteByCode(taskCode); if (delete > 0) { + List taskRelationList = processTaskRelationMapper.queryUpstreamByCode(projectCode, taskCode); + if (!processTaskRelationList.isEmpty()) { + int deleteRelation = 0; + int deleteRelationLog = 0; + for (ProcessTaskRelation processTaskRelation : taskRelationList) { + ProcessTaskRelationLog processTaskRelationLog = new ProcessTaskRelationLog(processTaskRelation); + deleteRelation += processTaskRelationMapper.deleteRelation(processTaskRelationLog); + deleteRelationLog += processTaskRelationLogMapper.deleteRelation(processTaskRelationLog); + } + if ((deleteRelation & deleteRelationLog) == 0) { + throw new ServiceException(Status.DELETE_TASK_PROCESS_RELATION_ERROR); + } + } putMsg(result, Status.SUCCESS); } else { putMsg(result, Status.DELETE_TASK_DEFINE_BY_CODE_ERROR); @@ -492,6 +505,7 @@ public class TaskDefinitionServiceImpl extends BaseServiceImpl implements TaskDe * @param releaseState releaseState * @return update result code */ + @Transactional(rollbackFor = RuntimeException.class) @Override public Map releaseTaskDefinition(User loginUser, long projectCode, long code, ReleaseState releaseState) { Project project = projectMapper.queryByCode(projectCode); @@ -510,11 +524,15 @@ public class TaskDefinitionServiceImpl extends BaseServiceImpl implements TaskDe putMsg(result, Status.TASK_DEFINE_NOT_EXIST, code); return result; } - + TaskDefinitionLog taskDefinitionLog = taskDefinitionLogMapper.queryByDefinitionCodeAndVersion(code, taskDefinition.getVersion()); + if (taskDefinitionLog == null) { + putMsg(result, Status.TASK_DEFINE_NOT_EXIST, code); + return result; + } switch (releaseState) { case OFFLINE: taskDefinition.setFlag(Flag.NO); - taskDefinitionMapper.updateById(taskDefinition); + taskDefinitionLog.setFlag(Flag.NO); break; case ONLINE: String resourceIds = taskDefinition.getResourceIds(); @@ -530,13 +548,18 @@ public class TaskDefinitionServiceImpl extends BaseServiceImpl implements TaskDe } } taskDefinition.setFlag(Flag.YES); - taskDefinitionMapper.updateById(taskDefinition); + taskDefinitionLog.setFlag(Flag.NO); break; default: putMsg(result, Status.REQUEST_PARAMS_NOT_VALID_ERROR, RELEASESTATE); return result; } - + int update = taskDefinitionMapper.updateById(taskDefinition); + int updateLog = taskDefinitionLogMapper.updateById(taskDefinitionLog); + if ((update == 0 && updateLog == 1) || (update == 1 && updateLog == 0)) { + putMsg(result, Status.UPDATE_TASK_DEFINITION_ERROR); + throw new ServiceException(Status.UPDATE_TASK_DEFINITION_ERROR); + } putMsg(result, Status.SUCCESS); return result; } diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/WorkFlowLineageServiceImpl.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/WorkFlowLineageServiceImpl.java index d665a1a184..1367de7d96 100644 --- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/WorkFlowLineageServiceImpl.java +++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/WorkFlowLineageServiceImpl.java @@ -177,6 +177,9 @@ public class WorkFlowLineageServiceImpl extends BaseServiceImpl implements WorkF private Set querySourceWorkFlowCodes(long projectCode, long workFlowCode, List taskDefinitionList) { Set sourceWorkFlowCodes = new HashSet<>(); + if (taskDefinitionList == null || taskDefinitionList.isEmpty()) { + return sourceWorkFlowCodes; + } List taskDefinitionLogs = taskDefinitionLogMapper.queryByTaskDefinitions(taskDefinitionList); for (TaskDefinitionLog taskDefinitionLog : taskDefinitionLogs) { if (taskDefinitionLog.getProjectCode() == projectCode) { 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 94f2ad5889..d8852b4ce0 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 @@ -207,9 +207,10 @@ public class TaskDefinitionServiceImplTest { Mockito.when(taskDefinitionLogMapper.queryByDefinitionCodeAndVersion(taskCode, version)) .thenReturn(new TaskDefinitionLog()); - + TaskDefinition taskDefinition = new TaskDefinition(); + taskDefinition.setProjectCode(projectCode); Mockito.when(taskDefinitionMapper.queryByCode(taskCode)) - .thenReturn(new TaskDefinition()); + .thenReturn(taskDefinition); Mockito.when(taskDefinitionMapper.updateById(new TaskDefinitionLog())).thenReturn(1); Map relation = taskDefinitionService .switchVersion(loginUser, projectCode, taskCode, version); @@ -306,7 +307,16 @@ public class TaskDefinitionServiceImplTest { // process definition offline putMsg(result, Status.SUCCESS); - Mockito.when(taskDefinitionMapper.queryByCode(taskCode)).thenReturn(new TaskDefinition()); + TaskDefinition taskDefinition = new TaskDefinition(); + taskDefinition.setProjectCode(projectCode); + taskDefinition.setVersion(1); + taskDefinition.setCode(taskCode); + String params = "{\"resourceList\":[],\"localParams\":[],\"rawScript\":\"echo 1\",\"conditionResult\":{\"successNode\":[\"\"],\"failedNode\":[\"\"]},\"dependence\":{}}"; + taskDefinition.setTaskParams(params); + taskDefinition.setTaskType(TaskType.SHELL.getDesc()); + Mockito.when(taskDefinitionMapper.queryByCode(taskCode)).thenReturn(taskDefinition); + TaskDefinitionLog taskDefinitionLog = new TaskDefinitionLog(taskDefinition); + Mockito.when(taskDefinitionLogMapper.queryByDefinitionCodeAndVersion(taskCode, taskDefinition.getVersion())).thenReturn(taskDefinitionLog); Map offlineTaskResult = taskDefinitionService.releaseTaskDefinition(loginUser, projectCode, taskCode, ReleaseState.OFFLINE); Assert.assertEquals(Status.SUCCESS, offlineTaskResult.get(Constants.STATUS)); diff --git a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessService.java b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessService.java index 2dcfd2d687..df7549cd76 100644 --- a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessService.java +++ b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessService.java @@ -126,9 +126,9 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; import org.springframework.transaction.annotation.Transactional; -import com.facebook.presto.jdbc.internal.guava.collect.Lists; import com.fasterxml.jackson.core.type.TypeReference; import com.fasterxml.jackson.databind.node.ObjectNode; +import com.google.common.collect.Lists; /** * process relative dao that some mappers in this. @@ -330,7 +330,7 @@ public class ProcessService { ProcessDefinition processDefinition = processDefineMapper.queryByCode(defineCode); if (processDefinition == null) { logger.error("process define not exists"); - return new ArrayList<>(); + return Lists.newArrayList(); } List processTaskRelations = processTaskRelationLogMapper.queryByProcessCodeAndVersion(processDefinition.getCode(), processDefinition.getVersion()); Set taskDefinitionSet = new HashSet<>(); @@ -339,8 +339,11 @@ public class ProcessService { taskDefinitionSet.add(new TaskDefinition(processTaskRelation.getPostTaskCode(), processTaskRelation.getPostTaskVersion())); } } + if (taskDefinitionSet.isEmpty()) { + return Lists.newArrayList(); + } List taskDefinitionLogs = taskDefinitionLogMapper.queryByTaskDefinitions(taskDefinitionSet); - return new ArrayList<>(taskDefinitionLogs); + return Lists.newArrayList(taskDefinitionLogs); } /** @@ -2367,6 +2370,9 @@ public class ProcessService { taskDefinitionSet.add(new TaskDefinition(processTaskRelation.getPostTaskCode(), processTaskRelation.getPostTaskVersion())); } } + if (taskDefinitionSet.isEmpty()) { + return Lists.newArrayList(); + } return taskDefinitionLogMapper.queryByTaskDefinitions(taskDefinitionSet); }