From c7f990791e558ba8f38a4ff27f840fff53d17a0a Mon Sep 17 00:00:00 2001 From: jackfanwan <61672564+jackfanwan@users.noreply.github.com> Date: Sun, 6 Nov 2022 20:49:07 +0800 Subject: [PATCH] cherry-pick [fix-12675]edit workflow related task, workflow's task version change --- .../dolphinscheduler/api/enums/Status.java | 16 ++++++++ .../impl/TaskDefinitionServiceImpl.java | 19 +++++++++ .../TaskDefinitionServiceImplTest.java | 41 ++++++++++++++++--- .../dao/mapper/ProcessTaskRelationMapper.java | 17 ++++++++ .../dao/mapper/ProcessTaskRelationMapper.xml | 31 ++++++++++++++ 5 files changed, 118 insertions(+), 6 deletions(-) diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/enums/Status.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/enums/Status.java index 47bb5dbbbf..cb9a8d60a7 100644 --- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/enums/Status.java +++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/enums/Status.java @@ -311,6 +311,22 @@ public enum Status { NOT_SUPPORT_UPDATE_TASK_DEFINITION(50056, "task state does not support modification", "当前任务不支持修改"), BATCH_EXECUTE_PROCESS_INSTANCE_ERROR(50058, "change process instance status error: {0}", "修改工作实例状态错误: {0}"), START_TASK_INSTANCE_ERROR(50059, "start task instance error", "运行任务流实例错误"), + DELETE_PROCESS_DEFINE_ERROR(50060, "delete process definition [{0}] error: {1}", "删除工作流定义[{0}]错误: {1}"), + CREATE_TASK_DEFINITION_LOG_ERROR(50061, "create task definition log {0} error", "创建任务操作记录 {0} 错误"), + DELETE_TASK_DEFINE_BY_CODE_MSG_ERROR(50062, "delete task definition {0} error", "删除任务定义 {0} 错误"), + TASK_DEFINITION_NOT_CHANGE(50063, "task definition {0} do not change", "任务定义 {0} 没有变化"), + TASK_DEFINITION_NOT_EXISTS(50064, "task definition {0} do not exists", "任务定义 {0} 不存在"), + UPDATE_UPSTREAM_TASK_PROCESS_RELATION_ERROR(50065, "update task upstream relation error", "更新任务上游关系错误"), + CREATE_PROCESS_TASK_RELATION_LOG_ERROR(50066, "create process task relation log {0}-{1} error", + "创建任务关系日志 {0}-{1} 错误"), + PROCESS_TASK_RELATION_NOT_EXPECT(50067, "process task relation number not expect, expect {0} but get {1}", + "工作流任务关系数量不符合预期，预期 {0} 但是实际 {1}"), + PROCESS_TASK_RELATION_BATCH_DELETE_ERROR(50068, "batch delete process task relation {0} error", + "批量删除工作流任务关系 {0} 错误"), + PROCESS_TASK_RELATION_BATCH_CREATE_ERROR(50069, "batch create process task relation {0} error", + "批量创建工作流任务关系 {0} 错误"), + PROCESS_TASK_RELATION_BATCH_UPDATE_ERROR(50070, "batch update process task relation error", + "批量修改工作流任务关系错误"), HDFS_NOT_STARTUP(60001, "hdfs not startup", "hdfs未启用"), STORAGE_NOT_STARTUP(60002, "storage not startup", "存储未启用"), 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 1d5c7f7f01..e6e57151dd 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 @@ -534,6 +534,25 @@ public class TaskDefinitionServiceImpl extends BaseServiceImpl implements TaskDe logger.info( "Update task definition and definitionLog complete, projectCode:{}, taskDefinitionCode:{}, newTaskVersion:{}.", projectCode, taskCode, taskDefinitionToUpdate.getVersion()); + // update process task relation + List processTaskRelations = processTaskRelationMapper + .queryByTaskCode(taskDefinitionToUpdate.getCode()); + if (CollectionUtils.isNotEmpty(processTaskRelations)) { + for (ProcessTaskRelation processTaskRelation : processTaskRelations) { + if (taskCode == processTaskRelation.getPreTaskCode()) { + processTaskRelation.setPreTaskVersion(version); + } else if (taskCode == processTaskRelation.getPostTaskCode()) { + processTaskRelation.setPostTaskVersion(version); + } + int count = processTaskRelationMapper.updateProcessTaskRelationTaskVersion(processTaskRelation); + if (count != 1) { + logger.error("batch update process task relation error, projectCode:{}, taskDefinitionCode:{}.", + projectCode, taskCode); + putMsg(result, Status.PROCESS_TASK_RELATION_BATCH_UPDATE_ERROR); + throw new ServiceException(Status.PROCESS_TASK_RELATION_BATCH_UPDATE_ERROR); + } + } + } return taskDefinitionToUpdate; } 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 61ab0f6f7c..288223723e 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 @@ -26,10 +26,7 @@ import org.apache.dolphinscheduler.common.enums.Priority; import org.apache.dolphinscheduler.common.enums.ReleaseState; import org.apache.dolphinscheduler.common.enums.UserType; import org.apache.dolphinscheduler.common.utils.JSONUtils; -import org.apache.dolphinscheduler.dao.entity.Project; -import org.apache.dolphinscheduler.dao.entity.TaskDefinition; -import org.apache.dolphinscheduler.dao.entity.TaskDefinitionLog; -import org.apache.dolphinscheduler.dao.entity.User; +import org.apache.dolphinscheduler.dao.entity.*; import org.apache.dolphinscheduler.dao.mapper.ProcessTaskRelationMapper; import org.apache.dolphinscheduler.dao.mapper.ProjectMapper; import org.apache.dolphinscheduler.dao.mapper.TaskDefinitionLogMapper; @@ -140,8 +137,19 @@ public class TaskDefinitionServiceImplTest { Mockito.when(taskDefinitionLogMapper.insert(Mockito.any(TaskDefinitionLog.class))).thenReturn(1); Mockito.when(taskDefinitionLogMapper.queryMaxVersionForDefinition(taskCode)).thenReturn(1); Mockito.when(taskPluginManager.checkTaskParameters(Mockito.any())).thenReturn(true); - result = taskDefinitionService.updateTaskDefinition(loginUser, projectCode, taskCode, taskDefinitionJson); - Assert.assertEquals(Status.SUCCESS, result.get(Constants.STATUS)); + Mockito.when(processTaskRelationMapper.queryByTaskCode(3)).thenReturn(getProcessTaskRelationList2()); + Mockito.when(processTaskRelationMapper + .updateProcessTaskRelationTaskVersion(Mockito.any(ProcessTaskRelation.class))).thenReturn(1); + result = taskDefinitionService.updateTaskDefinition(user, PROJECT_CODE, TASK_CODE, taskDefinitionJson); + Assertions.assertEquals(Status.SUCCESS, result.get(Constants.STATUS)); + // failure + Mockito.when(processTaskRelationMapper + .updateProcessTaskRelationTaskVersion(Mockito.any(ProcessTaskRelation.class))).thenReturn(2); + exception = Assertions.assertThrows(ServiceException.class, + () -> taskDefinitionService.updateTaskDefinition(user, PROJECT_CODE, TASK_CODE, taskDefinitionJson)); + Assertions.assertEquals(Status.PROCESS_TASK_RELATION_BATCH_UPDATE_ERROR.getCode(), + ((ServiceException) exception).getCode()); + } @Test @@ -336,4 +344,25 @@ public class TaskDefinitionServiceImplTest { Map failResult = taskDefinitionService.releaseTaskDefinition(loginUser, projectCode, taskCode, ReleaseState.getEnum(2)); Assert.assertEquals(Status.REQUEST_PARAMS_NOT_VALID_ERROR, failResult.get(Constants.STATUS)); } + + private List getProcessTaskRelationList2() { + List processTaskRelationList = new ArrayList<>(); + + ProcessTaskRelation processTaskRelation = new ProcessTaskRelation(); + processTaskRelation.setProjectCode(PROJECT_CODE); + processTaskRelation.setProcessDefinitionCode(PROCESS_DEFINITION_CODE); + processTaskRelation.setPreTaskCode(TASK_CODE); + processTaskRelation.setPostTaskCode(TASK_CODE + 1L); + + processTaskRelationList.add(processTaskRelation); + + ProcessTaskRelation processTaskRelation2 = new ProcessTaskRelation(); + processTaskRelation2.setProjectCode(PROJECT_CODE); + processTaskRelation2.setProcessDefinitionCode(PROCESS_DEFINITION_CODE); + processTaskRelation2.setPreTaskCode(TASK_CODE - 1); + processTaskRelation2.setPostTaskCode(TASK_CODE); + processTaskRelationList.add(processTaskRelation2); + + return processTaskRelationList; + } } diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/ProcessTaskRelationMapper.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/ProcessTaskRelationMapper.java index 4332b54960..7d1d18b169 100644 --- a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/ProcessTaskRelationMapper.java +++ b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/ProcessTaskRelationMapper.java @@ -17,6 +17,7 @@ package org.apache.dolphinscheduler.dao.mapper; +import com.baomidou.mybatisplus.core.metadata.IPage; import org.apache.dolphinscheduler.dao.entity.ProcessTaskRelation; import org.apache.dolphinscheduler.dao.entity.ProcessTaskRelationLog; @@ -197,5 +198,21 @@ public interface ProcessTaskRelationMapper extends BaseMapper queryDownstreamByProcessDefinitionCode(@Param("processDefinitionCode") long processDefinitionCode); + /** + * Filter process task relation + * + * @param page page + * @param processTaskRelation process definition object + * @return process task relation IPage + */ + IPage filterProcessTaskRelation(IPage page, + @Param("relation") ProcessTaskRelation processTaskRelation); + /** + * batch update process task relation version + * + * @param processTaskRelationList process task relation list + * @return update num + */ + int updateProcessTaskRelationTaskVersion(@Param("processTaskRelation") ProcessTaskRelation processTaskRelationList); } diff --git a/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/ProcessTaskRelationMapper.xml b/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/ProcessTaskRelationMapper.xml index 2a1892db2b..8afa0433c2 100644 --- a/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/ProcessTaskRelationMapper.xml +++ b/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/ProcessTaskRelationMapper.xml @@ -200,4 +200,35 @@ and post_task_code = #{postTaskCode} + + + + update t_ds_process_task_relation + set + pre_task_version=#{processTaskRelation.preTaskVersion}, + post_task_version=#{processTaskRelation.postTaskVersion} + where + id = #{processTaskRelation.id} +