Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[Fix-16202][api] Fix the issue of duplicate predecessor tasks #16204

Open
wants to merge 5 commits into
base: dev
Choose a base branch
from
Open

[Fix-16202][api] Fix the issue of duplicate predecessor tasks #16204

Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
0926ffa
Fix bug duplicate predecessor tasks #16202
xiaoqisu777 Jun 24, 2024
421599d
Merge branch 'dev' into patch-1
xiaoqisu777 Jun 25, 2024
2f8887c
Merge branch 'dev' into patch-1
SbloodyS Jun 25, 2024
7b0d4ae
Merge branch 'dev' into patch-1
davidzollo Jun 26, 2024
6d4a885
Merge branch 'dev' into patch-1
xiaoqisu777 Sep 3, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
173 changes: 173 additions & 0 deletions ...src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskDefinitionServiceImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -174,6 +174,179 @@ private WorkflowDefinition updateWorkflowLocation(User user, WorkflowDefinition
workflowUpdateRequest);
}

/**
* Create resource task definition
*
* @param loginUser login user
* @param taskCreateRequest task definition json
* @return new TaskDefinition have created
*/
@Override
@Transactional
public TaskDefinition createTaskDefinitionV2(User loginUser,
TaskCreateRequest taskCreateRequest) {
TaskDefinition taskDefinition = taskCreateRequest.convert2TaskDefinition();
ProcessDefinition processDefinition = processDefinitionMapper.queryByCode(taskCreateRequest.getWorkflowCode());
if (processDefinition == null) {
throw new ServiceException(Status.PROCESS_DEFINE_NOT_EXIST, taskCreateRequest.getWorkflowCode());
}
// Add project code from process definition if not exists
if (taskDefinition.getProjectCode() == 0L) {
taskDefinition.setProjectCode(processDefinition.getProjectCode());
}
this.checkTaskDefinitionValid(loginUser, taskDefinition, TASK_DEFINITION_CREATE);

long taskDefinitionCode;
try {
taskDefinitionCode = CodeGenerateUtils.genCode();
} catch (CodeGenerateException e) {
throw new ServiceException(Status.INTERNAL_SERVER_ERROR_ARGS);
}
taskDefinition.setCode(taskDefinitionCode);

int create = taskDefinitionMapper.insert(taskDefinition);
if (create <= 0) {
throw new ServiceException(Status.CREATE_TASK_DEFINITION_ERROR);
}
this.persist2TaskDefinitionLog(loginUser, taskDefinition);

// update related objects: task relationship, workflow's location(need to set to null and front-end will auto
// format it)
this.updateTaskUpstreams(loginUser, taskCreateRequest.getWorkflowCode(), taskDefinition.getCode(),
taskCreateRequest.getUpstreamTasksCodes());
this.updateWorkflowLocation(loginUser, processDefinition);
return taskDefinition;
}

