You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@dolphinscheduler.apache.org by zi...@apache.org on 2026/08/24 13:04:19 UTC
(dolphinscheduler) branch dev updated: [Improvement-18568][API] Remove obsolete task update-with-upstream API (#18569)
This is an automated email from the ASF dual-hosted git repository.
SbloodyS pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/dolphinscheduler.git
The following commit(s) were added to refs/heads/dev by this push:
new ca9d7007f7 [Improvement-18568][API] Remove obsolete task update-with-upstream API (#18569)
ca9d7007f7 is described below
commit ca9d7007f71c2fad17c37e723fd60fa6d565ba51
Author: Wenjun Ruan <we...@apache.org>
AuthorDate: Mon Aug 24 21:04:11 2026 +0800
[Improvement-18568][API] Remove obsolete task update-with-upstream API (#18569)
---
docs/docs/en/guide/upgrade/incompatible.md | 1 +
docs/docs/zh/guide/upgrade/incompatible.md | 1 +
.../api/controller/TaskDefinitionController.java | 33 ---
.../apache/dolphinscheduler/api/enums/Status.java | 5 -
.../api/service/TaskDefinitionService.java | 16 --
.../service/impl/TaskDefinitionServiceImpl.java | 316 ---------------------
.../api/service/TaskDefinitionServiceImplTest.java | 83 ------
.../dao/mapper/WorkflowTaskRelationMapper.java | 17 --
.../dao/repository/WorkflowTaskRelationDao.java | 4 -
.../impl/WorkflowTaskRelationDaoImpl.java | 11 -
.../dao/mapper/WorkflowTaskRelationMapper.xml | 22 --
11 files changed, 2 insertions(+), 507 deletions(-)
diff --git a/docs/docs/en/guide/upgrade/incompatible.md b/docs/docs/en/guide/upgrade/incompatible.md
index dd8e482802..55bc8a6041 100644
--- a/docs/docs/en/guide/upgrade/incompatible.md
+++ b/docs/docs/en/guide/upgrade/incompatible.md
@@ -48,4 +48,5 @@ This document records the incompatible updates between each version. You need to
* Add the `missed_fire_policy` column to `t_ds_schedules`. Existing schedules default to `FIRE_ALL_MISSED` to preserve the previous Quartz `IgnoreMisfires` behavior. ([#18464](https://github.com/apache/dolphinscheduler/pull/18464))
* Remove the obsolete Dynamic Task query API. ([#18556](https://github.com/apache/dolphinscheduler/issues/18556))
+* Remove the obsolete task update-with-upstream API `PUT /projects/{projectCode}/task-definition/{code}/with-upstream`. ([#18568](https://github.com/apache/dolphinscheduler/issues/18568))
diff --git a/docs/docs/zh/guide/upgrade/incompatible.md b/docs/docs/zh/guide/upgrade/incompatible.md
index ee324952dd..ba52dd5fe8 100644
--- a/docs/docs/zh/guide/upgrade/incompatible.md
+++ b/docs/docs/zh/guide/upgrade/incompatible.md
@@ -48,4 +48,5 @@
* 为 `t_ds_schedules` 表新增 `missed_fire_policy` 字段。现有定时默认使用 `FIRE_ALL_MISSED`，以保持原有 Quartz `IgnoreMisfires` 行为。([#18464](https://github.com/apache/dolphinscheduler/pull/18464))
* 移除已废弃的 Dynamic Task 查询接口。([#18556](https://github.com/apache/dolphinscheduler/issues/18556))
+* 移除已废弃的任务及其上游关系更新接口 `PUT /projects/{projectCode}/task-definition/{code}/with-upstream`。([#18568](https://github.com/apache/dolphinscheduler/issues/18568))
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/TaskDefinitionController.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/TaskDefinitionController.java
index 5610412a6d..8c96bff9c8 100644
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/TaskDefinitionController.java
+++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/TaskDefinitionController.java
@@ -23,7 +23,6 @@ import static org.apache.dolphinscheduler.api.enums.Status.QUERY_DETAIL_OF_TASK_
import static org.apache.dolphinscheduler.api.enums.Status.QUERY_TASK_DEFINITION_VERSIONS_ERROR;
import static org.apache.dolphinscheduler.api.enums.Status.RELEASE_TASK_DEFINITION_ERROR;
import static org.apache.dolphinscheduler.api.enums.Status.SWITCH_TASK_DEFINITION_VERSION_ERROR;
-import static org.apache.dolphinscheduler.api.enums.Status.UPDATE_TASK_DEFINITION_ERROR;
import org.apache.dolphinscheduler.api.audit.OperatorLog;
import org.apache.dolphinscheduler.api.audit.enums.AuditType;
@@ -43,7 +42,6 @@ import org.springframework.web.bind.annotation.DeleteMapping;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PostMapping;
-import org.springframework.web.bind.annotation.PutMapping;
import org.springframework.web.bind.annotation.RequestAttribute;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
@@ -64,37 +62,6 @@ public class TaskDefinitionController extends BaseController {
@Autowired
private TaskDefinitionService taskDefinitionService;
- /**
- * update task definition
- *
- * @param loginUser login user
- * @param projectCode project code
- * @param code task definition code
- * @param taskDefinitionJsonObj task definition json object
- * @param upstreamCodes upstream task codes, sep comma
- * @return update result code
- */
- @Operation(summary = "updateWithUpstream", description = "UPDATE_TASK_DEFINITION_NOTES")
- @Parameters({
- @Parameter(name = "projectCode", description = "PROJECT_CODE", required = true, schema = @Schema(implementation = long.class)),
- @Parameter(name = "code", description = "TASK_DEFINITION_CODE", required = true, schema = @Schema(implementation = long.class, example = "1")),
- @Parameter(name = "taskDefinitionJsonObj", description = "TASK_DEFINITION_JSON", required = true, schema = @Schema(implementation = String.class)),
- @Parameter(name = "upstreamCodes", description = "UPSTREAM_CODES", required = false, schema = @Schema(implementation = String.class))
- })
- @PutMapping(value = "/{code}/with-upstream")
- @ResponseStatus(HttpStatus.OK)
- @ApiException(UPDATE_TASK_DEFINITION_ERROR)
- @OperatorLog(auditType = AuditType.TASK_UPDATE)
- public Result<Long> updateTaskWithUpstream(@Parameter(hidden = true) @RequestAttribute(value = Constants.SESSION_USER) User loginUser,
- @Parameter(name = "projectCode", description = "PROJECT_CODE", required = true) @PathVariable long projectCode,
- @PathVariable(value = "code") long code,
- @RequestParam(value = "taskDefinitionJsonObj", required = true) String taskDefinitionJsonObj,
- @RequestParam(value = "upstreamCodes", required = false) String upstreamCodes) {
- Long updatedTaskCode = taskDefinitionService.updateTaskWithUpstream(loginUser, projectCode, code,
- taskDefinitionJsonObj, upstreamCodes);
- return Result.success(updatedTaskCode);
- }
-
/**
* query task definition version paging list info
*
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 b47a9b1089..05f951a16e 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
@@ -396,7 +396,6 @@ public enum Status {
MAIN_TABLE_USING_VERSION(50053, "the version that the master table is using", "主表正在使用该版本"),
PROJECT_WORKFLOW_NOT_MATCH(50054, "the project and the workflow is not match", "项目和工作流不匹配"),
DELETE_EDGE_ERROR(50055, "delete edge error", "删除工作流任务连接线错误"),
- NOT_SUPPORT_UPDATE_TASK_DEFINITION(50056, "task state does not support modification", "当前任务不支持修改"),
TASK_DEFINITION_NOT_MODIFY_ERROR(50057, "task [{0}] definition not modify error", "任务[{0}]定义未修改错误"),
BATCH_EXECUTE_WORKFLOW_INSTANCE_ERROR(50058, "change workflow instance status error: {0}", "修改工作实例状态错误: {0}"),
START_TASK_INSTANCE_ERROR(50059, "start task instance error", "运行任务流实例错误"),
@@ -406,16 +405,12 @@ public enum Status {
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_WORKFLOW_RELATION_ERROR(50065, "update task upstream relation error", "更新任务上游关系错误"),
- CREATE_WORKFLOW_TASK_RELATION_LOG_ERROR(50066, "create workflow task relation log {0}-{1} error",
- "创建任务关系日志 {0}-{1} 错误"),
WORKFLOW_TASK_RELATION_NOT_EXPECT(50067, "workflow task relation number not expect, expect {0} but get {1}",
"工作流任务关系数量不符合预期，预期 {0} 但是实际 {1}"),
WORKFLOW_TASK_RELATION_BATCH_DELETE_ERROR(50068, "batch delete workflow task relation {0} error",
"批量删除工作流任务关系 {0} 错误"),
WORKFLOW_TASK_RELATION_BATCH_CREATE_ERROR(50069, "batch create workflow task relation {0} error",
"批量创建工作流任务关系 {0} 错误"),
- WORKFLOW_TASK_RELATION_BATCH_UPDATE_ERROR(50070, "batch update workflow task relation error",
- "批量修改工作流任务关系错误"),
UPSTREAM_TASK_NOT_EXISTS(50071, "upstream task want to set dependence do not exists {0}", "指定的上游任务 {0} 不存在"),
WORKFLOW_INSTANCE_IS_NOT_FINISHED(50071, "the workflow instance is not finished, can not do this operation",
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 f2750efdca..b4e150be0e 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
@@ -51,22 +51,6 @@ public interface TaskDefinitionService {
TaskDefinition getTaskDefinition(User loginUser,
long taskCode);
- /**
- * update task definition and upstream
- *
- * @param loginUser login user
- * @param projectCode project code
- * @param taskCode task definition code
- * @param taskDefinitionJsonObj task definition json object
- * @param upstreamCodes upstream task codes, sep comma
- * @return updated task code
- */
- Long updateTaskWithUpstream(User loginUser,
- long projectCode,
- long taskCode,
- String taskDefinitionJsonObj,
- String upstreamCodes);
-
/**
* update task definition
*
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 b39b0de17c..cb1a8afc51 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
@@ -19,7 +19,6 @@ package org.apache.dolphinscheduler.api.service.impl;
import static org.apache.dolphinscheduler.api.constants.ApiFuncIdentificationConstant.TASK_DEFINITION;
import static org.apache.dolphinscheduler.api.constants.ApiFuncIdentificationConstant.TASK_VERSION_VIEW;
-import static org.apache.dolphinscheduler.plugin.task.api.TaskPluginManager.checkTaskParameters;
import org.apache.dolphinscheduler.api.enums.Status;
import org.apache.dolphinscheduler.api.exceptions.ServiceException;
@@ -33,44 +32,33 @@ import org.apache.dolphinscheduler.api.utils.Result;
import org.apache.dolphinscheduler.api.vo.TaskDefinitionVO;
import org.apache.dolphinscheduler.common.constants.Constants;
import org.apache.dolphinscheduler.common.enums.AuthorizationType;
-import org.apache.dolphinscheduler.common.enums.ConditionType;
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.common.enums.TimeoutFlag;
import org.apache.dolphinscheduler.common.utils.CodeGenerateUtils;
-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.WorkflowDefinition;
-import org.apache.dolphinscheduler.dao.entity.WorkflowDefinitionLog;
import org.apache.dolphinscheduler.dao.entity.WorkflowTaskRelation;
import org.apache.dolphinscheduler.dao.entity.WorkflowTaskRelationLog;
import org.apache.dolphinscheduler.dao.mapper.TaskDefinitionLogMapper;
-import org.apache.dolphinscheduler.dao.mapper.WorkflowDefinitionLogMapper;
import org.apache.dolphinscheduler.dao.repository.ProjectDao;
import org.apache.dolphinscheduler.dao.repository.TaskDefinitionDao;
import org.apache.dolphinscheduler.dao.repository.WorkflowDefinitionDao;
import org.apache.dolphinscheduler.dao.repository.WorkflowTaskRelationDao;
-import org.apache.dolphinscheduler.dao.repository.WorkflowTaskRelationLogDao;
import org.apache.dolphinscheduler.service.process.ProcessService;
import org.apache.commons.collections4.CollectionUtils;
-import org.apache.commons.collections4.MapUtils;
import org.apache.commons.lang3.StringUtils;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Date;
-import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
-import java.util.Map;
import java.util.Set;
-import java.util.function.Function;
import java.util.stream.Collectors;
import lombok.extern.slf4j.Slf4j;
@@ -102,9 +90,6 @@ public class TaskDefinitionServiceImpl extends BaseServiceImpl implements TaskDe
@Autowired
private WorkflowTaskRelationDao workflowTaskRelationDao;
- @Autowired
- private WorkflowTaskRelationLogDao workflowTaskRelationLogDao;
-
@Autowired
private WorkflowTaskRelationService workflowTaskRelationService;
@@ -114,12 +99,8 @@ public class TaskDefinitionServiceImpl extends BaseServiceImpl implements TaskDe
@Autowired
private ProcessService processService;
- @Autowired
- private WorkflowDefinitionLogMapper workflowDefinitionLogMapper;
-
@Autowired
private TaskDatasourcePermissionChecker taskDatasourcePermissionChecker;
-
/**
* query task definition
*
@@ -197,303 +178,6 @@ public class TaskDefinitionServiceImpl extends BaseServiceImpl implements TaskDe
return taskDefinition;
}
- /**
- * Update the task body and cascade through workflow task relations.
- *
- * @return the persisted {@link TaskDefinitionLog}, or {@code null} when the
- * task body is unchanged (callers may still need to apply upstream
- * changes). All other failure modes throw {@link ServiceException}.
- */
- private TaskDefinitionLog updateTask(User loginUser, long projectCode, long taskCode,
- String taskDefinitionJsonObj) {
- Project project = projectDao.queryByCode(projectCode);
-
- // check if user have write perm for project
- projectService.checkHasProjectWritePermissionThrowException(loginUser, project);
-
- TaskDefinition taskDefinition = taskDefinitionDao.queryByCode(taskCode);
- if (taskDefinition == null) {
- log.error("Task definition does not exist, taskDefinitionCode:{}.", taskCode);
- throw new ServiceException(Status.TASK_DEFINE_NOT_EXIST, String.valueOf(taskCode));
- }
- if (processService.isTaskOnline(taskCode) && taskDefinition.getFlag() == Flag.YES) {
- // if stream, can update task definition without online check
- if (taskDefinition.getTaskExecuteType() != TaskExecuteType.STREAM) {
- log.warn("Only {} type task can be updated without online check, taskDefinitionCode:{}.",
- TaskExecuteType.STREAM, taskCode);
- throw new ServiceException(Status.NOT_SUPPORT_UPDATE_TASK_DEFINITION);
- }
- }
- TaskDefinitionLog taskDefinitionToUpdate =
- JSONUtils.parseObject(taskDefinitionJsonObj, TaskDefinitionLog.class);
- if (taskDefinitionToUpdate == null) {
- log.warn("Parameter taskDefinitionJson is invalid.");
- throw new ServiceException(Status.DATA_IS_NOT_VALID, taskDefinitionJsonObj);
- }
- if (!checkTaskParameters(taskDefinitionToUpdate.getTaskType(), taskDefinitionToUpdate.getTaskParams())) {
- throw new ServiceException(Status.WORKFLOW_NODE_S_PARAMETER_INVALID, taskDefinitionToUpdate.getName());
- }
- taskDatasourcePermissionChecker.checkPermission(loginUser, Collections.singletonList(taskDefinitionToUpdate));
- if (TimeoutFlag.CLOSE == taskDefinition.getTimeoutFlag()) {
- taskDefinition.setTimeoutNotifyStrategy(null);
- }
- if (taskDefinition.equals(taskDefinitionToUpdate)) {
- log.warn("Task definition does not need update because no change, taskDefinitionCode:{}.", taskCode);
- return null;
- }
- Integer version = taskDefinitionLogMapper.queryMaxVersionForDefinition(taskCode);
- if (version == null || version == 0) {
- log.error("Max version task definitionLog can not be found in database, taskDefinitionCode:{}.",
- taskCode);
- throw new ServiceException(Status.DATA_IS_NOT_VALID, taskCode);
- }
- Date now = new Date();
- taskDefinitionToUpdate.setCode(taskCode);
- taskDefinitionToUpdate.setId(taskDefinition.getId());
- taskDefinitionToUpdate.setProjectCode(projectCode);
- taskDefinitionToUpdate.setUserId(taskDefinition.getUserId());
- taskDefinitionToUpdate.setVersion(++version);
- taskDefinitionToUpdate.setTaskType(taskDefinitionToUpdate.getTaskType().toUpperCase());
- taskDefinitionToUpdate.setUpdateTime(now);
- boolean updateSuccess = taskDefinitionDao.updateById(taskDefinitionToUpdate);
- taskDefinitionToUpdate.setOperator(loginUser.getId());
- taskDefinitionToUpdate.setOperateTime(now);
- taskDefinitionToUpdate.setCreateTime(now);
- taskDefinitionToUpdate.setId(null);
- int insert = taskDefinitionLogMapper.insert(taskDefinitionToUpdate);
- if (!updateSuccess || insert != 1) {
- log.error("Update task definition or definitionLog error, projectCode:{}, taskDefinitionCode:{}.",
- projectCode, taskCode);
- throw new ServiceException(Status.UPDATE_TASK_DEFINITION_ERROR);
- }
- log.info(
- "Update task definition and definitionLog complete, projectCode:{}, taskDefinitionCode:{}, newTaskVersion:{}.",
- projectCode, taskCode, taskDefinitionToUpdate.getVersion());
- // update workflow task relation
- List<WorkflowTaskRelation> workflowTaskRelations = workflowTaskRelationDao
- .queryWorkflowTaskRelationByTaskCodeAndTaskVersion(taskDefinitionToUpdate.getCode(),
- taskDefinition.getVersion());
- if (CollectionUtils.isNotEmpty(workflowTaskRelations)) {
- Map<Long, List<WorkflowTaskRelation>> workflowTaskRelationGroupList = workflowTaskRelations.stream()
- .collect(Collectors.groupingBy(WorkflowTaskRelation::getWorkflowDefinitionCode));
- for (Map.Entry<Long, List<WorkflowTaskRelation>> workflowTaskRelationMap : workflowTaskRelationGroupList
- .entrySet()) {
- Long workflowDefinitionCode = workflowTaskRelationMap.getKey();
- int workflowDefinitionVersion =
- workflowDefinitionLogMapper.queryMaxVersionForDefinition(workflowDefinitionCode)
- + 1;
- List<WorkflowTaskRelation> workflowTaskRelationList = workflowTaskRelationMap.getValue();
- for (WorkflowTaskRelation workflowTaskRelation : workflowTaskRelationList) {
- if (taskCode == workflowTaskRelation.getPreTaskCode()) {
- workflowTaskRelation.setPreTaskVersion(version);
- } else if (taskCode == workflowTaskRelation.getPostTaskCode()) {
- workflowTaskRelation.setPostTaskVersion(version);
- }
- workflowTaskRelation.setWorkflowDefinitionVersion(workflowDefinitionVersion);
- if (!workflowTaskRelationDao.updateWorkflowTaskRelationTaskVersion(workflowTaskRelation)) {
- log.error("batch update workflow task relation error, projectCode:{}, taskDefinitionCode:{}.",
- projectCode, taskCode);
- throw new ServiceException(Status.WORKFLOW_TASK_RELATION_BATCH_UPDATE_ERROR);
- }
- WorkflowTaskRelationLog workflowTaskRelationLog = new WorkflowTaskRelationLog(workflowTaskRelation);
- workflowTaskRelationLog.setOperator(loginUser.getId());
- workflowTaskRelationLog.setId(null);
- workflowTaskRelationLog.setOperateTime(now);
- int insertWorkflowTaskRelationLogCount = workflowTaskRelationLogDao.insert(workflowTaskRelationLog);
- if (insertWorkflowTaskRelationLogCount != 1) {
- log.error("batch update workflow task relation error, projectCode:{}, taskDefinitionCode:{}.",
- projectCode, taskCode);
- throw new ServiceException(Status.CREATE_WORKFLOW_TASK_RELATION_LOG_ERROR);
- }
- }
- WorkflowDefinition workflowDefinition =
- workflowDefinitionDao.queryByCode(workflowDefinitionCode).orElse(null);
- workflowDefinition.setVersion(workflowDefinitionVersion);
- workflowDefinition.setUpdateTime(now);
- workflowDefinition.setUserId(loginUser.getId());
- // update workflow definition
- boolean updateWorkflowDefinitionSuccess = workflowDefinitionDao.updateById(workflowDefinition);
- WorkflowDefinitionLog workflowDefinitionLog = new WorkflowDefinitionLog(workflowDefinition);
- workflowDefinitionLog.setOperateTime(now);
- workflowDefinitionLog.setId(null);
- workflowDefinitionLog.setOperator(loginUser.getId());
- int insertWorkflowDefinitionLogCount = workflowDefinitionLogMapper.insert(workflowDefinitionLog);
- if (!updateWorkflowDefinitionSuccess || insertWorkflowDefinitionLogCount != 1) {
- throw new ServiceException(Status.UPDATE_WORKFLOW_DEFINITION_ERROR);
- }
- }
- }
- return taskDefinitionToUpdate;
- }
-
- /**
- * update task definition and upstream
- *
- * @param loginUser login user
- * @param projectCode project code
- * @param taskCode task definition code
- * @param taskDefinitionJsonObj task definition json object
- * @param upstreamCodes upstream task codes, sep comma
- * @return updated task code
- */
- @Override
- public Long updateTaskWithUpstream(User loginUser, long projectCode, long taskCode,
- String taskDefinitionJsonObj, String upstreamCodes) {
- TaskDefinitionLog taskDefinitionToUpdate =
- updateTask(loginUser, projectCode, taskCode, taskDefinitionJsonObj);
- List<WorkflowTaskRelation> upstreamTaskRelations =
- workflowTaskRelationDao.queryUpstreamByCode(projectCode, taskCode);
- Set<Long> upstreamCodeSet =
- upstreamTaskRelations.stream().map(WorkflowTaskRelation::getPreTaskCode).collect(Collectors.toSet());
- Set<Long> upstreamTaskCodes = Collections.emptySet();
- if (StringUtils.isNotEmpty(upstreamCodes)) {
- upstreamTaskCodes = Arrays.stream(upstreamCodes.split(Constants.COMMA)).map(Long::parseLong)
- .collect(Collectors.toSet());
- }
- if (CollectionUtils.isEqualCollection(upstreamCodeSet, upstreamTaskCodes) && taskDefinitionToUpdate == null) {
- return taskCode;
- }
- Map<Long, TaskDefinition> queryUpStreamTaskCodeMap;
- if (CollectionUtils.isNotEmpty(upstreamTaskCodes)) {
- List<TaskDefinition> upstreamTaskDefinitionList = taskDefinitionDao.queryByCodes(upstreamTaskCodes);
- queryUpStreamTaskCodeMap = upstreamTaskDefinitionList.stream()
- .collect(Collectors.toMap(TaskDefinition::getCode, taskDefinition -> taskDefinition));
- // upstreamTaskCodes - queryUpStreamTaskCodeMap.keySet
- upstreamTaskCodes.removeAll(queryUpStreamTaskCodeMap.keySet());
- if (CollectionUtils.isNotEmpty(upstreamTaskCodes)) {
- String notExistTaskCodes = StringUtils.join(upstreamTaskCodes, Constants.COMMA);
- log.error("Some task definitions in parameter upstreamTaskCodes do not exist, notExistTaskCodes:{}.",
- notExistTaskCodes);
- throw new ServiceException(Status.TASK_DEFINE_NOT_EXIST, notExistTaskCodes);
- }
- } else {
- queryUpStreamTaskCodeMap = new HashMap<>();
- }
- if (MapUtils.isNotEmpty(queryUpStreamTaskCodeMap)) {
- WorkflowTaskRelation taskRelation = upstreamTaskRelations.get(0);
- List<WorkflowTaskRelation> workflowTaskRelations =
- workflowTaskRelationDao.queryByWorkflowDefinitionCode(taskRelation.getWorkflowDefinitionCode());
-
- // set upstream code list
- updateUpstreamTask(new HashSet<>(queryUpStreamTaskCodeMap.keySet()),
- taskCode, projectCode, taskRelation.getWorkflowDefinitionCode(), loginUser);
-
- List<WorkflowTaskRelation> workflowTaskRelationList = Lists.newArrayList(workflowTaskRelations);
- List<WorkflowTaskRelation> relationList = Lists.newArrayList();
- for (WorkflowTaskRelation workflowTaskRelation : workflowTaskRelationList) {
- if (workflowTaskRelation.getPostTaskCode() == taskCode) {
- if (queryUpStreamTaskCodeMap.containsKey(workflowTaskRelation.getPreTaskCode())
- && workflowTaskRelation.getPreTaskCode() != 0L) {
- queryUpStreamTaskCodeMap.remove(workflowTaskRelation.getPreTaskCode());
- } else {
- workflowTaskRelation.setPreTaskCode(0L);
- workflowTaskRelation.setPreTaskVersion(0);
- relationList.add(workflowTaskRelation);
- }
- }
- }
- workflowTaskRelationList.removeAll(relationList);
- for (Map.Entry<Long, TaskDefinition> queryUpStreamTask : queryUpStreamTaskCodeMap.entrySet()) {
- taskRelation.setPreTaskCode(queryUpStreamTask.getKey());
- taskRelation.setPreTaskVersion(queryUpStreamTask.getValue().getVersion());
- workflowTaskRelationList.add(taskRelation);
- }
- if (MapUtils.isEmpty(queryUpStreamTaskCodeMap) && CollectionUtils.isNotEmpty(workflowTaskRelationList)) {
- workflowTaskRelationList.add(workflowTaskRelationList.get(0));
- }
- }
- log.info(
- "Update task with upstream tasks complete, projectCode:{}, taskDefinitionCode:{}, upstreamTaskCodes:{}.",
- projectCode, taskCode, upstreamTaskCodes);
- return taskCode;
- }
-
- private void updateUpstreamTask(Set<Long> allPreTaskCodeSet, long taskCode, long projectCode,
- long workflowDefinitionCode, User loginUser) {
- // query all workflow task relation
- List<WorkflowTaskRelation> hadWorkflowTaskRelationList = workflowTaskRelationDao
- .queryUpstreamByCode(projectCode, taskCode);
- // remove pre
- Set<Long> removePreTaskSet = new HashSet<>();
- List<WorkflowTaskRelation> removePreTaskList = new ArrayList<>();
- // add pre
- Set<Long> addPreTaskSet = new HashSet<>();
- List<WorkflowTaskRelation> addPreTaskList = new ArrayList<>();
-
- List<WorkflowTaskRelationLog> workflowTaskRelationLogList = new ArrayList<>();
-
- // filter all workflow task relation
- if (CollectionUtils.isNotEmpty(hadWorkflowTaskRelationList)) {
- for (WorkflowTaskRelation workflowTaskRelation : hadWorkflowTaskRelationList) {
- if (workflowTaskRelation.getPreTaskCode() == 0) {
- continue;
- }
- // had
- if (allPreTaskCodeSet.contains(workflowTaskRelation.getPreTaskCode())) {
- allPreTaskCodeSet.remove(workflowTaskRelation.getPreTaskCode());
- } else {
- // remove
- removePreTaskSet.add(workflowTaskRelation.getPreTaskCode());
- workflowTaskRelation.setPreTaskCode(0);
- workflowTaskRelation.setPreTaskVersion(0);
- removePreTaskList.add(workflowTaskRelation);
- workflowTaskRelationLogList.add(createWorkflowTaskRelationLog(loginUser, workflowTaskRelation));
- }
- }
- }
- // add
- if (allPreTaskCodeSet.size() != 0) {
- addPreTaskSet.addAll(allPreTaskCodeSet);
- }
- // get add task code map
- allPreTaskCodeSet.add(Long.valueOf(taskCode));
- List<TaskDefinition> taskDefinitionList = taskDefinitionDao.queryByCodes(allPreTaskCodeSet);
- Map<Long, TaskDefinition> taskCodeMap = taskDefinitionList.stream().collect(Collectors
- .toMap(TaskDefinition::getCode, Function.identity(), (a, b) -> a));
-
- WorkflowDefinition workflowDefinition = workflowDefinitionDao.queryByCode(workflowDefinitionCode).orElse(null);
- TaskDefinition taskDefinition = taskCodeMap.get(taskCode);
-
- for (Long preTaskCode : addPreTaskSet) {
- TaskDefinition preTaskRelation = taskCodeMap.get(preTaskCode);
- WorkflowTaskRelation workflowTaskRelation = new WorkflowTaskRelation(
- null, workflowDefinition.getVersion(), projectCode, workflowDefinition.getCode(),
- preTaskRelation.getCode(), preTaskRelation.getVersion(),
- taskDefinition.getCode(), taskDefinition.getVersion(), ConditionType.NONE, "{}");
- addPreTaskList.add(workflowTaskRelation);
- workflowTaskRelationLogList.add(createWorkflowTaskRelationLog(loginUser, workflowTaskRelation));
- }
- int insert = 0;
- int remove = 0;
- int log = 0;
- // insert workflow task relation table data
- if (CollectionUtils.isNotEmpty(addPreTaskList)) {
- insert = workflowTaskRelationDao.batchInsert(addPreTaskList);
- }
- if (CollectionUtils.isNotEmpty(removePreTaskList)) {
- for (WorkflowTaskRelation workflowTaskRelation : removePreTaskList) {
- remove += workflowTaskRelationDao.updateById(workflowTaskRelation) ? 1 : 0;
- }
- }
- if (CollectionUtils.isNotEmpty(workflowTaskRelationLogList)) {
- log = workflowTaskRelationLogDao.batchInsert(workflowTaskRelationLogList);
- }
- if (insert + remove != log) {
- throw new RuntimeException("updateUpstreamTask error");
- }
- }
-
- private WorkflowTaskRelationLog createWorkflowTaskRelationLog(User loginUser,
- WorkflowTaskRelation workflowTaskRelation) {
- Date now = new Date();
- WorkflowTaskRelationLog workflowTaskRelationLog = new WorkflowTaskRelationLog(workflowTaskRelation);
- workflowTaskRelationLog.setOperator(loginUser.getId());
- workflowTaskRelationLog.setOperateTime(now);
- workflowTaskRelationLog.setCreateTime(now);
- workflowTaskRelationLog.setUpdateTime(now);
- return workflowTaskRelationLog;
- }
-
/**
* switch task definition
*
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 f34c1c7e81..364f89d722 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
@@ -56,13 +56,10 @@ import org.apache.dolphinscheduler.dao.repository.ProjectDao;
import org.apache.dolphinscheduler.dao.repository.TaskDefinitionDao;
import org.apache.dolphinscheduler.dao.repository.WorkflowDefinitionDao;
import org.apache.dolphinscheduler.dao.repository.WorkflowTaskRelationDao;
-import org.apache.dolphinscheduler.dao.repository.WorkflowTaskRelationLogDao;
-import org.apache.dolphinscheduler.plugin.task.api.TaskPluginManager;
import org.apache.dolphinscheduler.service.process.ProcessService;
import org.apache.dolphinscheduler.service.process.ProcessServiceImpl;
import java.util.ArrayList;
-import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -73,7 +70,6 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.InjectMocks;
import org.mockito.Mock;
-import org.mockito.MockedStatic;
import org.mockito.Mockito;
import org.mockito.junit.jupiter.MockitoExtension;
import org.mockito.junit.jupiter.MockitoSettings;
@@ -128,15 +124,11 @@ public class TaskDefinitionServiceImplTest {
@Mock
private WorkflowDefinitionDao workflowDefinitionDao;
- @Mock
- private WorkflowTaskRelationLogDao workflowTaskRelationLogDao;
-
private static final String TASK_PARAMETER =
"{\"resourceList\":[],\"localParams\":[],\"rawScript\":\"echo 1\",\"conditionResult\":{\"successNode\":[\"\"],\"failedNode\":[\"\"]},\"dependence\":{}}";;
private static final long PROJECT_CODE = 1L;
private static final long PROCESS_DEFINITION_CODE = 2L;
private static final long TASK_CODE = 3L;
- private static final String UPSTREAM_CODE = "3,5";
private static final int VERSION = 1;
private static final int RESOURCE_RATE = -1;
protected User user;
@@ -385,60 +377,6 @@ public class TaskDefinitionServiceImplTest {
Assertions.assertDoesNotThrow(() -> taskDefinitionService.getTaskDefinition(user, TASK_CODE));
}
- @Test
- public void testUpdateTaskWithUpstream() {
- try (
- MockedStatic<TaskPluginManager> taskPluginManagerMockedStatic =
- Mockito.mockStatic(TaskPluginManager.class)) {
- taskPluginManagerMockedStatic
- .when(() -> TaskPluginManager.checkTaskParameters(Mockito.any(), Mockito.any()))
- .thenReturn(true);
- String taskDefinitionJson = getTaskDefinitionJson();
- TaskDefinition taskDefinition = getTaskDefinition();
- taskDefinition.setFlag(Flag.NO);
- TaskDefinition taskDefinitionSecond = getTaskDefinition();
- taskDefinitionSecond.setCode(5);
-
- user.setUserType(UserType.ADMIN_USER);
- when(projectDao.queryByCode(PROJECT_CODE)).thenReturn(getProject());
- Mockito.doNothing().when(projectService).checkHasProjectWritePermissionThrowException(eq(user),
- eq(getProject()));
- when(taskDefinitionDao.queryByCode(TASK_CODE)).thenReturn(taskDefinition);
- when(taskDefinitionLogMapper.queryMaxVersionForDefinition(TASK_CODE)).thenReturn(1);
- when(taskDefinitionDao.updateById(Mockito.any())).thenReturn(true);
- when(taskDefinitionLogMapper.insert(Mockito.any())).thenReturn(1);
-
- when(taskDefinitionDao.queryByCodes(Mockito.anySet()))
- .thenReturn(Arrays.asList(taskDefinition, taskDefinitionSecond));
-
- when(workflowTaskRelationDao.queryUpstreamByCode(PROJECT_CODE, TASK_CODE))
- .thenReturn(getProcessTaskRelationListV2());
- when(workflowDefinitionDao.queryByCode(PROCESS_DEFINITION_CODE))
- .thenReturn(Optional.of(getProcessDefinition()));
- when(workflowTaskRelationDao.batchInsert(Mockito.anyList())).thenReturn(1);
- when(workflowTaskRelationDao.updateById(Mockito.any())).thenReturn(true);
- when(workflowTaskRelationLogDao.batchInsert(Mockito.anyList())).thenReturn(2);
- // success
- Long updatedTaskCode = taskDefinitionService.updateTaskWithUpstream(user, PROJECT_CODE, TASK_CODE,
- taskDefinitionJson, UPSTREAM_CODE);
- assertEquals(TASK_CODE, updatedTaskCode);
- Mockito.verify(taskDatasourcePermissionChecker).checkPermission(eq(user), anyList());
- user.setUserType(UserType.GENERAL_USER);
- }
- }
-
- private String getTaskDefinitionJson() {
- return "{\"name\":\"detail_up\",\"description\":\"\",\"taskType\":\"SHELL\",\"taskParams\":"
- + "\"{\\\"resourceList\\\":[],\\\"localParams\\\":[{\\\"prop\\\":\\\"datetime\\\","
- + "\\\"direct\\\":\\\"IN\\\",\\\"type\\\":\\\"VARCHAR\\\","
- + "\\\"value\\\":\\\"${system.datetime}\\\"}],\\\"rawScript\\\":\\\"echo ${datetime}\\\","
- + "\\\"conditionResult\\\":\\\"{\\\\\\\"successNode\\\\\\\":[\\\\\\\"\\\\\\\"],"
- + "\\\\\\\"failedNode\\\\\\\":[\\\\\\\"\\\\\\\"]}\\\",\\\"dependence\\\":{}}\","
- + "\"flag\":0,\"taskPriority\":0,\"workerGroup\":\"default\",\"failRetryTimes\":0,"
- + "\"failRetryInterval\":0,\"timeoutFlag\":0,\"timeoutNotifyStrategy\":0,\"timeout\":0,"
- + "\"delayTime\":0,\"resourceIds\":\"\"}";
- }
-
/**
* create admin user
*/
@@ -511,27 +449,6 @@ public class TaskDefinitionServiceImplTest {
return workflowTaskRelationList;
}
- private List<WorkflowTaskRelation> getProcessTaskRelationListV2() {
- List<WorkflowTaskRelation> workflowTaskRelationList = new ArrayList<>();
-
- WorkflowTaskRelation workflowTaskRelation = new WorkflowTaskRelation();
- fillProcessTaskRelation(workflowTaskRelation);
-
- workflowTaskRelationList.add(workflowTaskRelation);
- workflowTaskRelation = new WorkflowTaskRelation();
- fillProcessTaskRelation(workflowTaskRelation);
- workflowTaskRelation.setPreTaskCode(4L);
- workflowTaskRelationList.add(workflowTaskRelation);
- return workflowTaskRelationList;
- }
-
- private void fillProcessTaskRelation(WorkflowTaskRelation workflowTaskRelation) {
- workflowTaskRelation.setProjectCode(PROJECT_CODE);
- workflowTaskRelation.setWorkflowDefinitionCode(PROCESS_DEFINITION_CODE);
- workflowTaskRelation.setPreTaskCode(TASK_CODE);
- workflowTaskRelation.setPostTaskCode(TASK_CODE + 1L);
- }
-
private List<WorkflowTaskRelationLog> getProcessTaskRelationLogList() {
List<WorkflowTaskRelationLog> processTaskRelationLogList = new ArrayList<>();
diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/WorkflowTaskRelationMapper.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/WorkflowTaskRelationMapper.java
index a733aeaff9..917a9031fe 100644
--- a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/WorkflowTaskRelationMapper.java
+++ b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/WorkflowTaskRelationMapper.java
@@ -165,27 +165,10 @@ public interface WorkflowTaskRelationMapper extends BaseMapper<WorkflowTaskRelat
IPage<WorkflowTaskRelation> filterWorkflowTaskRelation(IPage<WorkflowTaskRelation> page,
@Param("relation") WorkflowTaskRelation workflowTaskRelation);
- /**
- * batch update workflow task relation version
- *
- * @param workflowTaskRelation workflow task relation list
- * @return update num
- */
- int updateWorkflowTaskRelationTaskVersion(@Param("workflowTaskRelation") WorkflowTaskRelation workflowTaskRelation);
-
Long queryTaskCodeByTaskName(@Param("workflowCode") Long workflowCode,
@Param("taskName") String taskName);
void deleteByWorkflowDefinitionCodeAndVersion(@Param("workflowDefinitionCode") long workflowDefinitionCode,
@Param("workflowDefinitionVersion") int workflowDefinitionVersion);
- /**
- * workflow task relation by taskCode and postTaskVersion
- *
- * @param taskCode taskCode
- * @param postTaskVersion postTaskVersion
- * @return ProcessTaskRelation
- */
- List<WorkflowTaskRelation> queryWorkflowTaskRelationByTaskCodeAndTaskVersion(@Param("taskCode") long taskCode,
- @Param("postTaskVersion") long postTaskVersion);
}
diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/WorkflowTaskRelationDao.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/WorkflowTaskRelationDao.java
index d84dd0c796..d0b161d839 100644
--- a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/WorkflowTaskRelationDao.java
+++ b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/WorkflowTaskRelationDao.java
@@ -36,12 +36,8 @@ public interface WorkflowTaskRelationDao extends IDao<WorkflowTaskRelation> {
List<WorkflowTaskRelation> queryDownstreamByWorkflowDefinitionCode(long workflowDefinitionCode);
- boolean updateWorkflowTaskRelationTaskVersion(WorkflowTaskRelation workflowTaskRelation);
-
void deleteByWorkflowDefinitionCodeAndVersion(long workflowDefinitionCode, int workflowDefinitionVersion);
- List<WorkflowTaskRelation> queryWorkflowTaskRelationByTaskCodeAndTaskVersion(long taskCode, long postTaskVersion);
-
List<WorkflowTaskRelation> queryByCode(long projectCode, long workflowDefinitionCode, long preTaskCode,
long postTaskCode);
}
diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/impl/WorkflowTaskRelationDaoImpl.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/impl/WorkflowTaskRelationDaoImpl.java
index 79d84f84c1..c8a9281ca5 100644
--- a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/impl/WorkflowTaskRelationDaoImpl.java
+++ b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/impl/WorkflowTaskRelationDaoImpl.java
@@ -69,22 +69,11 @@ public class WorkflowTaskRelationDaoImpl extends BaseDao<WorkflowTaskRelation, W
return mybatisMapper.queryDownstreamByWorkflowDefinitionCode(workflowDefinitionCode);
}
- @Override
- public boolean updateWorkflowTaskRelationTaskVersion(WorkflowTaskRelation workflowTaskRelation) {
- return mybatisMapper.updateWorkflowTaskRelationTaskVersion(workflowTaskRelation) > 0;
- }
-
@Override
public void deleteByWorkflowDefinitionCodeAndVersion(long workflowDefinitionCode, int workflowDefinitionVersion) {
mybatisMapper.deleteByWorkflowDefinitionCodeAndVersion(workflowDefinitionCode, workflowDefinitionVersion);
}
- @Override
- public List<WorkflowTaskRelation> queryWorkflowTaskRelationByTaskCodeAndTaskVersion(long taskCode,
- long postTaskVersion) {
- return mybatisMapper.queryWorkflowTaskRelationByTaskCodeAndTaskVersion(taskCode, postTaskVersion);
- }
-
@Override
public List<WorkflowTaskRelation> queryByCode(long projectCode, long workflowDefinitionCode, long preTaskCode,
long postTaskCode) {
diff --git a/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/WorkflowTaskRelationMapper.xml b/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/WorkflowTaskRelationMapper.xml
index 7fa82d3892..a46d221242 100644
--- a/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/WorkflowTaskRelationMapper.xml
+++ b/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/WorkflowTaskRelationMapper.xml
@@ -183,14 +183,6 @@
where r.workflow_definition_code = #{workflowCode}
and d.name = #{taskName}
</select>
- <update id="updateWorkflowTaskRelationTaskVersion">
- update t_ds_workflow_task_relation
- set pre_task_version=#{workflowTaskRelation.preTaskVersion},
- post_task_version=#{workflowTaskRelation.postTaskVersion},
- workflow_definition_version=#{workflowTaskRelation.workflowDefinitionVersion}
- where id = #{workflowTaskRelation.id}
- </update>
-
<delete id="deleteByWorkflowDefinitionCodeAndVersion">
delete
from t_ds_workflow_task_relation
@@ -198,18 +190,4 @@
and workflow_definition_version = #{workflowDefinitionVersion}
</delete>
- <select id="queryWorkflowTaskRelationByTaskCodeAndTaskVersion" resultType="org.apache.dolphinscheduler.dao.entity.WorkflowTaskRelation">
- select
- <include refid="baseSql"/>
- from t_ds_workflow_task_relation
- WHERE workflow_definition_code in (
- SELECT
- workflow_definition_code
- FROM
- t_ds_workflow_task_relation
- WHERE
- post_task_code = #{taskCode}
- and post_task_version = #{postTaskVersion}
- )
- </select>
</mapper>