github
/
DolphinScheduler
mirror of https://gitee.com/dolphinscheduler/DolphinScheduler.git
1
0
Fork 0
Code Issues Releases Wiki Activity
Browse Source

Fix update TaskDefinition error (#12060)

3.2.0-release
Wenjun Ruan 2 years ago committed by GitHub
parent
0bf639b851
commit
26dad5e2a8
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 221 additions and 120 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 292
      dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskDefinitionServiceImpl.java
  2. 45
      dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessServiceImpl.java

292
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskDefinitionServiceImpl.java
Unescape View File

@ -17,8 +17,16 @@
package org.apache.dolphinscheduler.api.service.impl; 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_DEFINITION_CREATE;
import static org.apache.dolphinscheduler.api.constants.ApiFuncIdentificationConstant.TASK_DEFINITION_DELETE;
import static org.apache.dolphinscheduler.api.constants.ApiFuncIdentificationConstant.TASK_DEFINITION_UPDATE;
import static org.apache.dolphinscheduler.api.constants.ApiFuncIdentificationConstant.TASK_VERSION_VIEW;
import static org.apache.dolphinscheduler.api.constants.ApiFuncIdentificationConstant.WORKFLOW_SWITCH_TO_THIS_VERSION;
import org.apache.dolphinscheduler.api.enums.Status; import org.apache.dolphinscheduler.api.enums.Status;
import org.apache.dolphinscheduler.api.exceptions.ServiceException; import org.apache.dolphinscheduler.api.exceptions.ServiceException;
import org.apache.dolphinscheduler.api.permission.PermissionCheck;
import org.apache.dolphinscheduler.api.service.ProjectService; import org.apache.dolphinscheduler.api.service.ProjectService;
import org.apache.dolphinscheduler.api.service.TaskDefinitionService; import org.apache.dolphinscheduler.api.service.TaskDefinitionService;
import org.apache.dolphinscheduler.api.utils.PageInfo; import org.apache.dolphinscheduler.api.utils.PageInfo;
@ -46,7 +54,6 @@ import org.apache.dolphinscheduler.dao.mapper.ProjectMapper;
import org.apache.dolphinscheduler.dao.mapper.TaskDefinitionLogMapper; import org.apache.dolphinscheduler.dao.mapper.TaskDefinitionLogMapper;
import org.apache.dolphinscheduler.dao.mapper.TaskDefinitionMapper; import org.apache.dolphinscheduler.dao.mapper.TaskDefinitionMapper;
import org.apache.dolphinscheduler.plugin.task.api.parameters.ParametersNode; import org.apache.dolphinscheduler.plugin.task.api.parameters.ParametersNode;
import org.apache.dolphinscheduler.api.permission.PermissionCheck;
import org.apache.dolphinscheduler.service.process.ProcessService; import org.apache.dolphinscheduler.service.process.ProcessService;
import org.apache.dolphinscheduler.service.task.TaskPluginManager; import org.apache.dolphinscheduler.service.task.TaskPluginManager;
@ -73,8 +80,6 @@ import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page; import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.google.common.collect.Lists; import com.google.common.collect.Lists;
import static org.apache.dolphinscheduler.api.constants.ApiFuncIdentificationConstant.*;
/** /**
* task definition service impl * task definition service impl
*/ */
@ -122,8 +127,9 @@ public class TaskDefinitionServiceImpl extends BaseServiceImpl implements TaskDe
long projectCode, long projectCode,
String taskDefinitionJson) { String taskDefinitionJson) {
Project project = projectMapper.queryByCode(projectCode); Project project = projectMapper.queryByCode(projectCode);
//check user access for project // check user access for project
Map<String, Object> result = projectService.checkProjectAndAuth(loginUser, project, projectCode, TASK_DEFINITION_CREATE); Map<String, Object> result =
projectService.checkProjectAndAuth(loginUser, project, projectCode, TASK_DEFINITION_CREATE);
if (result.get(Constants.STATUS) != Status.SUCCESS) { if (result.get(Constants.STATUS) != Status.SUCCESS) {
return result; return result;
} }
@ -153,7 +159,8 @@ public class TaskDefinitionServiceImpl extends BaseServiceImpl implements TaskDe
} }
Map<String, Object> resData = new HashMap<>(); Map<String, Object> resData = new HashMap<>();
resData.put("total", taskDefinitionLogs.size()); resData.put("total", taskDefinitionLogs.size());
resData.put("code", StringUtils.join(taskDefinitionLogs.stream().map(TaskDefinition::getCode).collect(Collectors.toList()), ",")); resData.put("code", StringUtils
.join(taskDefinitionLogs.stream().map(TaskDefinition::getCode).collect(Collectors.toList()), ","));
putMsg(result, Status.SUCCESS); putMsg(result, Status.SUCCESS);
result.put(Constants.DATA_LIST, resData); result.put(Constants.DATA_LIST, resData);
return result; return result;
@ -177,8 +184,9 @@ public class TaskDefinitionServiceImpl extends BaseServiceImpl implements TaskDe
String taskDefinitionJsonObj, String taskDefinitionJsonObj,
String upstreamCodes) { String upstreamCodes) {
Project project = projectMapper.queryByCode(projectCode); Project project = projectMapper.queryByCode(projectCode);
//check user access for project // check user access for project
Map<String, Object> result = projectService.checkProjectAndAuth(loginUser, project, projectCode,TASK_DEFINITION_CREATE); Map<String, Object> result =
projectService.checkProjectAndAuth(loginUser, project, projectCode, TASK_DEFINITION_CREATE);
if (result.get(Constants.STATUS) != Status.SUCCESS) { if (result.get(Constants.STATUS) != Status.SUCCESS) {
return result; return result;
} }
@ -211,25 +219,27 @@ public class TaskDefinitionServiceImpl extends BaseServiceImpl implements TaskDe
} }
long taskCode = taskDefinition.getCode(); long taskCode = taskDefinition.getCode();
if (taskCode == 0) { if (taskCode == 0) {
try { taskDefinition.setCode(CodeGenerateUtils.getInstance().genCode());
taskCode = CodeGenerateUtils.getInstance().genCode();
taskDefinition.setCode(taskCode);
} catch (CodeGenerateException e) {
logger.error("Generate task definition code error.", e);
putMsg(result, Status.INTERNAL_SERVER_ERROR_ARGS, taskDefinitionJsonObj);
return result;
}
} }
List<ProcessTaskRelationLog> processTaskRelationLogList = Lists.newArrayList(); List<ProcessTaskRelationLog> processTaskRelationLogList =
processTaskRelationMapper.queryByProcessCode(projectCode, processDefinitionCode)
.stream()
.map(ProcessTaskRelationLog::new)
.collect(Collectors.toList());
if (StringUtils.isNotBlank(upstreamCodes)) { if (StringUtils.isNotBlank(upstreamCodes)) {
Set<Long> upstreamTaskCodes = Arrays.stream(upstreamCodes.split(Constants.COMMA)).map(Long::parseLong).collect(Collectors.toSet()); Set<Long> upstreamTaskCodes = Arrays.stream(upstreamCodes.split(Constants.COMMA)).map(Long::parseLong)
.collect(Collectors.toSet());
List<TaskDefinition> upstreamTaskDefinitionList = taskDefinitionMapper.queryByCodeList(upstreamTaskCodes); List<TaskDefinition> upstreamTaskDefinitionList = taskDefinitionMapper.queryByCodeList(upstreamTaskCodes);
Set<Long> queryUpStreamTaskCodes = upstreamTaskDefinitionList.stream().map(TaskDefinition::getCode).collect(Collectors.toSet()); Set<Long> queryUpStreamTaskCodes =
upstreamTaskDefinitionList.stream().map(TaskDefinition::getCode).collect(Collectors.toSet());
// upstreamTaskCodes - queryUpStreamTaskCodes // upstreamTaskCodes - queryUpStreamTaskCodes
Set<Long> diffCode = upstreamTaskCodes.stream().filter(code -> !queryUpStreamTaskCodes.contains(code)).collect(Collectors.toSet()); Set<Long> diffCode = upstreamTaskCodes.stream().filter(code -> !queryUpStreamTaskCodes.contains(code))
.collect(Collectors.toSet());
if (!diffCode.isEmpty()) { if (!diffCode.isEmpty()) {
String taskCodes = StringUtils.join(diffCode, Constants.COMMA); String taskCodes = StringUtils.join(diffCode, Constants.COMMA);
logger.error("Some task definitions with parameter upstreamCodes do not exist, taskDefinitionCodes:{}.", taskCodes); logger.error("Some task definitions with parameter upstreamCodes do not exist, taskDefinitionCodes:{}.",
taskCodes);
putMsg(result, Status.TASK_DEFINE_NOT_EXIST, taskCodes); putMsg(result, Status.TASK_DEFINE_NOT_EXIST, taskCodes);
return result; return result;
} }
@ -243,10 +253,6 @@ public class TaskDefinitionServiceImpl extends BaseServiceImpl implements TaskDe
processTaskRelationLog.setConditionParams("{}"); processTaskRelationLog.setConditionParams("{}");
processTaskRelationLogList.add(processTaskRelationLog); processTaskRelationLogList.add(processTaskRelationLog);
} }
List<ProcessTaskRelation> processTaskRelationList = processTaskRelationMapper.queryByProcessCode(projectCode, processDefinitionCode);
if (!processTaskRelationList.isEmpty()) {
processTaskRelationLogList.addAll(processTaskRelationList.stream().map(ProcessTaskRelationLog::new).collect(Collectors.toList()));
}
} else { } else {
ProcessTaskRelationLog processTaskRelationLog = new ProcessTaskRelationLog(); ProcessTaskRelationLog processTaskRelationLog = new ProcessTaskRelationLog();
processTaskRelationLog.setPreTaskCode(0); processTaskRelationLog.setPreTaskCode(0);
@ -257,22 +263,30 @@ public class TaskDefinitionServiceImpl extends BaseServiceImpl implements TaskDe
processTaskRelationLog.setConditionParams("{}"); processTaskRelationLog.setConditionParams("{}");
processTaskRelationLogList.add(processTaskRelationLog); processTaskRelationLogList.add(processTaskRelationLog);
} }
int insertResult = processService.saveTaskRelation(loginUser, projectCode, processDefinition.getCode(), processDefinition.getVersion(), int insertResult = processService.saveTaskRelation(loginUser, projectCode, processDefinition.getCode(),
processDefinition.getVersion(),
processTaskRelationLogList, Lists.newArrayList(), Boolean.TRUE); processTaskRelationLogList, Lists.newArrayList(), Boolean.TRUE);
if (insertResult != Constants.EXIT_CODE_SUCCESS) { if (insertResult != Constants.EXIT_CODE_SUCCESS) {
logger.error("Save new version process task relations error, processDefinitionCode:{}, processDefinitionVersion:{}.", processDefinition.getCode(), processDefinition.getVersion()); logger.error(
"Save new version process task relations error, processDefinitionCode:{}, processDefinitionVersion:{}.",
processDefinition.getCode(), processDefinition.getVersion());
putMsg(result, Status.CREATE_PROCESS_TASK_RELATION_ERROR); putMsg(result, Status.CREATE_PROCESS_TASK_RELATION_ERROR);
throw new ServiceException(Status.CREATE_PROCESS_TASK_RELATION_ERROR); throw new ServiceException(Status.CREATE_PROCESS_TASK_RELATION_ERROR);
} else } else
logger.info("Save new version process task relations complete, processDefinitionCode:{}, processDefinitionVersion:{}.", processDefinition.getCode(), processDefinition.getVersion()); logger.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); int saveTaskResult =
processService.saveTaskDefine(loginUser, projectCode, Lists.newArrayList(taskDefinition), Boolean.TRUE);
if (saveTaskResult == Constants.DEFINITION_FAILURE) { if (saveTaskResult == Constants.DEFINITION_FAILURE) {
logger.error("Save task definition error, projectCode:{}, taskDefinitionCode:{}.", projectCode, taskDefinition.getCode()); logger.error("Save task definition error, projectCode:{}, taskDefinitionCode:{}.", projectCode,
taskDefinition.getCode());
putMsg(result, Status.CREATE_TASK_DEFINITION_ERROR); putMsg(result, Status.CREATE_TASK_DEFINITION_ERROR);
throw new ServiceException(Status.CREATE_TASK_DEFINITION_ERROR); throw new ServiceException(Status.CREATE_TASK_DEFINITION_ERROR);
} else } else
logger.info("Save task definition complete, projectCode:{}, taskDefinitionCode:{}.", projectCode, taskDefinition.getCode()); logger.info("Save task definition complete, projectCode:{}, taskDefinitionCode:{}.", projectCode,
taskDefinition.getCode());
putMsg(result, Status.SUCCESS); putMsg(result, Status.SUCCESS);
result.put(Constants.DATA_LIST, taskDefinition); result.put(Constants.DATA_LIST, taskDefinition);
return result; return result;
@ -287,10 +301,12 @@ public class TaskDefinitionServiceImpl extends BaseServiceImpl implements TaskDe
* @param taskName task name * @param taskName task name
*/ */
@Override @Override
public Map<String, Object> queryTaskDefinitionByName(User loginUser, long projectCode, long processCode, String taskName) { public Map<String, Object> queryTaskDefinitionByName(User loginUser, long projectCode, long processCode,
String taskName) {
Project project = projectMapper.queryByCode(projectCode); Project project = projectMapper.queryByCode(projectCode);
//check user access for project // check user access for project
Map<String, Object> result = projectService.checkProjectAndAuth(loginUser, project, projectCode,TASK_DEFINITION); Map<String, Object> result =
projectService.checkProjectAndAuth(loginUser, project, projectCode, TASK_DEFINITION);
if (result.get(Constants.STATUS) != Status.SUCCESS) { if (result.get(Constants.STATUS) != Status.SUCCESS) {
return result; return result;
} }
@ -318,8 +334,9 @@ public class TaskDefinitionServiceImpl extends BaseServiceImpl implements TaskDe
@Override @Override
public Map<String, Object> deleteTaskDefinitionByCode(User loginUser, long projectCode, long taskCode) { public Map<String, Object> deleteTaskDefinitionByCode(User loginUser, long projectCode, long taskCode) {
Project project = projectMapper.queryByCode(projectCode); Project project = projectMapper.queryByCode(projectCode);
//check user access for project // check user access for project
Map<String, Object> result = projectService.checkProjectAndAuth(loginUser, project, projectCode,TASK_DEFINITION_DELETE); Map<String, Object> result =
projectService.checkProjectAndAuth(loginUser, project, projectCode, TASK_DEFINITION_DELETE);
if (result.get(Constants.STATUS) != Status.SUCCESS) { if (result.get(Constants.STATUS) != Status.SUCCESS) {
return result; return result;
} }
@ -335,33 +352,42 @@ public class TaskDefinitionServiceImpl extends BaseServiceImpl implements TaskDe
return result; return result;
} }
if (processService.isTaskOnline(taskCode) && taskDefinition.getFlag() == Flag.YES) { if (processService.isTaskOnline(taskCode) && taskDefinition.getFlag() == Flag.YES) {
logger.warn("Task definition can not be deleted due to task state online, taskDefinitionCode:{}.", taskCode); logger.warn("Task definition can not be deleted due to task state online, taskDefinitionCode:{}.",
taskCode);
putMsg(result, Status.TASK_DEFINE_STATE_ONLINE, taskCode); putMsg(result, Status.TASK_DEFINE_STATE_ONLINE, taskCode);
return result; return result;
} }
List<ProcessTaskRelation> processTaskRelationList = processTaskRelationMapper.queryDownstreamByTaskCode(taskCode); List<ProcessTaskRelation> processTaskRelationList =
processTaskRelationMapper.queryDownstreamByTaskCode(taskCode);
if (!processTaskRelationList.isEmpty()) { if (!processTaskRelationList.isEmpty()) {
Set<Long> postTaskCodes = processTaskRelationList Set<Long> postTaskCodes = processTaskRelationList
.stream() .stream()
.map(ProcessTaskRelation::getPostTaskCode) .map(ProcessTaskRelation::getPostTaskCode)
.collect(Collectors.toSet()); .collect(Collectors.toSet());
String postTaskCodesStr = StringUtils.join(postTaskCodes, ","); String postTaskCodesStr = StringUtils.join(postTaskCodes, ",");
logger.warn("Task definition can not be deleted due to downstream tasks, taskDefinitionCode:{}, downstreamTaskCodes:{}", logger.warn(
"Task definition can not be deleted due to downstream tasks, taskDefinitionCode:{}, downstreamTaskCodes:{}",
taskCode, postTaskCodesStr); taskCode, postTaskCodesStr);
putMsg(result, Status.TASK_HAS_DOWNSTREAM, postTaskCodesStr); putMsg(result, Status.TASK_HAS_DOWNSTREAM, postTaskCodesStr);
return result; return result;
} }
int delete = taskDefinitionMapper.deleteByCode(taskCode); int delete = taskDefinitionMapper.deleteByCode(taskCode);
if (delete > 0) { if (delete > 0) {
List<ProcessTaskRelation> taskRelationList = processTaskRelationMapper.queryUpstreamByCode(projectCode, taskCode); List<ProcessTaskRelation> taskRelationList =
processTaskRelationMapper.queryUpstreamByCode(projectCode, taskCode);
if (!taskRelationList.isEmpty()) { if (!taskRelationList.isEmpty()) {
logger.info("Task definition has upstream tasks, start handle them after delete task, taskDefinitionCode:{}.", taskCode); logger.info(
"Task definition has upstream tasks, start handle them after delete task, taskDefinitionCode:{}.",
taskCode);
long processDefinitionCode = taskRelationList.get(0).getProcessDefinitionCode(); long processDefinitionCode = taskRelationList.get(0).getProcessDefinitionCode();
List<ProcessTaskRelation> processTaskRelations = processTaskRelationMapper.queryByProcessCode(projectCode, processDefinitionCode); List<ProcessTaskRelation> processTaskRelations =
List<ProcessTaskRelation> relationList = processTaskRelations.stream().filter(r -> r.getPostTaskCode() != taskCode).collect(Collectors.toList()); processTaskRelationMapper.queryByProcessCode(projectCode, processDefinitionCode);
List<ProcessTaskRelation> relationList = processTaskRelations.stream()
.filter(r -> r.getPostTaskCode() != taskCode).collect(Collectors.toList());
updateDag(loginUser, result, processDefinitionCode, relationList, Lists.newArrayList()); updateDag(loginUser, result, processDefinitionCode, relationList, Lists.newArrayList());
} else { } else {
logger.info("Task definition delete complete, projectCode:{}, taskDefinitionCode:{}.", projectCode, taskCode); logger.info("Task definition delete complete, projectCode:{}, taskDefinitionCode:{}.", projectCode,
taskCode);
putMsg(result, Status.SUCCESS); putMsg(result, Status.SUCCESS);
} }
} else { } else {
@ -372,7 +398,8 @@ public class TaskDefinitionServiceImpl extends BaseServiceImpl implements TaskDe
return result; return result;
} }
private void updateDag(User loginUser, Map<String, Object> result, long processDefinitionCode, List<ProcessTaskRelation> processTaskRelationList, private void updateDag(User loginUser, Map<String, Object> result, long processDefinitionCode,
List<ProcessTaskRelation> processTaskRelationList,
List<TaskDefinitionLog> taskDefinitionLogs) { List<TaskDefinitionLog> taskDefinitionLogs) {
ProcessDefinition processDefinition = processDefinitionMapper.queryByCode(processDefinitionCode); ProcessDefinition processDefinition = processDefinitionMapper.queryByCode(processDefinitionCode);
if (processDefinition == null) { if (processDefinition == null) {
@ -381,19 +408,27 @@ public class TaskDefinitionServiceImpl extends BaseServiceImpl implements TaskDe
} }
int insertVersion = processService.saveProcessDefine(loginUser, processDefinition, Boolean.TRUE, Boolean.TRUE); int insertVersion = processService.saveProcessDefine(loginUser, processDefinition, Boolean.TRUE, Boolean.TRUE);
if (insertVersion <= 0) { if (insertVersion <= 0) {
logger.error("Update process definition error, projectCode:{}, processDefinitionCode:{}.", processDefinition.getProjectCode(), processDefinitionCode); logger.error("Update process definition error, projectCode:{}, processDefinitionCode:{}.",
processDefinition.getProjectCode(), processDefinitionCode);
throw new ServiceException(Status.UPDATE_PROCESS_DEFINITION_ERROR); throw new ServiceException(Status.UPDATE_PROCESS_DEFINITION_ERROR);
} else } else
logger.info("Save new version process definition complete, projectCode:{}, processDefinitionCode:{}, newVersion:{}.", processDefinition.getProjectCode(), processDefinitionCode, insertVersion); logger.info(
List<ProcessTaskRelationLog> relationLogs = processTaskRelationList.stream().map(ProcessTaskRelationLog::new).collect(Collectors.toList()); "Save new version process definition complete, projectCode:{}, processDefinitionCode:{}, newVersion:{}.",
int insertResult = processService.saveTaskRelation(loginUser, processDefinition.getProjectCode(), processDefinition.getCode(), processDefinition.getProjectCode(), processDefinitionCode, insertVersion);
List<ProcessTaskRelationLog> relationLogs =
processTaskRelationList.stream().map(ProcessTaskRelationLog::new).collect(Collectors.toList());
int insertResult = processService.saveTaskRelation(loginUser, processDefinition.getProjectCode(),
processDefinition.getCode(),
insertVersion, relationLogs, taskDefinitionLogs, Boolean.TRUE); insertVersion, relationLogs, taskDefinitionLogs, Boolean.TRUE);
if (insertResult == Constants.EXIT_CODE_SUCCESS) { if (insertResult == Constants.EXIT_CODE_SUCCESS) {
logger.info("Save new version task relations complete, projectCode:{}, processDefinitionCode:{}, newVersion:{}.", processDefinition.getProjectCode(), processDefinitionCode, insertVersion); logger.info(
"Save new version task relations complete, projectCode:{}, processDefinitionCode:{}, newVersion:{}.",
processDefinition.getProjectCode(), processDefinitionCode, insertVersion);
putMsg(result, Status.SUCCESS); putMsg(result, Status.SUCCESS);
result.put(Constants.DATA_LIST, processDefinition); result.put(Constants.DATA_LIST, processDefinition);
} else { } else {
logger.error("Update task relations error, projectCode:{}, processDefinitionCode:{}.", processDefinition.getProjectCode(), processDefinitionCode); logger.error("Update task relations error, projectCode:{}, processDefinitionCode:{}.",
processDefinition.getProjectCode(), processDefinitionCode);
putMsg(result, Status.UPDATE_PROCESS_DEFINITION_ERROR); putMsg(result, Status.UPDATE_PROCESS_DEFINITION_ERROR);
throw new ServiceException(Status.UPDATE_PROCESS_DEFINITION_ERROR); throw new ServiceException(Status.UPDATE_PROCESS_DEFINITION_ERROR);
} }
@ -409,18 +444,25 @@ public class TaskDefinitionServiceImpl extends BaseServiceImpl implements TaskDe
*/ */
@Transactional @Transactional
@Override @Override
public Map<String, Object> updateTaskDefinition(User loginUser, long projectCode, long taskCode, String taskDefinitionJsonObj) { public Map<String, Object> updateTaskDefinition(User loginUser, long projectCode, long taskCode,
String taskDefinitionJsonObj) {
Map<String, Object> result = new HashMap<>(); Map<String, Object> result = new HashMap<>();
TaskDefinitionLog taskDefinitionToUpdate = updateTask(loginUser, projectCode, taskCode, taskDefinitionJsonObj, result); TaskDefinitionLog taskDefinitionToUpdate =
updateTask(loginUser, projectCode, taskCode, taskDefinitionJsonObj, result);
if (taskDefinitionToUpdate == null) { if (taskDefinitionToUpdate == null) {
return result; return result;
} }
List<ProcessTaskRelation> taskRelationList = processTaskRelationMapper.queryUpstreamByCode(projectCode, taskCode); List<ProcessTaskRelation> taskRelationList =
processTaskRelationMapper.queryUpstreamByCode(projectCode, taskCode);
if (!taskRelationList.isEmpty()) { if (!taskRelationList.isEmpty()) {
logger.info("Task definition has upstream tasks, start handle them after update task, taskDefinitionCode:{}.", taskCode); logger.info(
"Task definition has upstream tasks, start handle them after update task, taskDefinitionCode:{}.",
taskCode);
long processDefinitionCode = taskRelationList.get(0).getProcessDefinitionCode(); long processDefinitionCode = taskRelationList.get(0).getProcessDefinitionCode();
List<ProcessTaskRelation> processTaskRelations = processTaskRelationMapper.queryByProcessCode(projectCode, processDefinitionCode); List<ProcessTaskRelation> processTaskRelations =
updateDag(loginUser, result, processDefinitionCode, processTaskRelations, Lists.newArrayList(taskDefinitionToUpdate)); processTaskRelationMapper.queryByProcessCode(projectCode, processDefinitionCode);
updateDag(loginUser, result, processDefinitionCode, processTaskRelations,
Lists.newArrayList(taskDefinitionToUpdate));
} }
logger.info("Update task definition complete, projectCode:{}, taskDefinitionCode:{}.", projectCode, taskCode); logger.info("Update task definition complete, projectCode:{}, taskDefinitionCode:{}.", projectCode, taskCode);
result.put(Constants.DATA_LIST, taskCode); result.put(Constants.DATA_LIST, taskCode);
@ -428,10 +470,11 @@ public class TaskDefinitionServiceImpl extends BaseServiceImpl implements TaskDe
return result; return result;
} }
private TaskDefinitionLog updateTask(User loginUser, long projectCode, long taskCode, String taskDefinitionJsonObj, Map<String, Object> result) { private TaskDefinitionLog updateTask(User loginUser, long projectCode, long taskCode, String taskDefinitionJsonObj,
Map<String, Object> result) {
Project project = projectMapper.queryByCode(projectCode); Project project = projectMapper.queryByCode(projectCode);
//check user access for project // check user access for project
result.putAll(projectService.checkProjectAndAuth(loginUser, project, projectCode,TASK_DEFINITION_UPDATE)); result.putAll(projectService.checkProjectAndAuth(loginUser, project, projectCode, TASK_DEFINITION_UPDATE));
if (result.get(Constants.STATUS) != Status.SUCCESS) { if (result.get(Constants.STATUS) != Status.SUCCESS) {
return null; return null;
} }
@ -444,12 +487,14 @@ public class TaskDefinitionServiceImpl extends BaseServiceImpl implements TaskDe
if (processService.isTaskOnline(taskCode) && taskDefinition.getFlag() == Flag.YES) { if (processService.isTaskOnline(taskCode) && taskDefinition.getFlag() == Flag.YES) {
// if stream, can update task definition without online check // if stream, can update task definition without online check
if (taskDefinition.getTaskExecuteType() != TaskExecuteType.STREAM) { if (taskDefinition.getTaskExecuteType() != TaskExecuteType.STREAM) {
logger.warn("Only {} type task can be updated without online check, taskDefinitionCode:{}.", TaskExecuteType.STREAM, taskCode); logger.warn("Only {} type task can be updated without online check, taskDefinitionCode:{}.",
TaskExecuteType.STREAM, taskCode);
putMsg(result, Status.NOT_SUPPORT_UPDATE_TASK_DEFINITION); putMsg(result, Status.NOT_SUPPORT_UPDATE_TASK_DEFINITION);
return null; return null;
} }
} }
TaskDefinitionLog taskDefinitionToUpdate = JSONUtils.parseObject(taskDefinitionJsonObj, TaskDefinitionLog.class); TaskDefinitionLog taskDefinitionToUpdate =
JSONUtils.parseObject(taskDefinitionJsonObj, TaskDefinitionLog.class);
if (taskDefinition.equals(taskDefinitionToUpdate)) { if (taskDefinition.equals(taskDefinitionToUpdate)) {
logger.warn("Task definition does not need update because no change, taskDefinitionCode:{}.", taskCode); logger.warn("Task definition does not need update because no change, taskDefinitionCode:{}.", taskCode);
return null; return null;
@ -464,13 +509,15 @@ public class TaskDefinitionServiceImpl extends BaseServiceImpl implements TaskDe
.taskParams(taskDefinitionToUpdate.getTaskParams()) .taskParams(taskDefinitionToUpdate.getTaskParams())
.dependence(taskDefinitionToUpdate.getDependence()) .dependence(taskDefinitionToUpdate.getDependence())
.build())) { .build())) {
logger.warn("Task definition parameters are invalid, taskDefinitionName:{}.", taskDefinitionToUpdate.getName()); logger.warn("Task definition parameters are invalid, taskDefinitionName:{}.",
taskDefinitionToUpdate.getName());
putMsg(result, Status.PROCESS_NODE_S_PARAMETER_INVALID, taskDefinitionToUpdate.getName()); putMsg(result, Status.PROCESS_NODE_S_PARAMETER_INVALID, taskDefinitionToUpdate.getName());
return null; return null;
} }
Integer version = taskDefinitionLogMapper.queryMaxVersionForDefinition(taskCode); Integer version = taskDefinitionLogMapper.queryMaxVersionForDefinition(taskCode);
if (version == null || version == 0) { if (version == null || version == 0) {
logger.error("Max version task definitionLog can not be found in database, taskDefinitionCode:{}.", taskCode); logger.error("Max version task definitionLog can not be found in database, taskDefinitionCode:{}.",
taskCode);
putMsg(result, Status.DATA_IS_NOT_VALID, taskCode); putMsg(result, Status.DATA_IS_NOT_VALID, taskCode);
return null; return null;
} }
@ -490,11 +537,13 @@ public class TaskDefinitionServiceImpl extends BaseServiceImpl implements TaskDe
taskDefinitionToUpdate.setId(null); taskDefinitionToUpdate.setId(null);
int insert = taskDefinitionLogMapper.insert(taskDefinitionToUpdate); int insert = taskDefinitionLogMapper.insert(taskDefinitionToUpdate);
if ((update & insert) != 1) { if ((update & insert) != 1) {
logger.error("Update task definition or definitionLog error, projectCode:{}, taskDefinitionCode:{}.", projectCode, taskCode); logger.error("Update task definition or definitionLog error, projectCode:{}, taskDefinitionCode:{}.",
projectCode, taskCode);
putMsg(result, Status.UPDATE_TASK_DEFINITION_ERROR); putMsg(result, Status.UPDATE_TASK_DEFINITION_ERROR);
throw new ServiceException(Status.UPDATE_TASK_DEFINITION_ERROR); throw new ServiceException(Status.UPDATE_TASK_DEFINITION_ERROR);
} else } else
logger.info("Update task definition and definitionLog complete, projectCode:{}, taskDefinitionCode:{}, newTaskVersion:{}.", logger.info(
"Update task definition and definitionLog complete, projectCode:{}, taskDefinitionCode:{}, newTaskVersion:{}.",
projectCode, taskCode, taskDefinitionToUpdate.getVersion()); projectCode, taskCode, taskDefinitionToUpdate.getVersion());
return taskDefinitionToUpdate; return taskDefinitionToUpdate;
} }
@ -510,17 +559,22 @@ public class TaskDefinitionServiceImpl extends BaseServiceImpl implements TaskDe
* @return update result code * @return update result code
*/ */
@Override @Override
public Map<String, Object> updateTaskWithUpstream(User loginUser, long projectCode, long taskCode, String taskDefinitionJsonObj, String upstreamCodes) { public Map<String, Object> updateTaskWithUpstream(User loginUser, long projectCode, long taskCode,
String taskDefinitionJsonObj, String upstreamCodes) {
Map<String, Object> result = new HashMap<>(); Map<String, Object> result = new HashMap<>();
TaskDefinitionLog taskDefinitionToUpdate = updateTask(loginUser, projectCode, taskCode, taskDefinitionJsonObj, result); TaskDefinitionLog taskDefinitionToUpdate =
updateTask(loginUser, projectCode, taskCode, taskDefinitionJsonObj, result);
if (result.get(Constants.STATUS) != Status.SUCCESS && taskDefinitionToUpdate == null) { if (result.get(Constants.STATUS) != Status.SUCCESS && taskDefinitionToUpdate == null) {
return result; return result;
} }
List<ProcessTaskRelation> upstreamTaskRelations = processTaskRelationMapper.queryUpstreamByCode(projectCode, taskCode); List<ProcessTaskRelation> upstreamTaskRelations =
Set<Long> upstreamCodeSet = upstreamTaskRelations.stream().map(ProcessTaskRelation::getPreTaskCode).collect(Collectors.toSet()); processTaskRelationMapper.queryUpstreamByCode(projectCode, taskCode);
Set<Long> upstreamCodeSet =
upstreamTaskRelations.stream().map(ProcessTaskRelation::getPreTaskCode).collect(Collectors.toSet());
Set<Long> upstreamTaskCodes = Collections.emptySet(); Set<Long> upstreamTaskCodes = Collections.emptySet();
if (StringUtils.isNotEmpty(upstreamCodes)) { if (StringUtils.isNotEmpty(upstreamCodes)) {
upstreamTaskCodes = Arrays.stream(upstreamCodes.split(Constants.COMMA)).map(Long::parseLong).collect(Collectors.toSet()); upstreamTaskCodes = Arrays.stream(upstreamCodes.split(Constants.COMMA)).map(Long::parseLong)
.collect(Collectors.toSet());
} }
if (CollectionUtils.isEqualCollection(upstreamCodeSet, upstreamTaskCodes) && taskDefinitionToUpdate == null) { if (CollectionUtils.isEqualCollection(upstreamCodeSet, upstreamTaskCodes) && taskDefinitionToUpdate == null) {
putMsg(result, Status.SUCCESS); putMsg(result, Status.SUCCESS);
@ -533,12 +587,14 @@ public class TaskDefinitionServiceImpl extends BaseServiceImpl implements TaskDe
Map<Long, TaskDefinition> queryUpStreamTaskCodeMap; Map<Long, TaskDefinition> queryUpStreamTaskCodeMap;
if (!upstreamTaskCodes.isEmpty()) { if (!upstreamTaskCodes.isEmpty()) {
List<TaskDefinition> upstreamTaskDefinitionList = taskDefinitionMapper.queryByCodeList(upstreamTaskCodes); List<TaskDefinition> upstreamTaskDefinitionList = taskDefinitionMapper.queryByCodeList(upstreamTaskCodes);
queryUpStreamTaskCodeMap = upstreamTaskDefinitionList.stream().collect(Collectors.toMap(TaskDefinition::getCode, taskDefinition -> taskDefinition)); queryUpStreamTaskCodeMap = upstreamTaskDefinitionList.stream()
.collect(Collectors.toMap(TaskDefinition::getCode, taskDefinition -> taskDefinition));
// upstreamTaskCodes - queryUpStreamTaskCodeMap.keySet // upstreamTaskCodes - queryUpStreamTaskCodeMap.keySet
upstreamTaskCodes.removeAll(queryUpStreamTaskCodeMap.keySet()); upstreamTaskCodes.removeAll(queryUpStreamTaskCodeMap.keySet());
if (!upstreamTaskCodes.isEmpty()) { if (!upstreamTaskCodes.isEmpty()) {
String notExistTaskCodes = StringUtils.join(upstreamTaskCodes, Constants.COMMA); String notExistTaskCodes = StringUtils.join(upstreamTaskCodes, Constants.COMMA);
logger.error("Some task definitions in parameter upstreamTaskCodes do not exist, notExistTaskCodes:{}.", notExistTaskCodes); logger.error("Some task definitions in parameter upstreamTaskCodes do not exist, notExistTaskCodes:{}.",
notExistTaskCodes);
putMsg(result, Status.TASK_DEFINE_NOT_EXIST, notExistTaskCodes); putMsg(result, Status.TASK_DEFINE_NOT_EXIST, notExistTaskCodes);
return result; return result;
} }
@ -547,12 +603,14 @@ public class TaskDefinitionServiceImpl extends BaseServiceImpl implements TaskDe
} }
if (!upstreamTaskRelations.isEmpty()) { if (!upstreamTaskRelations.isEmpty()) {
ProcessTaskRelation taskRelation = upstreamTaskRelations.get(0); ProcessTaskRelation taskRelation = upstreamTaskRelations.get(0);
List<ProcessTaskRelation> processTaskRelations = processTaskRelationMapper.queryByProcessCode(projectCode, taskRelation.getProcessDefinitionCode()); List<ProcessTaskRelation> processTaskRelations =
processTaskRelationMapper.queryByProcessCode(projectCode, taskRelation.getProcessDefinitionCode());
List<ProcessTaskRelation> processTaskRelationList = Lists.newArrayList(processTaskRelations); List<ProcessTaskRelation> processTaskRelationList = Lists.newArrayList(processTaskRelations);
List<ProcessTaskRelation> relationList = Lists.newArrayList(); List<ProcessTaskRelation> relationList = Lists.newArrayList();
for (ProcessTaskRelation processTaskRelation : processTaskRelationList) { for (ProcessTaskRelation processTaskRelation : processTaskRelationList) {
if (processTaskRelation.getPostTaskCode() == taskCode) { if (processTaskRelation.getPostTaskCode() == taskCode) {
if (queryUpStreamTaskCodeMap.containsKey(processTaskRelation.getPreTaskCode()) && processTaskRelation.getPreTaskCode() != 0L) { if (queryUpStreamTaskCodeMap.containsKey(processTaskRelation.getPreTaskCode())
&& processTaskRelation.getPreTaskCode() != 0L) {
queryUpStreamTaskCodeMap.remove(processTaskRelation.getPreTaskCode()); queryUpStreamTaskCodeMap.remove(processTaskRelation.getPreTaskCode());
} else { } else {
processTaskRelation.setPreTaskCode(0L); processTaskRelation.setPreTaskCode(0L);
@ -570,16 +628,17 @@ public class TaskDefinitionServiceImpl extends BaseServiceImpl implements TaskDe
if (queryUpStreamTaskCodeMap.isEmpty() && !processTaskRelationList.isEmpty()) { if (queryUpStreamTaskCodeMap.isEmpty() && !processTaskRelationList.isEmpty()) {
processTaskRelationList.add(processTaskRelationList.get(0)); processTaskRelationList.add(processTaskRelationList.get(0));
} }
updateDag(loginUser, result, taskRelation.getProcessDefinitionCode(), processTaskRelations, Lists.newArrayList(taskDefinitionToUpdate)); updateDag(loginUser, result, taskRelation.getProcessDefinitionCode(), processTaskRelations,
Lists.newArrayList(taskDefinitionToUpdate));
} }
logger.info("Update task with upstream tasks complete, projectCode:{}, taskDefinitionCode:{}, upstreamTaskCodes:{}.", logger.info(
"Update task with upstream tasks complete, projectCode:{}, taskDefinitionCode:{}, upstreamTaskCodes:{}.",
projectCode, taskCode, upstreamTaskCodes); projectCode, taskCode, upstreamTaskCodes);
result.put(Constants.DATA_LIST, taskCode); result.put(Constants.DATA_LIST, taskCode);
putMsg(result, Status.SUCCESS); putMsg(result, Status.SUCCESS);
return result; return result;
} }
/** /**
* switch task definition * switch task definition
* *
@ -592,13 +651,15 @@ public class TaskDefinitionServiceImpl extends BaseServiceImpl implements TaskDe
@Override @Override
public Map<String, Object> switchVersion(User loginUser, long projectCode, long taskCode, int version) { public Map<String, Object> switchVersion(User loginUser, long projectCode, long taskCode, int version) {
Project project = projectMapper.queryByCode(projectCode); Project project = projectMapper.queryByCode(projectCode);
//check user access for project // check user access for project
Map<String, Object> result = projectService.checkProjectAndAuth(loginUser, project,projectCode,WORKFLOW_SWITCH_TO_THIS_VERSION); Map<String, Object> result =
projectService.checkProjectAndAuth(loginUser, project, projectCode, WORKFLOW_SWITCH_TO_THIS_VERSION);
if (result.get(Constants.STATUS) != Status.SUCCESS) { if (result.get(Constants.STATUS) != Status.SUCCESS) {
return result; return result;
} }
if (processService.isTaskOnline(taskCode)) { if (processService.isTaskOnline(taskCode)) {
logger.warn("Task definition version can not be switched due to process definition is {}, taskDefinitionCode:{}.", logger.warn(
"Task definition version can not be switched due to process definition is {}, taskDefinitionCode:{}.",
ReleaseState.ONLINE.getDescp(), taskCode); ReleaseState.ONLINE.getDescp(), taskCode);
putMsg(result, Status.PROCESS_DEFINE_STATE_ONLINE); putMsg(result, Status.PROCESS_DEFINE_STATE_ONLINE);
return result; return result;
@ -609,20 +670,28 @@ public class TaskDefinitionServiceImpl extends BaseServiceImpl implements TaskDe
putMsg(result, Status.TASK_DEFINE_NOT_EXIST, String.valueOf(taskCode)); putMsg(result, Status.TASK_DEFINE_NOT_EXIST, String.valueOf(taskCode));
return result; return result;
} }
TaskDefinitionLog taskDefinitionUpdate = taskDefinitionLogMapper.queryByDefinitionCodeAndVersion(taskCode, version); TaskDefinitionLog taskDefinitionUpdate =
taskDefinitionLogMapper.queryByDefinitionCodeAndVersion(taskCode, version);
taskDefinitionUpdate.setUserId(loginUser.getId()); taskDefinitionUpdate.setUserId(loginUser.getId());
taskDefinitionUpdate.setUpdateTime(new Date()); taskDefinitionUpdate.setUpdateTime(new Date());
taskDefinitionUpdate.setId(taskDefinition.getId()); taskDefinitionUpdate.setId(taskDefinition.getId());
int switchVersion = taskDefinitionMapper.updateById(taskDefinitionUpdate); int switchVersion = taskDefinitionMapper.updateById(taskDefinitionUpdate);
if (switchVersion > 0) { if (switchVersion > 0) {
List<ProcessTaskRelation> taskRelationList = processTaskRelationMapper.queryUpstreamByCode(projectCode, taskCode); List<ProcessTaskRelation> taskRelationList =
processTaskRelationMapper.queryUpstreamByCode(projectCode, taskCode);
if (!taskRelationList.isEmpty()) { if (!taskRelationList.isEmpty()) {
logger.info("Task definition has upstream tasks, start handle them after switch task, taskDefinitionCode:{}.", taskCode); logger.info(
"Task definition has upstream tasks, start handle them after switch task, taskDefinitionCode:{}.",
taskCode);
long processDefinitionCode = taskRelationList.get(0).getProcessDefinitionCode(); long processDefinitionCode = taskRelationList.get(0).getProcessDefinitionCode();
List<ProcessTaskRelation> processTaskRelations = processTaskRelationMapper.queryByProcessCode(projectCode, processDefinitionCode); List<ProcessTaskRelation> processTaskRelations =
updateDag(loginUser, result, processDefinitionCode, processTaskRelations, Lists.newArrayList(taskDefinitionUpdate)); processTaskRelationMapper.queryByProcessCode(projectCode, processDefinitionCode);
updateDag(loginUser, result, processDefinitionCode, processTaskRelations,
Lists.newArrayList(taskDefinitionUpdate));
} else { } else {
logger.info("Task definition version switch complete, switch task version to {}, taskDefinitionCode:{}.", version, taskCode); logger.info(
"Task definition version switch complete, switch task version to {}, taskDefinitionCode:{}.",
version, taskCode);
putMsg(result, Status.SUCCESS); putMsg(result, Status.SUCCESS);
} }
} else { } else {
@ -641,7 +710,8 @@ public class TaskDefinitionServiceImpl extends BaseServiceImpl implements TaskDe
Result result = new Result(); Result result = new Result();
Project project = projectMapper.queryByCode(projectCode); Project project = projectMapper.queryByCode(projectCode);
// check user access for project // check user access for project
Map<String, Object> checkResult = projectService.checkProjectAndAuth(loginUser, project, projectCode,TASK_VERSION_VIEW); Map<String, Object> checkResult =
projectService.checkProjectAndAuth(loginUser, project, projectCode, TASK_VERSION_VIEW);
Status resultStatus = (Status) checkResult.get(Constants.STATUS); Status resultStatus = (Status) checkResult.get(Constants.STATUS);
if (resultStatus != Status.SUCCESS) { if (resultStatus != Status.SUCCESS) {
putMsg(result, resultStatus); putMsg(result, resultStatus);
@ -649,7 +719,8 @@ public class TaskDefinitionServiceImpl extends BaseServiceImpl implements TaskDe
} }
PageInfo<TaskDefinitionLog> pageInfo = new PageInfo<>(pageNo, pageSize); PageInfo<TaskDefinitionLog> pageInfo = new PageInfo<>(pageNo, pageSize);
Page<TaskDefinitionLog> page = new Page<>(pageNo, pageSize); Page<TaskDefinitionLog> page = new Page<>(pageNo, pageSize);
IPage<TaskDefinitionLog> taskDefinitionVersionsPaging = taskDefinitionLogMapper.queryTaskDefinitionVersionsPaging(page, taskCode, projectCode); IPage<TaskDefinitionLog> taskDefinitionVersionsPaging =
taskDefinitionLogMapper.queryTaskDefinitionVersionsPaging(page, taskCode, projectCode);
List<TaskDefinitionLog> taskDefinitionLogs = taskDefinitionVersionsPaging.getRecords(); List<TaskDefinitionLog> taskDefinitionLogs = taskDefinitionVersionsPaging.getRecords();
pageInfo.setTotalList(taskDefinitionLogs); pageInfo.setTotalList(taskDefinitionLogs);
@ -662,8 +733,9 @@ public class TaskDefinitionServiceImpl extends BaseServiceImpl implements TaskDe
@Override @Override
public Map<String, Object> deleteByCodeAndVersion(User loginUser, long projectCode, long taskCode, int version) { public Map<String, Object> deleteByCodeAndVersion(User loginUser, long projectCode, long taskCode, int version) {
Project project = projectMapper.queryByCode(projectCode); Project project = projectMapper.queryByCode(projectCode);
//check user access for project // check user access for project
Map<String, Object> result = projectService.checkProjectAndAuth(loginUser, project, projectCode,TASK_DEFINITION_DELETE); Map<String, Object> result =
projectService.checkProjectAndAuth(loginUser, project, projectCode, TASK_DEFINITION_DELETE);
if (result.get(Constants.STATUS) != Status.SUCCESS) { if (result.get(Constants.STATUS) != Status.SUCCESS) {
return result; return result;
} }
@ -674,14 +746,16 @@ public class TaskDefinitionServiceImpl extends BaseServiceImpl implements TaskDe
putMsg(result, Status.TASK_DEFINE_NOT_EXIST, String.valueOf(taskCode)); putMsg(result, Status.TASK_DEFINE_NOT_EXIST, String.valueOf(taskCode));
} else { } else {
if (taskDefinition.getVersion() == version) { if (taskDefinition.getVersion() == version) {
logger.warn("Task definition can not be deleted due to version is being used, projectCode:{}, taskDefinitionCode:{}, version:{}.", logger.warn(
"Task definition can not be deleted due to version is being used, projectCode:{}, taskDefinitionCode:{}, version:{}.",
projectCode, taskCode, version); projectCode, taskCode, version);
putMsg(result, Status.MAIN_TABLE_USING_VERSION); putMsg(result, Status.MAIN_TABLE_USING_VERSION);
return result; return result;
} }
int delete = taskDefinitionLogMapper.deleteByCodeAndVersion(taskCode, version); int delete = taskDefinitionLogMapper.deleteByCodeAndVersion(taskCode, version);
if (delete > 0) { if (delete > 0) {
logger.info("Task definition version delete complete, projectCode:{}, taskDefinitionCode:{}, version:{}.", logger.info(
"Task definition version delete complete, projectCode:{}, taskDefinitionCode:{}, version:{}.",
projectCode, taskCode, version); projectCode, taskCode, version);
putMsg(result, Status.SUCCESS); putMsg(result, Status.SUCCESS);
} else { } else {
@ -696,8 +770,9 @@ public class TaskDefinitionServiceImpl extends BaseServiceImpl implements TaskDe
@Override @Override
public Map<String, Object> queryTaskDefinitionDetail(User loginUser, long projectCode, long taskCode) { public Map<String, Object> queryTaskDefinitionDetail(User loginUser, long projectCode, long taskCode) {
Project project = projectMapper.queryByCode(projectCode); Project project = projectMapper.queryByCode(projectCode);
//check user access for project // check user access for project
Map<String, Object> result = projectService.checkProjectAndAuth(loginUser, project, projectCode,TASK_DEFINITION); Map<String, Object> result =
projectService.checkProjectAndAuth(loginUser, project, projectCode, TASK_DEFINITION);
if (result.get(Constants.STATUS) != Status.SUCCESS) { if (result.get(Constants.STATUS) != Status.SUCCESS) {
return result; return result;
} }
@ -724,8 +799,9 @@ public class TaskDefinitionServiceImpl extends BaseServiceImpl implements TaskDe
Integer pageSize) { Integer pageSize) {
Result result = new Result(); Result result = new Result();
Project project = projectMapper.queryByCode(projectCode); Project project = projectMapper.queryByCode(projectCode);
//check user access for project // check user access for project
Map<String, Object> checkResult = projectService.checkProjectAndAuth(loginUser, project, projectCode,TASK_DEFINITION); Map<String, Object> checkResult =
projectService.checkProjectAndAuth(loginUser, project, projectCode, TASK_DEFINITION);
Status resultStatus = (Status) checkResult.get(Constants.STATUS); Status resultStatus = (Status) checkResult.get(Constants.STATUS);
if (resultStatus != Status.SUCCESS) { if (resultStatus != Status.SUCCESS) {
putMsg(result, resultStatus); putMsg(result, resultStatus);
@ -733,7 +809,8 @@ public class TaskDefinitionServiceImpl extends BaseServiceImpl implements TaskDe
} }
taskType = taskType == null ? StringUtils.EMPTY : taskType; taskType = taskType == null ? StringUtils.EMPTY : taskType;
Page<TaskMainInfo> page = new Page<>(pageNo, pageSize); Page<TaskMainInfo> page = new Page<>(pageNo, pageSize);
IPage<TaskMainInfo> taskMainInfoIPage = taskDefinitionMapper.queryDefineListPaging(page, projectCode, searchWorkflowName, IPage<TaskMainInfo> taskMainInfoIPage =
taskDefinitionMapper.queryDefineListPaging(page, projectCode, searchWorkflowName,
searchTaskName, taskType, taskExecuteType); searchTaskName, taskType, taskExecuteType);
List<TaskMainInfo> records = taskMainInfoIPage.getRecords(); List<TaskMainInfo> records = taskMainInfoIPage.getRecords();
if (!records.isEmpty()) { if (!records.isEmpty()) {
@ -800,10 +877,11 @@ public class TaskDefinitionServiceImpl extends BaseServiceImpl implements TaskDe
*/ */
@Transactional @Transactional
@Override @Override
public Map<String, Object> releaseTaskDefinition(User loginUser, long projectCode, long code, ReleaseState releaseState) { public Map<String, Object> releaseTaskDefinition(User loginUser, long projectCode, long code,
ReleaseState releaseState) {
Project project = projectMapper.queryByCode(projectCode); Project project = projectMapper.queryByCode(projectCode);
//check user access for project // check user access for project
Map<String, Object> result = projectService.checkProjectAndAuth(loginUser, project, projectCode,null); Map<String, Object> result = projectService.checkProjectAndAuth(loginUser, project, projectCode, null);
Status resultStatus = (Status) result.get(Constants.STATUS); Status resultStatus = (Status) result.get(Constants.STATUS);
if (resultStatus != Status.SUCCESS) { if (resultStatus != Status.SUCCESS) {
return result; return result;
@ -817,7 +895,8 @@ public class TaskDefinitionServiceImpl extends BaseServiceImpl implements TaskDe
putMsg(result, Status.TASK_DEFINE_NOT_EXIST, String.valueOf(code)); putMsg(result, Status.TASK_DEFINE_NOT_EXIST, String.valueOf(code));
return result; return result;
} }
TaskDefinitionLog taskDefinitionLog = taskDefinitionLogMapper.queryByDefinitionCodeAndVersion(code, taskDefinition.getVersion()); TaskDefinitionLog taskDefinitionLog =
taskDefinitionLogMapper.queryByDefinitionCodeAndVersion(code, taskDefinition.getVersion());
if (taskDefinitionLog == null) { if (taskDefinitionLog == null) {
logger.error("Task definition does not exist, taskDefinitionCode:{}.", code); logger.error("Task definition does not exist, taskDefinitionCode:{}.", code);
putMsg(result, Status.TASK_DEFINE_NOT_EXIST, String.valueOf(code)); putMsg(result, Status.TASK_DEFINE_NOT_EXIST, String.valueOf(code));
@ -831,8 +910,10 @@ public class TaskDefinitionServiceImpl extends BaseServiceImpl implements TaskDe
case ONLINE: case ONLINE:
String resourceIds = taskDefinition.getResourceIds(); String resourceIds = taskDefinition.getResourceIds();
if (StringUtils.isNotBlank(resourceIds)) { if (StringUtils.isNotBlank(resourceIds)) {
Integer[] resourceIdArray = Arrays.stream(resourceIds.split(",")).map(Integer::parseInt).toArray(Integer[]::new); Integer[] resourceIdArray =
PermissionCheck<Integer> permissionCheck = new PermissionCheck(AuthorizationType.RESOURCE_FILE_ID, processService, resourceIdArray, loginUser.getId(), logger); Arrays.stream(resourceIds.split(",")).map(Integer::parseInt).toArray(Integer[]::new);
PermissionCheck<Integer> permissionCheck = new PermissionCheck(AuthorizationType.RESOURCE_FILE_ID,
processService, resourceIdArray, loginUser.getId(), logger);
try { try {
permissionCheck.checkPermission(); permissionCheck.checkPermission();
} catch (Exception e) { } catch (Exception e) {
@ -856,7 +937,8 @@ public class TaskDefinitionServiceImpl extends BaseServiceImpl implements TaskDe
putMsg(result, Status.UPDATE_TASK_DEFINITION_ERROR); putMsg(result, Status.UPDATE_TASK_DEFINITION_ERROR);
throw new ServiceException(Status.UPDATE_TASK_DEFINITION_ERROR); throw new ServiceException(Status.UPDATE_TASK_DEFINITION_ERROR);
} }
logger.error("Update taskDefinition state or taskDefinitionLog state to complete, taskDefinitionCode:{}.", code); logger.error("Update taskDefinition state or taskDefinitionLog state to complete, taskDefinitionCode:{}.",
code);
putMsg(result, Status.SUCCESS); putMsg(result, Status.SUCCESS);
return result; return result;
} }

45
dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessServiceImpl.java
Unescape View File

@ -602,7 +602,8 @@ public class ProcessServiceImpl implements ProcessService {
@Override @Override
public void removeTaskLogFile(Integer processInstanceId) { public void removeTaskLogFile(Integer processInstanceId) {
ProcessInstance processInstance = processInstanceMapper.selectById(processInstanceId); ProcessInstance processInstance = processInstanceMapper.selectById(processInstanceId);
List<TaskInstance> taskInstanceList = findValidTaskListByProcessId(processInstanceId,processInstance.getTestFlag()); List<TaskInstance> taskInstanceList =
findValidTaskListByProcessId(processInstanceId, processInstance.getTestFlag());
if (CollectionUtils.isEmpty(taskInstanceList)) { if (CollectionUtils.isEmpty(taskInstanceList)) {
return; return;
} }
@ -623,7 +624,8 @@ public class ProcessServiceImpl implements ProcessService {
@Override @Override
public void deleteWorkTaskInstanceByProcessInstanceId(int processInstanceId) { public void deleteWorkTaskInstanceByProcessInstanceId(int processInstanceId) {
ProcessInstance processInstance = processInstanceMapper.selectById(processInstanceId); ProcessInstance processInstance = processInstanceMapper.selectById(processInstanceId);
List<TaskInstance> taskInstanceList = findValidTaskListByProcessId(processInstanceId,processInstance.getTestFlag()); List<TaskInstance> taskInstanceList =
findValidTaskListByProcessId(processInstanceId, processInstance.getTestFlag());
if (CollectionUtils.isEmpty(taskInstanceList)) { if (CollectionUtils.isEmpty(taskInstanceList)) {
return; return;
} }
@ -1037,7 +1039,8 @@ public class ProcessServiceImpl implements ProcessService {
case COMPLEMENT_DATA: case COMPLEMENT_DATA:
// delete all the valid tasks when complement data if id is not null // delete all the valid tasks when complement data if id is not null
if (processInstance.getId() != null) { if (processInstance.getId() != null) {
List<TaskInstance> taskInstanceList = this.findValidTaskListByProcessId(processInstance.getId(), processInstance.getTestFlag()); List<TaskInstance> taskInstanceList =
this.findValidTaskListByProcessId(processInstance.getId(), processInstance.getTestFlag());
for (TaskInstance taskInstance : taskInstanceList) { for (TaskInstance taskInstance : taskInstanceList) {
taskInstance.setFlag(Flag.NO); taskInstance.setFlag(Flag.NO);
this.updateTaskInstance(taskInstance); this.updateTaskInstance(taskInstance);
@ -1051,7 +1054,8 @@ public class ProcessServiceImpl implements ProcessService {
processInstance.setCommandParam(JSONUtils.toJsonString(cmdParam)); processInstance.setCommandParam(JSONUtils.toJsonString(cmdParam));
} }
// delete all the valid tasks when repeat running // delete all the valid tasks when repeat running
List<TaskInstance> validTaskList = findValidTaskListByProcessId(processInstance.getId(),processInstance.getTestFlag()); List<TaskInstance> validTaskList =
findValidTaskListByProcessId(processInstance.getId(), processInstance.getTestFlag());
for (TaskInstance taskInstance : validTaskList) { for (TaskInstance taskInstance : validTaskList) {
taskInstance.setFlag(Flag.NO); taskInstance.setFlag(Flag.NO);
updateTaskInstance(taskInstance); updateTaskInstance(taskInstance);
@ -1654,7 +1658,8 @@ public class ProcessServiceImpl implements ProcessService {
if (failureStrategy == FailureStrategy.CONTINUE) { if (failureStrategy == FailureStrategy.CONTINUE) {
return true; return true;
} }
List<TaskInstance> taskInstances = this.findValidTaskListByProcessId(taskInstance.getProcessInstanceId(),taskInstance.getTestFlag()); List<TaskInstance> taskInstances =
this.findValidTaskListByProcessId(taskInstance.getProcessInstanceId(), taskInstance.getTestFlag());
for (TaskInstance task : taskInstances) { for (TaskInstance task : taskInstances) {
if (task.getState() == TaskExecutionStatus.FAILURE if (task.getState() == TaskExecutionStatus.FAILURE
@ -1862,7 +1867,8 @@ public class ProcessServiceImpl implements ProcessService {
@Override @Override
public List<TaskInstance> findPreviousTaskListByWorkProcessId(Integer processInstanceId) { public List<TaskInstance> findPreviousTaskListByWorkProcessId(Integer processInstanceId) {
ProcessInstance processInstance = processInstanceMapper.selectById(processInstanceId); ProcessInstance processInstance = processInstanceMapper.selectById(processInstanceId);
return taskInstanceMapper.findValidTaskListByProcessId(processInstanceId, Flag.NO, processInstance.getTestFlag()); return taskInstanceMapper.findValidTaskListByProcessId(processInstanceId, Flag.NO,
processInstance.getTestFlag());
} }
/** /**
@ -2194,7 +2200,8 @@ public class ProcessServiceImpl implements ProcessService {
* @return process instance * @return process instance
*/ */
@Override @Override
public ProcessInstance findLastSchedulerProcessInterval(Long definitionCode, DateInterval dateInterval, int testFlag) { public ProcessInstance findLastSchedulerProcessInterval(Long definitionCode, DateInterval dateInterval,
int testFlag) {
return processInstanceMapper.queryLastSchedulerProcess(definitionCode, return processInstanceMapper.queryLastSchedulerProcess(definitionCode,
dateInterval.getStartTime(), dateInterval.getStartTime(),
dateInterval.getEndTime(), dateInterval.getEndTime(),
@ -2535,6 +2542,7 @@ public class ProcessServiceImpl implements ProcessService {
taskDefinitionLog.setCreateTime(definitionCodeAndVersion.getCreateTime()); taskDefinitionLog.setCreateTime(definitionCodeAndVersion.getCreateTime());
updateTaskDefinitionLogs.add(taskDefinitionLog); updateTaskDefinitionLogs.add(taskDefinitionLog);
} }
if (CollectionUtils.isNotEmpty(updateTaskDefinitionLogs)) { if (CollectionUtils.isNotEmpty(updateTaskDefinitionLogs)) {
List<Long> taskDefinitionCodes = updateTaskDefinitionLogs List<Long> taskDefinitionCodes = updateTaskDefinitionLogs
.stream() .stream()
@ -2554,15 +2562,24 @@ public class ProcessServiceImpl implements ProcessService {
// for each taskDefinitionLog, we will insert a new version into db // for each taskDefinitionLog, we will insert a new version into db
// and update the origin one if exist // and update the origin one if exist
int updateResult = updateTaskDefinitionLogs.size(); int updateResult = 0;
int insertResult = newTaskDefinitionLogs.size(); int insertResult = 0;
if (CollectionUtils.isNotEmpty(taskDefinitionLogs)) { if (CollectionUtils.isNotEmpty(newTaskDefinitionLogs)) {
insertResult = taskDefinitionLogMapper.batchInsert(taskDefinitionLogs); insertResult += taskDefinitionLogMapper.batchInsert(newTaskDefinitionLogs);
}
if (CollectionUtils.isNotEmpty(updateTaskDefinitionLogs)) {
insertResult += taskDefinitionLogMapper.batchInsert(updateTaskDefinitionLogs);
} }
if (CollectionUtils.isNotEmpty(newTaskDefinitionLogs) && Boolean.TRUE.equals(syncDefine)) { if (CollectionUtils.isNotEmpty(newTaskDefinitionLogs) && Boolean.TRUE.equals(syncDefine)) {
updateResult += taskDefinitionMapper.batchInsert(newTaskDefinitionLogs); updateResult += taskDefinitionMapper.batchInsert(newTaskDefinitionLogs);
} }
if (CollectionUtils.isNotEmpty(updateTaskDefinitionLogs) && Boolean.TRUE.equals(syncDefine)) {
for (TaskDefinitionLog taskDefinitionLog : updateTaskDefinitionLogs) {
updateResult += taskDefinitionMapper.updateById(taskDefinitionLog);
}
}
return (insertResult & updateResult) > 0 ? 1 : Constants.EXIT_CODE_SUCCESS; return (insertResult & updateResult) > 0 ? 1 : Constants.EXIT_CODE_SUCCESS;
} }
@ -3157,7 +3174,8 @@ public class ProcessServiceImpl implements ProcessService {
ProcessInstance processInstance = findProcessInstanceDetailById(task.getProcessInstanceId()).orElse(null); ProcessInstance processInstance = findProcessInstanceDetailById(task.getProcessInstanceId()).orElse(null);
if (processInstance != null if (processInstance != null
&& (processInstance.getState().isFailure() || processInstance.getState().isStop())) { && (processInstance.getState().isFailure() || processInstance.getState().isStop())) {
List<TaskInstance> validTaskList = findValidTaskListByProcessId(processInstance.getId(), processInstance.getTestFlag()); List<TaskInstance> validTaskList =
findValidTaskListByProcessId(processInstance.getId(), processInstance.getTestFlag());
List<Long> instanceTaskCodeList = List<Long> instanceTaskCodeList =
validTaskList.stream().map(TaskInstance::getTaskCode).collect(Collectors.toList()); validTaskList.stream().map(TaskInstance::getTaskCode).collect(Collectors.toList());
List<ProcessTaskRelation> taskRelations = findRelationByCode(processInstance.getProcessDefinitionCode(), List<ProcessTaskRelation> taskRelations = findRelationByCode(processInstance.getProcessDefinitionCode(),
@ -3183,7 +3201,8 @@ public class ProcessServiceImpl implements ProcessService {
@Override @Override
public Integer queryTestDataSourceId(Integer onlineDataSourceId) { public Integer queryTestDataSourceId(Integer onlineDataSourceId) {
Integer testDataSourceId = dataSourceMapper.queryTestDataSourceId(onlineDataSourceId); Integer testDataSourceId = dataSourceMapper.queryTestDataSourceId(onlineDataSourceId);
if (testDataSourceId!=null) return testDataSourceId; if (testDataSourceId != null)
return testDataSourceId;
return null; return null;
} }
} }

Write Preview
Loading…
Cancel
Save