/**
* create single task definition that binds the workflow
*
* @param loginUser login user
* @param projectCode project code
* @param processDefinitionCode process definition code
* @param taskDefinitionJsonObj task definition json object
* @param upstreamCodes upstream task codes, sep comma
* @return create result code
*/
@Transactional
@Override
public Map<String, Object> createTaskBindsWorkFlow(User loginUser,
long projectCode,
long processDefinitionCode,
String taskDefinitionJsonObj,
String upstreamCodes) {
Project project = projectMapper.queryByCode(projectCode);
// check if user have write perm for project
Map<String, Object> result = new HashMap<>();
boolean hasProjectAndWritePerm = projectService.hasProjectAndWritePerm(loginUser, project, result);
if (!hasProjectAndWritePerm) {
return result;
}
ProcessDefinition processDefinition = processDefinitionMapper.queryByCode(processDefinitionCode);
if (processDefinition == null || projectCode != processDefinition.getProjectCode()) {
log.error("Process definition does not exist, processDefinitionCode:{}.", processDefinitionCode);
putMsg(result, Status.PROCESS_DEFINE_NOT_EXIST, String.valueOf(processDefinitionCode));
return result;
}
if (processDefinition.getReleaseState() == ReleaseState.ONLINE) {
log.warn("Task definition can not be created due to process definition is {}, processDefinitionCode:{}.",
ReleaseState.ONLINE.getDescp(), processDefinition.getCode());
putMsg(result, Status.PROCESS_DEFINE_STATE_ONLINE, String.valueOf(processDefinitionCode));
return result;
}
TaskDefinitionLog taskDefinition = JSONUtils.parseObject(taskDefinitionJsonObj, TaskDefinitionLog.class);
if (taskDefinition == null) {
log.warn("Parameter taskDefinitionJsonObj is invalid json.");
putMsg(result, Status.DATA_IS_NOT_VALID, taskDefinitionJsonObj);
return result;
}
if (!checkTaskParameters(taskDefinition.getTaskType(), taskDefinition.getTaskParams())) {
putMsg(result, Status.PROCESS_NODE_S_PARAMETER_INVALID, taskDefinition.getName());
return result;
}
long taskCode = taskDefinition.getCode();
if (taskCode == 0) {
taskDefinition.setCode(CodeGenerateUtils.genCode());
}
List<ProcessTaskRelationLog> processTaskRelationLogList =
processTaskRelationMapper.queryByProcessCode(processDefinitionCode)
.stream()
.map(ProcessTaskRelationLog::new)
.collect(Collectors.toList());

if (StringUtils.isNotBlank(upstreamCodes)) {
Set<Long> upstreamTaskCodes = Arrays.stream(upstreamCodes.split(Constants.COMMA)).map(Long::parseLong)
.collect(Collectors.toSet());
List<TaskDefinition> upstreamTaskDefinitionList = taskDefinitionMapper.queryByCodeList(upstreamTaskCodes);
Set<Long> queryUpStreamTaskCodes =
upstreamTaskDefinitionList.stream().map(TaskDefinition::getCode).collect(Collectors.toSet());
// upstreamTaskCodes - queryUpStreamTaskCodes
Set<Long> diffCode = upstreamTaskCodes.stream().filter(code -> !queryUpStreamTaskCodes.contains(code))
.collect(Collectors.toSet());
if (CollectionUtils.isNotEmpty(diffCode)) {
String taskCodes = StringUtils.join(diffCode, Constants.COMMA);
log.error("Some task definitions with parameter upstreamCodes do not exist, taskDefinitionCodes:{}.",
taskCodes);
putMsg(result, Status.TASK_DEFINE_NOT_EXIST, taskCodes);
return result;
}
for (TaskDefinition upstreamTask : upstreamTaskDefinitionList) {
ProcessTaskRelationLog processTaskRelationLog = new ProcessTaskRelationLog();
processTaskRelationLog.setPreTaskCode(upstreamTask.getCode());
processTaskRelationLog.setPreTaskVersion(upstreamTask.getVersion());
processTaskRelationLog.setPostTaskCode(taskCode);
processTaskRelationLog.setPostTaskVersion(Constants.VERSION_FIRST);
processTaskRelationLog.setConditionType(ConditionType.NONE);
processTaskRelationLog.setConditionParams("{}");
processTaskRelationLogList.add(processTaskRelationLog);
}
} else {
ProcessTaskRelationLog processTaskRelationLog = new ProcessTaskRelationLog();
processTaskRelationLog.setPreTaskCode(0);
processTaskRelationLog.setPreTaskVersion(0);
processTaskRelationLog.setPostTaskCode(taskCode);
processTaskRelationLog.setPostTaskVersion(Constants.VERSION_FIRST);
processTaskRelationLog.setConditionType(ConditionType.NONE);
processTaskRelationLog.setConditionParams("{}");
processTaskRelationLogList.add(processTaskRelationLog);
}
int insertResult = processService.saveTaskRelation(loginUser, projectCode, processDefinition.getCode(),
processDefinition.getVersion()+1,
processTaskRelationLogList, Lists.newArrayList(), Boolean.TRUE);
if (insertResult != Constants.EXIT_CODE_SUCCESS) {
log.error(
"Save new version process task relations error, processDefinitionCode:{}, processDefinitionVersion:{}.",
processDefinition.getCode(), processDefinition.getVersion());
putMsg(result, Status.CREATE_PROCESS_TASK_RELATION_ERROR);
throw new ServiceException(Status.CREATE_PROCESS_TASK_RELATION_ERROR);
} else
log.info(
"Save new version process task relations complete, processDefinitionCode:{}, processDefinitionVersion:{}.",
processDefinition.getCode(), processDefinition.getVersion());

int saveTaskResult =
processService.saveTaskDefine(loginUser, projectCode, Lists.newArrayList(taskDefinition), Boolean.TRUE);
if (saveTaskResult == Constants.DEFINITION_FAILURE) {
log.error("Save task definition error, projectCode:{}, taskDefinitionCode:{}.", projectCode,
taskDefinition.getCode());
putMsg(result, Status.CREATE_TASK_DEFINITION_ERROR);
throw new ServiceException(Status.CREATE_TASK_DEFINITION_ERROR);
} else
log.info("Save task definition complete, projectCode:{}, taskDefinitionCode:{}.", projectCode,
taskDefinition.getCode());

int insertVersion = processService.saveProcessDefine(loginUser, processDefinition, Boolean.TRUE, Boolean.TRUE);
if (insertVersion == 0) {
throw new ServiceException(Status.CREATE_PROCESS_DEFINITION_ERROR);
} else
log.info("Save process definition complete, processCode:{}, processVersion:{}.",
processDefinition.getCode(), insertVersion);

putMsg(result, Status.SUCCESS);
result.put(Constants.DATA_LIST, taskDefinition);
return result;
}

/**
* query task definition
*
Expand Down
Loading