You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@dolphinscheduler.apache.org by GitBox <gi...@apache.org> on 2021/02/01 16:50:14 UTC
[GitHub] [incubator-dolphinscheduler] JinyLeeChina opened a new pull request #4654: [Feature][JsonSplit] add task delete/update/switch method
JinyLeeChina opened a new pull request #4654:
URL: https://github.com/apache/incubator-dolphinscheduler/pull/4654
add task delete/update/switch method #4417
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-dolphinscheduler] lenboo commented on a change in pull request #4654: [Feature][JsonSplit] add task delete/update/switch method
Posted by GitBox <gi...@apache.org>.
lenboo commented on a change in pull request #4654:
URL: https://github.com/apache/incubator-dolphinscheduler/pull/4654#discussion_r568299687
##########
File path: dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskDefinitionServiceImpl.java
##########
@@ -100,14 +111,8 @@
}
TaskNode taskNode = JSONUtils.parseObject(taskDefinitionJson, TaskNode.class);
- if (taskNode == null) {
- logger.error("taskDefinitionJson is not valid json");
- putMsg(result, Status.DATA_IS_NOT_VALID, taskDefinitionJson);
- return result;
- }
- if (!CheckUtils.checkTaskNodeParameters(taskNode.getParams(), taskNode.getName())) {
- logger.error("task node {} parameter invalid", taskNode.getName());
- putMsg(result, Status.PROCESS_NODE_S_PARAMETER_INVALID, taskNode.getName());
+ checkTaskNode(result, taskNode, taskDefinitionJson);
Review comment:
missing 'result = ？'
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-dolphinscheduler] JinyLeeChina commented on pull request #4654: [Feature][JsonSplit] add task delete/update/switch method
Posted by GitBox <gi...@apache.org>.
JinyLeeChina commented on pull request #4654:
URL: https://github.com/apache/incubator-dolphinscheduler/pull/4654#issuecomment-771375917
> result is a map, which is passed into the method through the parameter
> result is a map, which is passed into the method through the parameter
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-dolphinscheduler] simon824 commented on a change in pull request #4654: [Feature][JsonSplit] add task delete/update/switch method
Posted by GitBox <gi...@apache.org>.
simon824 commented on a change in pull request #4654:
URL: https://github.com/apache/incubator-dolphinscheduler/pull/4654#discussion_r568452836
##########
File path: dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskDefinitionServiceImpl.java
##########
@@ -204,5 +211,188 @@ private String getResourceIds(TaskDefinition taskDefinition) {
return result;
}
+ /**
+ * delete task definition
+ *
+ * @param loginUser login user
+ * @param projectName project name
+ * @param taskCode task code
+ */
+ @Transactional(rollbackFor = RuntimeException.class)
+ @Override
+ public Map<String, Object> deleteTaskDefinitionByCode(User loginUser, String projectName, Long taskCode) {
+ Map<String, Object> result = new HashMap<>(5);
+ Project project = projectMapper.queryByName(projectName);
+
+ Map<String, Object> checkResult = projectService.checkProjectAndAuth(loginUser, project, projectName);
+ Status resultEnum = (Status) checkResult.get(Constants.STATUS);
+ if (resultEnum != Status.SUCCESS) {
+ return checkResult;
+ }
+ checkTaskRelation(result, taskCode);
+ resultEnum = (Status) result.get(Constants.STATUS);
+ if (resultEnum != Status.SUCCESS) {
+ return result;
+ }
+ int delete = taskDefinitionMapper.deleteByCode(taskCode);
+ if (delete > 0) {
+ putMsg(result, Status.SUCCESS);
+ } else {
+ putMsg(result, Status.DELETE_PROCESS_DEFINE_BY_ID_ERROR);
+ }
+ return result;
+ }
+
+ /**
+ * update task definition
+ *
+ * @param loginUser login user
+ * @param projectName project name
+ * @param taskCode task code
+ * @param taskDefinitionJson task definition json
+ */
+ @Override
+ public Map<String, Object> updateTaskDefinition(User loginUser, String projectName, Long taskCode, String taskDefinitionJson) {
+ Map<String, Object> result = new HashMap<>(5);
+ Project project = projectMapper.queryByName(projectName);
+
+ Map<String, Object> checkResult = projectService.checkProjectAndAuth(loginUser, project, projectName);
+ Status resultEnum = (Status) checkResult.get(Constants.STATUS);
+ if (resultEnum != Status.SUCCESS) {
+ return checkResult;
+ }
+ checkTaskRelation(result, taskCode);
+ if (result.get(Constants.STATUS) != Status.SUCCESS) {
+ return result;
+ }
+ TaskDefinition taskDefinition = taskDefinitionMapper.queryByDefinitionCode(taskCode);
+ if (taskDefinition == null) {
+ putMsg(result, Status.TASK_DEFINE_NOT_EXIST, taskCode);
+ return result;
+ }
+ TaskNode taskNode = JSONUtils.parseObject(taskDefinitionJson, TaskNode.class);
+ checkTaskNode(result, taskNode, taskDefinitionJson);
+ if (result.get(Constants.STATUS) != Status.SUCCESS) {
+ return result;
+ }
+
+ List<TaskDefinitionLog> taskDefinitionLogs = taskDefinitionLogMapper.queryByDefinitionCode(taskCode);
+ int version = taskDefinitionLogs
+ .stream()
+ .map(TaskDefinitionLog::getVersion)
+ .max((x, y) -> x > y ? x : y)
+ .orElse(0) + 1;
+ Date now = new Date();
+ taskDefinition.setVersion(version);
+ taskDefinition.setCode(taskCode);
+ taskDefinition.setName(taskNode.getName());
+ taskDefinition.setDescription(taskNode.getDesc());
+ taskDefinition.setProjectCode(project.getCode());
+ taskDefinition.setUserId(loginUser.getId());
+ taskDefinition.setTaskType(TaskType.of(taskNode.getType()));
+ taskDefinition.setTaskParams(taskNode.getParams());
+ taskDefinition.setFlag(taskNode.isForbidden() ? Flag.NO : Flag.YES);
+ taskDefinition.setTaskPriority(taskNode.getTaskInstancePriority());
+ taskDefinition.setWorkerGroup(taskNode.getWorkerGroup());
+ taskDefinition.setFailRetryTimes(taskNode.getMaxRetryTimes());
+ taskDefinition.setFailRetryInterval(taskNode.getRetryInterval());
+ taskDefinition.setTimeoutFlag(taskNode.getTaskTimeoutParameter().getEnable() ? TimeoutFlag.OPEN : TimeoutFlag.CLOSE);
+ taskDefinition.setTaskTimeoutStrategy(taskNode.getTaskTimeoutParameter().getStrategy());
+ taskDefinition.setTimeout(taskNode.getTaskTimeoutParameter().getInterval());
+ taskDefinition.setUpdateTime(now);
+ taskDefinition.setResourceIds(getResourceIds(taskDefinition));
+ taskDefinitionMapper.updateById(taskDefinition);
+ // save task definition log
+ TaskDefinitionLog taskDefinitionLog = new TaskDefinitionLog();
+ taskDefinitionLog.set(taskDefinition);
+ taskDefinitionLog.setOperator(loginUser.getId());
+ taskDefinitionLog.setOperateTime(now);
+ taskDefinitionLogMapper.insert(taskDefinitionLog);
+ result.put(Constants.DATA_LIST, taskCode);
+ putMsg(result, Status.SUCCESS);
+ return result;
+ }
+
+ public void checkTaskRelation(Map<String, Object> result, Long taskCode) {
+ List<ProcessTaskRelation> processTaskRelationList = processTaskRelationMapper.queryByTaskCode(taskCode, taskCode);
+ if (!processTaskRelationList.isEmpty()) {
+ Set<Long> processDefinitionCodes = processTaskRelationList
+ .stream()
+ .map(ProcessTaskRelation::getProcessDefinitionCode)
+ .collect(Collectors.toSet());
+ List<ProcessDefinition> processDefinitionList = processDefinitionMapper.queryByCodes(processDefinitionCodes);
+ // check process definition is already online
+ for (ProcessDefinition processDefinition : processDefinitionList) {
+ if (processDefinition.getReleaseState() == ReleaseState.ONLINE) {
+ putMsg(result, Status.PROCESS_DEFINE_STATE_ONLINE, processDefinition.getCode());
+ return;
+ }
+ }
+ }
+ }
+
+ public void checkTaskNode(Map<String, Object> result, TaskNode taskNode, String taskDefinitionJson) {
+ if (taskNode == null) {
+ logger.error("taskDefinitionJson is not valid json");
+ putMsg(result, Status.DATA_IS_NOT_VALID, taskDefinitionJson);
+ return;
+ }
+ if (!CheckUtils.checkTaskNodeParameters(taskNode.getParams(), taskNode.getName())) {
+ logger.error("task node {} parameter invalid", taskNode.getName());
+ putMsg(result, Status.PROCESS_NODE_S_PARAMETER_INVALID, taskNode.getName());
+ }
+ }
+
+ /**
+ * update task definition
+ *
+ * @param loginUser login user
+ * @param projectName project name
+ * @param taskCode task code
+ * @param version the version user want to switch
+ */
+ @Override
+ public Map<String, Object> switchVersion(User loginUser, String projectName, Long taskCode, int version) {
+ Map<String, Object> result = new HashMap<>(5);
+ Project project = projectMapper.queryByName(projectName);
+
+ Map<String, Object> checkResult = projectService.checkProjectAndAuth(loginUser, project, projectName);
+ Status resultEnum = (Status) checkResult.get(Constants.STATUS);
+ if (resultEnum != Status.SUCCESS) {
+ return checkResult;
+ }
+ checkTaskRelation(result, taskCode);
+ if (result.get(Constants.STATUS) != Status.SUCCESS) {
+ return result;
+ }
+ TaskDefinition taskDefinition = taskDefinitionMapper.queryByDefinitionCode(taskCode);
+ if (taskDefinition == null) {
+ putMsg(result, Status.TASK_DEFINE_NOT_EXIST, taskCode);
+ return result;
+ }
+ TaskDefinitionLog taskDefinitionLog = taskDefinitionLogMapper.queryByDefinitionCodeAndVersion(taskCode, version);
+ taskDefinition.setVersion(version);
+ taskDefinition.setCode(taskCode);
+ taskDefinition.setName(taskDefinitionLog.getName());
+ taskDefinition.setDescription(taskDefinitionLog.getDescription());
+ taskDefinition.setProjectCode(taskDefinitionLog.getProjectCode());
+ taskDefinition.setUserId(loginUser.getId());
+ taskDefinition.setTaskType(taskDefinitionLog.getTaskType());
+ taskDefinition.setTaskParams(taskDefinitionLog.getTaskParams());
+ taskDefinition.setFlag(taskDefinitionLog.getFlag());
+ taskDefinition.setTaskPriority(taskDefinitionLog.getTaskPriority());
+ taskDefinition.setWorkerGroup(taskDefinitionLog.getWorkerGroup());
+ taskDefinition.setFailRetryTimes(taskDefinitionLog.getFailRetryTimes());
+ taskDefinition.setFailRetryInterval(taskDefinitionLog.getFailRetryInterval());
+ taskDefinition.setTimeoutFlag(taskDefinitionLog.getTimeoutFlag());
+ taskDefinition.setTaskTimeoutStrategy(taskDefinitionLog.getTaskTimeoutStrategy());
+ taskDefinition.setTimeout(taskDefinitionLog.getTimeout());
+ taskDefinition.setUpdateTime(new Date());
+ taskDefinition.setResourceIds(taskDefinitionLog.getResourceIds());
+ taskDefinitionMapper.updateById(taskDefinition);
Review comment:
missing setId or updateByCode here?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-dolphinscheduler] JinyLeeChina closed pull request #4654: [Feature][JsonSplit] add task delete/update/switch method
Posted by GitBox <gi...@apache.org>.
JinyLeeChina closed pull request #4654:
URL: https://github.com/apache/incubator-dolphinscheduler/pull/4654
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-dolphinscheduler] JinyLeeChina closed pull request #4654: [Feature][JsonSplit] add task delete/update/switch method
Posted by GitBox <gi...@apache.org>.
JinyLeeChina closed pull request #4654:
URL: https://github.com/apache/incubator-dolphinscheduler/pull/4654
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-dolphinscheduler] JinyLeeChina commented on a change in pull request #4654: [Feature][JsonSplit] add task delete/update/switch method
Posted by GitBox <gi...@apache.org>.
JinyLeeChina commented on a change in pull request #4654:
URL: https://github.com/apache/incubator-dolphinscheduler/pull/4654#discussion_r568484233
##########
File path: dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/ProcessTaskRelationService.java
##########
@@ -48,5 +48,77 @@
String conditionType,
String conditionParams);
+
+ /**
+ * query process task relation
+ *
+ * @param loginUser login user
+ * @param projectName project name
+ * @param processDefinitionCode process definition code
+ */
+ Map<String, Object> queryProcessTaskRelation(User loginUser,
+ String projectName,
+ Long processDefinitionCode);
+
+ /**
+ * delete process task relation
+ *
+ * @param loginUser login user
+ * @param projectName project name
+ * @param processDefinitionCode process definition code
+ */
+ Map<String, Object> deleteTaskDefinitionByProcess(User loginUser,
+ String projectName,
+ Long processDefinitionCode);
+
Review comment:
OK
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-dolphinscheduler] JinyLeeChina removed a comment on pull request #4654: [Feature][JsonSplit] add task delete/update/switch method
Posted by GitBox <gi...@apache.org>.
JinyLeeChina removed a comment on pull request #4654:
URL: https://github.com/apache/incubator-dolphinscheduler/pull/4654#issuecomment-771375917
> result is a map, which is passed into the method through the parameter
> result is a map, which is passed into the method through the parameter
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-dolphinscheduler] simon824 commented on a change in pull request #4654: [Feature][JsonSplit] add task delete/update/switch method
Posted by GitBox <gi...@apache.org>.
simon824 commented on a change in pull request #4654:
URL: https://github.com/apache/incubator-dolphinscheduler/pull/4654#discussion_r568452836
##########
File path: dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskDefinitionServiceImpl.java
##########
@@ -204,5 +211,188 @@ private String getResourceIds(TaskDefinition taskDefinition) {
return result;
}
+ /**
+ * delete task definition
+ *
+ * @param loginUser login user
+ * @param projectName project name
+ * @param taskCode task code
+ */
+ @Transactional(rollbackFor = RuntimeException.class)
+ @Override
+ public Map<String, Object> deleteTaskDefinitionByCode(User loginUser, String projectName, Long taskCode) {
+ Map<String, Object> result = new HashMap<>(5);
+ Project project = projectMapper.queryByName(projectName);
+
+ Map<String, Object> checkResult = projectService.checkProjectAndAuth(loginUser, project, projectName);
+ Status resultEnum = (Status) checkResult.get(Constants.STATUS);
+ if (resultEnum != Status.SUCCESS) {
+ return checkResult;
+ }
+ checkTaskRelation(result, taskCode);
+ resultEnum = (Status) result.get(Constants.STATUS);
+ if (resultEnum != Status.SUCCESS) {
+ return result;
+ }
+ int delete = taskDefinitionMapper.deleteByCode(taskCode);
+ if (delete > 0) {
+ putMsg(result, Status.SUCCESS);
+ } else {
+ putMsg(result, Status.DELETE_PROCESS_DEFINE_BY_ID_ERROR);
+ }
+ return result;
+ }
+
+ /**
+ * update task definition
+ *
+ * @param loginUser login user
+ * @param projectName project name
+ * @param taskCode task code
+ * @param taskDefinitionJson task definition json
+ */
+ @Override
+ public Map<String, Object> updateTaskDefinition(User loginUser, String projectName, Long taskCode, String taskDefinitionJson) {
+ Map<String, Object> result = new HashMap<>(5);
+ Project project = projectMapper.queryByName(projectName);
+
+ Map<String, Object> checkResult = projectService.checkProjectAndAuth(loginUser, project, projectName);
+ Status resultEnum = (Status) checkResult.get(Constants.STATUS);
+ if (resultEnum != Status.SUCCESS) {
+ return checkResult;
+ }
+ checkTaskRelation(result, taskCode);
+ if (result.get(Constants.STATUS) != Status.SUCCESS) {
+ return result;
+ }
+ TaskDefinition taskDefinition = taskDefinitionMapper.queryByDefinitionCode(taskCode);
+ if (taskDefinition == null) {
+ putMsg(result, Status.TASK_DEFINE_NOT_EXIST, taskCode);
+ return result;
+ }
+ TaskNode taskNode = JSONUtils.parseObject(taskDefinitionJson, TaskNode.class);
+ checkTaskNode(result, taskNode, taskDefinitionJson);
+ if (result.get(Constants.STATUS) != Status.SUCCESS) {
+ return result;
+ }
+
+ List<TaskDefinitionLog> taskDefinitionLogs = taskDefinitionLogMapper.queryByDefinitionCode(taskCode);
+ int version = taskDefinitionLogs
+ .stream()
+ .map(TaskDefinitionLog::getVersion)
+ .max((x, y) -> x > y ? x : y)
+ .orElse(0) + 1;
+ Date now = new Date();
+ taskDefinition.setVersion(version);
+ taskDefinition.setCode(taskCode);
+ taskDefinition.setName(taskNode.getName());
+ taskDefinition.setDescription(taskNode.getDesc());
+ taskDefinition.setProjectCode(project.getCode());
+ taskDefinition.setUserId(loginUser.getId());
+ taskDefinition.setTaskType(TaskType.of(taskNode.getType()));
+ taskDefinition.setTaskParams(taskNode.getParams());
+ taskDefinition.setFlag(taskNode.isForbidden() ? Flag.NO : Flag.YES);
+ taskDefinition.setTaskPriority(taskNode.getTaskInstancePriority());
+ taskDefinition.setWorkerGroup(taskNode.getWorkerGroup());
+ taskDefinition.setFailRetryTimes(taskNode.getMaxRetryTimes());
+ taskDefinition.setFailRetryInterval(taskNode.getRetryInterval());
+ taskDefinition.setTimeoutFlag(taskNode.getTaskTimeoutParameter().getEnable() ? TimeoutFlag.OPEN : TimeoutFlag.CLOSE);
+ taskDefinition.setTaskTimeoutStrategy(taskNode.getTaskTimeoutParameter().getStrategy());
+ taskDefinition.setTimeout(taskNode.getTaskTimeoutParameter().getInterval());
+ taskDefinition.setUpdateTime(now);
+ taskDefinition.setResourceIds(getResourceIds(taskDefinition));
+ taskDefinitionMapper.updateById(taskDefinition);
+ // save task definition log
+ TaskDefinitionLog taskDefinitionLog = new TaskDefinitionLog();
+ taskDefinitionLog.set(taskDefinition);
+ taskDefinitionLog.setOperator(loginUser.getId());
+ taskDefinitionLog.setOperateTime(now);
+ taskDefinitionLogMapper.insert(taskDefinitionLog);
+ result.put(Constants.DATA_LIST, taskCode);
+ putMsg(result, Status.SUCCESS);
+ return result;
+ }
+
+ public void checkTaskRelation(Map<String, Object> result, Long taskCode) {
+ List<ProcessTaskRelation> processTaskRelationList = processTaskRelationMapper.queryByTaskCode(taskCode, taskCode);
+ if (!processTaskRelationList.isEmpty()) {
+ Set<Long> processDefinitionCodes = processTaskRelationList
+ .stream()
+ .map(ProcessTaskRelation::getProcessDefinitionCode)
+ .collect(Collectors.toSet());
+ List<ProcessDefinition> processDefinitionList = processDefinitionMapper.queryByCodes(processDefinitionCodes);
+ // check process definition is already online
+ for (ProcessDefinition processDefinition : processDefinitionList) {
+ if (processDefinition.getReleaseState() == ReleaseState.ONLINE) {
+ putMsg(result, Status.PROCESS_DEFINE_STATE_ONLINE, processDefinition.getCode());
+ return;
+ }
+ }
+ }
+ }
+
+ public void checkTaskNode(Map<String, Object> result, TaskNode taskNode, String taskDefinitionJson) {
+ if (taskNode == null) {
+ logger.error("taskDefinitionJson is not valid json");
+ putMsg(result, Status.DATA_IS_NOT_VALID, taskDefinitionJson);
+ return;
+ }
+ if (!CheckUtils.checkTaskNodeParameters(taskNode.getParams(), taskNode.getName())) {
+ logger.error("task node {} parameter invalid", taskNode.getName());
+ putMsg(result, Status.PROCESS_NODE_S_PARAMETER_INVALID, taskNode.getName());
+ }
+ }
+
+ /**
+ * update task definition
+ *
+ * @param loginUser login user
+ * @param projectName project name
+ * @param taskCode task code
+ * @param version the version user want to switch
+ */
+ @Override
+ public Map<String, Object> switchVersion(User loginUser, String projectName, Long taskCode, int version) {
+ Map<String, Object> result = new HashMap<>(5);
+ Project project = projectMapper.queryByName(projectName);
+
+ Map<String, Object> checkResult = projectService.checkProjectAndAuth(loginUser, project, projectName);
+ Status resultEnum = (Status) checkResult.get(Constants.STATUS);
+ if (resultEnum != Status.SUCCESS) {
+ return checkResult;
+ }
+ checkTaskRelation(result, taskCode);
+ if (result.get(Constants.STATUS) != Status.SUCCESS) {
+ return result;
+ }
+ TaskDefinition taskDefinition = taskDefinitionMapper.queryByDefinitionCode(taskCode);
+ if (taskDefinition == null) {
+ putMsg(result, Status.TASK_DEFINE_NOT_EXIST, taskCode);
+ return result;
+ }
+ TaskDefinitionLog taskDefinitionLog = taskDefinitionLogMapper.queryByDefinitionCodeAndVersion(taskCode, version);
+ taskDefinition.setVersion(version);
+ taskDefinition.setCode(taskCode);
+ taskDefinition.setName(taskDefinitionLog.getName());
+ taskDefinition.setDescription(taskDefinitionLog.getDescription());
+ taskDefinition.setProjectCode(taskDefinitionLog.getProjectCode());
+ taskDefinition.setUserId(loginUser.getId());
+ taskDefinition.setTaskType(taskDefinitionLog.getTaskType());
+ taskDefinition.setTaskParams(taskDefinitionLog.getTaskParams());
+ taskDefinition.setFlag(taskDefinitionLog.getFlag());
+ taskDefinition.setTaskPriority(taskDefinitionLog.getTaskPriority());
+ taskDefinition.setWorkerGroup(taskDefinitionLog.getWorkerGroup());
+ taskDefinition.setFailRetryTimes(taskDefinitionLog.getFailRetryTimes());
+ taskDefinition.setFailRetryInterval(taskDefinitionLog.getFailRetryInterval());
+ taskDefinition.setTimeoutFlag(taskDefinitionLog.getTimeoutFlag());
+ taskDefinition.setTaskTimeoutStrategy(taskDefinitionLog.getTaskTimeoutStrategy());
+ taskDefinition.setTimeout(taskDefinitionLog.getTimeout());
+ taskDefinition.setUpdateTime(new Date());
+ taskDefinition.setResourceIds(taskDefinitionLog.getResourceIds());
+ taskDefinitionMapper.updateById(taskDefinition);
Review comment:
missing setId or updateByCode here?
##########
File path: dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/ProcessTaskRelationService.java
##########
@@ -48,5 +48,77 @@
String conditionType,
String conditionParams);
+
+ /**
+ * query process task relation
+ *
+ * @param loginUser login user
+ * @param projectName project name
+ * @param processDefinitionCode process definition code
+ */
+ Map<String, Object> queryProcessTaskRelation(User loginUser,
+ String projectName,
+ Long processDefinitionCode);
+
+ /**
+ * delete process task relation
+ *
+ * @param loginUser login user
+ * @param projectName project name
+ * @param processDefinitionCode process definition code
+ */
+ Map<String, Object> deleteTaskDefinitionByProcess(User loginUser,
+ String projectName,
+ Long processDefinitionCode);
+
Review comment:
The function name deleteTaskDefinitionByProcessCode maybe more easy to understand
##########
File path: dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskDefinitionServiceImpl.java
##########
@@ -204,5 +211,188 @@ private String getResourceIds(TaskDefinition taskDefinition) {
return result;
}
+ /**
+ * delete task definition
+ *
+ * @param loginUser login user
+ * @param projectName project name
+ * @param taskCode task code
+ */
+ @Transactional(rollbackFor = RuntimeException.class)
+ @Override
+ public Map<String, Object> deleteTaskDefinitionByCode(User loginUser, String projectName, Long taskCode) {
+ Map<String, Object> result = new HashMap<>(5);
+ Project project = projectMapper.queryByName(projectName);
+
+ Map<String, Object> checkResult = projectService.checkProjectAndAuth(loginUser, project, projectName);
+ Status resultEnum = (Status) checkResult.get(Constants.STATUS);
+ if (resultEnum != Status.SUCCESS) {
+ return checkResult;
+ }
+ checkTaskRelation(result, taskCode);
+ resultEnum = (Status) result.get(Constants.STATUS);
+ if (resultEnum != Status.SUCCESS) {
+ return result;
+ }
+ int delete = taskDefinitionMapper.deleteByCode(taskCode);
+ if (delete > 0) {
+ putMsg(result, Status.SUCCESS);
+ } else {
+ putMsg(result, Status.DELETE_PROCESS_DEFINE_BY_ID_ERROR);
+ }
+ return result;
+ }
+
+ /**
+ * update task definition
+ *
+ * @param loginUser login user
+ * @param projectName project name
+ * @param taskCode task code
+ * @param taskDefinitionJson task definition json
+ */
+ @Override
+ public Map<String, Object> updateTaskDefinition(User loginUser, String projectName, Long taskCode, String taskDefinitionJson) {
+ Map<String, Object> result = new HashMap<>(5);
+ Project project = projectMapper.queryByName(projectName);
+
+ Map<String, Object> checkResult = projectService.checkProjectAndAuth(loginUser, project, projectName);
+ Status resultEnum = (Status) checkResult.get(Constants.STATUS);
+ if (resultEnum != Status.SUCCESS) {
+ return checkResult;
+ }
+ checkTaskRelation(result, taskCode);
+ if (result.get(Constants.STATUS) != Status.SUCCESS) {
+ return result;
+ }
+ TaskDefinition taskDefinition = taskDefinitionMapper.queryByDefinitionCode(taskCode);
+ if (taskDefinition == null) {
+ putMsg(result, Status.TASK_DEFINE_NOT_EXIST, taskCode);
+ return result;
+ }
+ TaskNode taskNode = JSONUtils.parseObject(taskDefinitionJson, TaskNode.class);
+ checkTaskNode(result, taskNode, taskDefinitionJson);
+ if (result.get(Constants.STATUS) != Status.SUCCESS) {
+ return result;
+ }
+
+ List<TaskDefinitionLog> taskDefinitionLogs = taskDefinitionLogMapper.queryByDefinitionCode(taskCode);
+ int version = taskDefinitionLogs
+ .stream()
+ .map(TaskDefinitionLog::getVersion)
+ .max((x, y) -> x > y ? x : y)
+ .orElse(0) + 1;
+ Date now = new Date();
+ taskDefinition.setVersion(version);
+ taskDefinition.setCode(taskCode);
+ taskDefinition.setName(taskNode.getName());
+ taskDefinition.setDescription(taskNode.getDesc());
+ taskDefinition.setProjectCode(project.getCode());
+ taskDefinition.setUserId(loginUser.getId());
+ taskDefinition.setTaskType(TaskType.of(taskNode.getType()));
+ taskDefinition.setTaskParams(taskNode.getParams());
+ taskDefinition.setFlag(taskNode.isForbidden() ? Flag.NO : Flag.YES);
+ taskDefinition.setTaskPriority(taskNode.getTaskInstancePriority());
+ taskDefinition.setWorkerGroup(taskNode.getWorkerGroup());
+ taskDefinition.setFailRetryTimes(taskNode.getMaxRetryTimes());
+ taskDefinition.setFailRetryInterval(taskNode.getRetryInterval());
+ taskDefinition.setTimeoutFlag(taskNode.getTaskTimeoutParameter().getEnable() ? TimeoutFlag.OPEN : TimeoutFlag.CLOSE);
+ taskDefinition.setTaskTimeoutStrategy(taskNode.getTaskTimeoutParameter().getStrategy());
+ taskDefinition.setTimeout(taskNode.getTaskTimeoutParameter().getInterval());
+ taskDefinition.setUpdateTime(now);
+ taskDefinition.setResourceIds(getResourceIds(taskDefinition));
+ taskDefinitionMapper.updateById(taskDefinition);
Review comment:
missing setId or updateByCode here?
##########
File path: dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskDefinitionServiceImpl.java
##########
@@ -204,5 +211,188 @@ private String getResourceIds(TaskDefinition taskDefinition) {
return result;
}
+ /**
+ * delete task definition
+ *
+ * @param loginUser login user
+ * @param projectName project name
+ * @param taskCode task code
+ */
+ @Transactional(rollbackFor = RuntimeException.class)
+ @Override
+ public Map<String, Object> deleteTaskDefinitionByCode(User loginUser, String projectName, Long taskCode) {
+ Map<String, Object> result = new HashMap<>(5);
+ Project project = projectMapper.queryByName(projectName);
+
+ Map<String, Object> checkResult = projectService.checkProjectAndAuth(loginUser, project, projectName);
+ Status resultEnum = (Status) checkResult.get(Constants.STATUS);
+ if (resultEnum != Status.SUCCESS) {
+ return checkResult;
+ }
+ checkTaskRelation(result, taskCode);
+ resultEnum = (Status) result.get(Constants.STATUS);
+ if (resultEnum != Status.SUCCESS) {
+ return result;
+ }
+ int delete = taskDefinitionMapper.deleteByCode(taskCode);
+ if (delete > 0) {
+ putMsg(result, Status.SUCCESS);
+ } else {
+ putMsg(result, Status.DELETE_PROCESS_DEFINE_BY_ID_ERROR);
+ }
+ return result;
+ }
+
+ /**
+ * update task definition
+ *
+ * @param loginUser login user
+ * @param projectName project name
+ * @param taskCode task code
+ * @param taskDefinitionJson task definition json
+ */
+ @Override
+ public Map<String, Object> updateTaskDefinition(User loginUser, String projectName, Long taskCode, String taskDefinitionJson) {
+ Map<String, Object> result = new HashMap<>(5);
+ Project project = projectMapper.queryByName(projectName);
+
+ Map<String, Object> checkResult = projectService.checkProjectAndAuth(loginUser, project, projectName);
+ Status resultEnum = (Status) checkResult.get(Constants.STATUS);
+ if (resultEnum != Status.SUCCESS) {
+ return checkResult;
+ }
+ checkTaskRelation(result, taskCode);
+ if (result.get(Constants.STATUS) != Status.SUCCESS) {
+ return result;
+ }
+ TaskDefinition taskDefinition = taskDefinitionMapper.queryByDefinitionCode(taskCode);
+ if (taskDefinition == null) {
+ putMsg(result, Status.TASK_DEFINE_NOT_EXIST, taskCode);
+ return result;
+ }
+ TaskNode taskNode = JSONUtils.parseObject(taskDefinitionJson, TaskNode.class);
+ checkTaskNode(result, taskNode, taskDefinitionJson);
+ if (result.get(Constants.STATUS) != Status.SUCCESS) {
+ return result;
+ }
+
+ List<TaskDefinitionLog> taskDefinitionLogs = taskDefinitionLogMapper.queryByDefinitionCode(taskCode);
+ int version = taskDefinitionLogs
+ .stream()
+ .map(TaskDefinitionLog::getVersion)
+ .max((x, y) -> x > y ? x : y)
+ .orElse(0) + 1;
+ Date now = new Date();
+ taskDefinition.setVersion(version);
+ taskDefinition.setCode(taskCode);
+ taskDefinition.setName(taskNode.getName());
+ taskDefinition.setDescription(taskNode.getDesc());
+ taskDefinition.setProjectCode(project.getCode());
+ taskDefinition.setUserId(loginUser.getId());
+ taskDefinition.setTaskType(TaskType.of(taskNode.getType()));
+ taskDefinition.setTaskParams(taskNode.getParams());
+ taskDefinition.setFlag(taskNode.isForbidden() ? Flag.NO : Flag.YES);
+ taskDefinition.setTaskPriority(taskNode.getTaskInstancePriority());
+ taskDefinition.setWorkerGroup(taskNode.getWorkerGroup());
+ taskDefinition.setFailRetryTimes(taskNode.getMaxRetryTimes());
+ taskDefinition.setFailRetryInterval(taskNode.getRetryInterval());
+ taskDefinition.setTimeoutFlag(taskNode.getTaskTimeoutParameter().getEnable() ? TimeoutFlag.OPEN : TimeoutFlag.CLOSE);
+ taskDefinition.setTaskTimeoutStrategy(taskNode.getTaskTimeoutParameter().getStrategy());
+ taskDefinition.setTimeout(taskNode.getTaskTimeoutParameter().getInterval());
+ taskDefinition.setUpdateTime(now);
+ taskDefinition.setResourceIds(getResourceIds(taskDefinition));
+ taskDefinitionMapper.updateById(taskDefinition);
+ // save task definition log
+ TaskDefinitionLog taskDefinitionLog = new TaskDefinitionLog();
+ taskDefinitionLog.set(taskDefinition);
+ taskDefinitionLog.setOperator(loginUser.getId());
+ taskDefinitionLog.setOperateTime(now);
+ taskDefinitionLogMapper.insert(taskDefinitionLog);
+ result.put(Constants.DATA_LIST, taskCode);
+ putMsg(result, Status.SUCCESS);
+ return result;
+ }
+
+ public void checkTaskRelation(Map<String, Object> result, Long taskCode) {
+ List<ProcessTaskRelation> processTaskRelationList = processTaskRelationMapper.queryByTaskCode(taskCode, taskCode);
+ if (!processTaskRelationList.isEmpty()) {
+ Set<Long> processDefinitionCodes = processTaskRelationList
+ .stream()
+ .map(ProcessTaskRelation::getProcessDefinitionCode)
+ .collect(Collectors.toSet());
+ List<ProcessDefinition> processDefinitionList = processDefinitionMapper.queryByCodes(processDefinitionCodes);
+ // check process definition is already online
+ for (ProcessDefinition processDefinition : processDefinitionList) {
+ if (processDefinition.getReleaseState() == ReleaseState.ONLINE) {
+ putMsg(result, Status.PROCESS_DEFINE_STATE_ONLINE, processDefinition.getCode());
+ return;
+ }
+ }
+ }
+ }
+
+ public void checkTaskNode(Map<String, Object> result, TaskNode taskNode, String taskDefinitionJson) {
+ if (taskNode == null) {
+ logger.error("taskDefinitionJson is not valid json");
+ putMsg(result, Status.DATA_IS_NOT_VALID, taskDefinitionJson);
+ return;
+ }
+ if (!CheckUtils.checkTaskNodeParameters(taskNode.getParams(), taskNode.getName())) {
+ logger.error("task node {} parameter invalid", taskNode.getName());
+ putMsg(result, Status.PROCESS_NODE_S_PARAMETER_INVALID, taskNode.getName());
+ }
+ }
+
+ /**
+ * update task definition
+ *
+ * @param loginUser login user
+ * @param projectName project name
+ * @param taskCode task code
+ * @param version the version user want to switch
+ */
+ @Override
+ public Map<String, Object> switchVersion(User loginUser, String projectName, Long taskCode, int version) {
+ Map<String, Object> result = new HashMap<>(5);
+ Project project = projectMapper.queryByName(projectName);
+
+ Map<String, Object> checkResult = projectService.checkProjectAndAuth(loginUser, project, projectName);
+ Status resultEnum = (Status) checkResult.get(Constants.STATUS);
+ if (resultEnum != Status.SUCCESS) {
+ return checkResult;
+ }
+ checkTaskRelation(result, taskCode);
+ if (result.get(Constants.STATUS) != Status.SUCCESS) {
+ return result;
+ }
+ TaskDefinition taskDefinition = taskDefinitionMapper.queryByDefinitionCode(taskCode);
+ if (taskDefinition == null) {
+ putMsg(result, Status.TASK_DEFINE_NOT_EXIST, taskCode);
+ return result;
+ }
+ TaskDefinitionLog taskDefinitionLog = taskDefinitionLogMapper.queryByDefinitionCodeAndVersion(taskCode, version);
+ taskDefinition.setVersion(version);
+ taskDefinition.setCode(taskCode);
+ taskDefinition.setName(taskDefinitionLog.getName());
+ taskDefinition.setDescription(taskDefinitionLog.getDescription());
+ taskDefinition.setProjectCode(taskDefinitionLog.getProjectCode());
+ taskDefinition.setUserId(loginUser.getId());
+ taskDefinition.setTaskType(taskDefinitionLog.getTaskType());
+ taskDefinition.setTaskParams(taskDefinitionLog.getTaskParams());
+ taskDefinition.setFlag(taskDefinitionLog.getFlag());
+ taskDefinition.setTaskPriority(taskDefinitionLog.getTaskPriority());
+ taskDefinition.setWorkerGroup(taskDefinitionLog.getWorkerGroup());
+ taskDefinition.setFailRetryTimes(taskDefinitionLog.getFailRetryTimes());
+ taskDefinition.setFailRetryInterval(taskDefinitionLog.getFailRetryInterval());
+ taskDefinition.setTimeoutFlag(taskDefinitionLog.getTimeoutFlag());
+ taskDefinition.setTaskTimeoutStrategy(taskDefinitionLog.getTaskTimeoutStrategy());
+ taskDefinition.setTimeout(taskDefinitionLog.getTimeout());
+ taskDefinition.setUpdateTime(new Date());
+ taskDefinition.setResourceIds(taskDefinitionLog.getResourceIds());
+ taskDefinitionMapper.updateById(taskDefinition);
Review comment:
missing setId or updateByCode here?
##########
File path: dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskDefinitionServiceImpl.java
##########
@@ -204,5 +211,188 @@ private String getResourceIds(TaskDefinition taskDefinition) {
return result;
}
+ /**
+ * delete task definition
+ *
+ * @param loginUser login user
+ * @param projectName project name
+ * @param taskCode task code
+ */
+ @Transactional(rollbackFor = RuntimeException.class)
+ @Override
+ public Map<String, Object> deleteTaskDefinitionByCode(User loginUser, String projectName, Long taskCode) {
+ Map<String, Object> result = new HashMap<>(5);
+ Project project = projectMapper.queryByName(projectName);
+
+ Map<String, Object> checkResult = projectService.checkProjectAndAuth(loginUser, project, projectName);
+ Status resultEnum = (Status) checkResult.get(Constants.STATUS);
+ if (resultEnum != Status.SUCCESS) {
+ return checkResult;
+ }
+ checkTaskRelation(result, taskCode);
+ resultEnum = (Status) result.get(Constants.STATUS);
+ if (resultEnum != Status.SUCCESS) {
+ return result;
+ }
+ int delete = taskDefinitionMapper.deleteByCode(taskCode);
+ if (delete > 0) {
+ putMsg(result, Status.SUCCESS);
+ } else {
+ putMsg(result, Status.DELETE_PROCESS_DEFINE_BY_ID_ERROR);
+ }
+ return result;
+ }
+
+ /**
+ * update task definition
+ *
+ * @param loginUser login user
+ * @param projectName project name
+ * @param taskCode task code
+ * @param taskDefinitionJson task definition json
+ */
+ @Override
+ public Map<String, Object> updateTaskDefinition(User loginUser, String projectName, Long taskCode, String taskDefinitionJson) {
+ Map<String, Object> result = new HashMap<>(5);
+ Project project = projectMapper.queryByName(projectName);
+
+ Map<String, Object> checkResult = projectService.checkProjectAndAuth(loginUser, project, projectName);
+ Status resultEnum = (Status) checkResult.get(Constants.STATUS);
+ if (resultEnum != Status.SUCCESS) {
+ return checkResult;
+ }
+ checkTaskRelation(result, taskCode);
+ if (result.get(Constants.STATUS) != Status.SUCCESS) {
+ return result;
+ }
+ TaskDefinition taskDefinition = taskDefinitionMapper.queryByDefinitionCode(taskCode);
+ if (taskDefinition == null) {
+ putMsg(result, Status.TASK_DEFINE_NOT_EXIST, taskCode);
+ return result;
+ }
+ TaskNode taskNode = JSONUtils.parseObject(taskDefinitionJson, TaskNode.class);
+ checkTaskNode(result, taskNode, taskDefinitionJson);
+ if (result.get(Constants.STATUS) != Status.SUCCESS) {
+ return result;
+ }
+
+ List<TaskDefinitionLog> taskDefinitionLogs = taskDefinitionLogMapper.queryByDefinitionCode(taskCode);
+ int version = taskDefinitionLogs
+ .stream()
+ .map(TaskDefinitionLog::getVersion)
+ .max((x, y) -> x > y ? x : y)
+ .orElse(0) + 1;
+ Date now = new Date();
+ taskDefinition.setVersion(version);
+ taskDefinition.setCode(taskCode);
+ taskDefinition.setName(taskNode.getName());
+ taskDefinition.setDescription(taskNode.getDesc());
+ taskDefinition.setProjectCode(project.getCode());
+ taskDefinition.setUserId(loginUser.getId());
+ taskDefinition.setTaskType(TaskType.of(taskNode.getType()));
+ taskDefinition.setTaskParams(taskNode.getParams());
+ taskDefinition.setFlag(taskNode.isForbidden() ? Flag.NO : Flag.YES);
+ taskDefinition.setTaskPriority(taskNode.getTaskInstancePriority());
+ taskDefinition.setWorkerGroup(taskNode.getWorkerGroup());
+ taskDefinition.setFailRetryTimes(taskNode.getMaxRetryTimes());
+ taskDefinition.setFailRetryInterval(taskNode.getRetryInterval());
+ taskDefinition.setTimeoutFlag(taskNode.getTaskTimeoutParameter().getEnable() ? TimeoutFlag.OPEN : TimeoutFlag.CLOSE);
+ taskDefinition.setTaskTimeoutStrategy(taskNode.getTaskTimeoutParameter().getStrategy());
+ taskDefinition.setTimeout(taskNode.getTaskTimeoutParameter().getInterval());
+ taskDefinition.setUpdateTime(now);
+ taskDefinition.setResourceIds(getResourceIds(taskDefinition));
+ taskDefinitionMapper.updateById(taskDefinition);
Review comment:
missing setId or updateByCode here?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-dolphinscheduler] JinyLeeChina commented on pull request #4654: [Feature][JsonSplit] add task delete/update/switch method
Posted by GitBox <gi...@apache.org>.
JinyLeeChina commented on pull request #4654:
URL: https://github.com/apache/incubator-dolphinscheduler/pull/4654#issuecomment-771375917
> result is a map, which is passed into the method through the parameter
> result is a map, which is passed into the method through the parameter
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-dolphinscheduler] lenboo merged pull request #4654: [Feature][JsonSplit] add task delete/update/switch method
Posted by GitBox <gi...@apache.org>.
lenboo merged pull request #4654:
URL: https://github.com/apache/incubator-dolphinscheduler/pull/4654
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-dolphinscheduler] simon824 commented on a change in pull request #4654: [Feature][JsonSplit] add task delete/update/switch method
Posted by GitBox <gi...@apache.org>.
simon824 commented on a change in pull request #4654:
URL: https://github.com/apache/incubator-dolphinscheduler/pull/4654#discussion_r568452836
##########
File path: dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskDefinitionServiceImpl.java
##########
@@ -204,5 +211,188 @@ private String getResourceIds(TaskDefinition taskDefinition) {
return result;
}
+ /**
+ * delete task definition
+ *
+ * @param loginUser login user
+ * @param projectName project name
+ * @param taskCode task code
+ */
+ @Transactional(rollbackFor = RuntimeException.class)
+ @Override
+ public Map<String, Object> deleteTaskDefinitionByCode(User loginUser, String projectName, Long taskCode) {
+ Map<String, Object> result = new HashMap<>(5);
+ Project project = projectMapper.queryByName(projectName);
+
+ Map<String, Object> checkResult = projectService.checkProjectAndAuth(loginUser, project, projectName);
+ Status resultEnum = (Status) checkResult.get(Constants.STATUS);
+ if (resultEnum != Status.SUCCESS) {
+ return checkResult;
+ }
+ checkTaskRelation(result, taskCode);
+ resultEnum = (Status) result.get(Constants.STATUS);
+ if (resultEnum != Status.SUCCESS) {
+ return result;
+ }
+ int delete = taskDefinitionMapper.deleteByCode(taskCode);
+ if (delete > 0) {
+ putMsg(result, Status.SUCCESS);
+ } else {
+ putMsg(result, Status.DELETE_PROCESS_DEFINE_BY_ID_ERROR);
+ }
+ return result;
+ }
+
+ /**
+ * update task definition
+ *
+ * @param loginUser login user
+ * @param projectName project name
+ * @param taskCode task code
+ * @param taskDefinitionJson task definition json
+ */
+ @Override
+ public Map<String, Object> updateTaskDefinition(User loginUser, String projectName, Long taskCode, String taskDefinitionJson) {
+ Map<String, Object> result = new HashMap<>(5);
+ Project project = projectMapper.queryByName(projectName);
+
+ Map<String, Object> checkResult = projectService.checkProjectAndAuth(loginUser, project, projectName);
+ Status resultEnum = (Status) checkResult.get(Constants.STATUS);
+ if (resultEnum != Status.SUCCESS) {
+ return checkResult;
+ }
+ checkTaskRelation(result, taskCode);
+ if (result.get(Constants.STATUS) != Status.SUCCESS) {
+ return result;
+ }
+ TaskDefinition taskDefinition = taskDefinitionMapper.queryByDefinitionCode(taskCode);
+ if (taskDefinition == null) {
+ putMsg(result, Status.TASK_DEFINE_NOT_EXIST, taskCode);
+ return result;
+ }
+ TaskNode taskNode = JSONUtils.parseObject(taskDefinitionJson, TaskNode.class);
+ checkTaskNode(result, taskNode, taskDefinitionJson);
+ if (result.get(Constants.STATUS) != Status.SUCCESS) {
+ return result;
+ }
+
+ List<TaskDefinitionLog> taskDefinitionLogs = taskDefinitionLogMapper.queryByDefinitionCode(taskCode);
+ int version = taskDefinitionLogs
+ .stream()
+ .map(TaskDefinitionLog::getVersion)
+ .max((x, y) -> x > y ? x : y)
+ .orElse(0) + 1;
+ Date now = new Date();
+ taskDefinition.setVersion(version);
+ taskDefinition.setCode(taskCode);
+ taskDefinition.setName(taskNode.getName());
+ taskDefinition.setDescription(taskNode.getDesc());
+ taskDefinition.setProjectCode(project.getCode());
+ taskDefinition.setUserId(loginUser.getId());
+ taskDefinition.setTaskType(TaskType.of(taskNode.getType()));
+ taskDefinition.setTaskParams(taskNode.getParams());
+ taskDefinition.setFlag(taskNode.isForbidden() ? Flag.NO : Flag.YES);
+ taskDefinition.setTaskPriority(taskNode.getTaskInstancePriority());
+ taskDefinition.setWorkerGroup(taskNode.getWorkerGroup());
+ taskDefinition.setFailRetryTimes(taskNode.getMaxRetryTimes());
+ taskDefinition.setFailRetryInterval(taskNode.getRetryInterval());
+ taskDefinition.setTimeoutFlag(taskNode.getTaskTimeoutParameter().getEnable() ? TimeoutFlag.OPEN : TimeoutFlag.CLOSE);
+ taskDefinition.setTaskTimeoutStrategy(taskNode.getTaskTimeoutParameter().getStrategy());
+ taskDefinition.setTimeout(taskNode.getTaskTimeoutParameter().getInterval());
+ taskDefinition.setUpdateTime(now);
+ taskDefinition.setResourceIds(getResourceIds(taskDefinition));
+ taskDefinitionMapper.updateById(taskDefinition);
+ // save task definition log
+ TaskDefinitionLog taskDefinitionLog = new TaskDefinitionLog();
+ taskDefinitionLog.set(taskDefinition);
+ taskDefinitionLog.setOperator(loginUser.getId());
+ taskDefinitionLog.setOperateTime(now);
+ taskDefinitionLogMapper.insert(taskDefinitionLog);
+ result.put(Constants.DATA_LIST, taskCode);
+ putMsg(result, Status.SUCCESS);
+ return result;
+ }
+
+ public void checkTaskRelation(Map<String, Object> result, Long taskCode) {
+ List<ProcessTaskRelation> processTaskRelationList = processTaskRelationMapper.queryByTaskCode(taskCode, taskCode);
+ if (!processTaskRelationList.isEmpty()) {
+ Set<Long> processDefinitionCodes = processTaskRelationList
+ .stream()
+ .map(ProcessTaskRelation::getProcessDefinitionCode)
+ .collect(Collectors.toSet());
+ List<ProcessDefinition> processDefinitionList = processDefinitionMapper.queryByCodes(processDefinitionCodes);
+ // check process definition is already online
+ for (ProcessDefinition processDefinition : processDefinitionList) {
+ if (processDefinition.getReleaseState() == ReleaseState.ONLINE) {
+ putMsg(result, Status.PROCESS_DEFINE_STATE_ONLINE, processDefinition.getCode());
+ return;
+ }
+ }
+ }
+ }
+
+ public void checkTaskNode(Map<String, Object> result, TaskNode taskNode, String taskDefinitionJson) {
+ if (taskNode == null) {
+ logger.error("taskDefinitionJson is not valid json");
+ putMsg(result, Status.DATA_IS_NOT_VALID, taskDefinitionJson);
+ return;
+ }
+ if (!CheckUtils.checkTaskNodeParameters(taskNode.getParams(), taskNode.getName())) {
+ logger.error("task node {} parameter invalid", taskNode.getName());
+ putMsg(result, Status.PROCESS_NODE_S_PARAMETER_INVALID, taskNode.getName());
+ }
+ }
+
+ /**
+ * update task definition
+ *
+ * @param loginUser login user
+ * @param projectName project name
+ * @param taskCode task code
+ * @param version the version user want to switch
+ */
+ @Override
+ public Map<String, Object> switchVersion(User loginUser, String projectName, Long taskCode, int version) {
+ Map<String, Object> result = new HashMap<>(5);
+ Project project = projectMapper.queryByName(projectName);
+
+ Map<String, Object> checkResult = projectService.checkProjectAndAuth(loginUser, project, projectName);
+ Status resultEnum = (Status) checkResult.get(Constants.STATUS);
+ if (resultEnum != Status.SUCCESS) {
+ return checkResult;
+ }
+ checkTaskRelation(result, taskCode);
+ if (result.get(Constants.STATUS) != Status.SUCCESS) {
+ return result;
+ }
+ TaskDefinition taskDefinition = taskDefinitionMapper.queryByDefinitionCode(taskCode);
+ if (taskDefinition == null) {
+ putMsg(result, Status.TASK_DEFINE_NOT_EXIST, taskCode);
+ return result;
+ }
+ TaskDefinitionLog taskDefinitionLog = taskDefinitionLogMapper.queryByDefinitionCodeAndVersion(taskCode, version);
+ taskDefinition.setVersion(version);
+ taskDefinition.setCode(taskCode);
+ taskDefinition.setName(taskDefinitionLog.getName());
+ taskDefinition.setDescription(taskDefinitionLog.getDescription());
+ taskDefinition.setProjectCode(taskDefinitionLog.getProjectCode());
+ taskDefinition.setUserId(loginUser.getId());
+ taskDefinition.setTaskType(taskDefinitionLog.getTaskType());
+ taskDefinition.setTaskParams(taskDefinitionLog.getTaskParams());
+ taskDefinition.setFlag(taskDefinitionLog.getFlag());
+ taskDefinition.setTaskPriority(taskDefinitionLog.getTaskPriority());
+ taskDefinition.setWorkerGroup(taskDefinitionLog.getWorkerGroup());
+ taskDefinition.setFailRetryTimes(taskDefinitionLog.getFailRetryTimes());
+ taskDefinition.setFailRetryInterval(taskDefinitionLog.getFailRetryInterval());
+ taskDefinition.setTimeoutFlag(taskDefinitionLog.getTimeoutFlag());
+ taskDefinition.setTaskTimeoutStrategy(taskDefinitionLog.getTaskTimeoutStrategy());
+ taskDefinition.setTimeout(taskDefinitionLog.getTimeout());
+ taskDefinition.setUpdateTime(new Date());
+ taskDefinition.setResourceIds(taskDefinitionLog.getResourceIds());
+ taskDefinitionMapper.updateById(taskDefinition);
Review comment:
missing setId or updateByCode here?
##########
File path: dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/ProcessTaskRelationService.java
##########
@@ -48,5 +48,77 @@
String conditionType,
String conditionParams);
+
+ /**
+ * query process task relation
+ *
+ * @param loginUser login user
+ * @param projectName project name
+ * @param processDefinitionCode process definition code
+ */
+ Map<String, Object> queryProcessTaskRelation(User loginUser,
+ String projectName,
+ Long processDefinitionCode);
+
+ /**
+ * delete process task relation
+ *
+ * @param loginUser login user
+ * @param projectName project name
+ * @param processDefinitionCode process definition code
+ */
+ Map<String, Object> deleteTaskDefinitionByProcess(User loginUser,
+ String projectName,
+ Long processDefinitionCode);
+
Review comment:
The function name deleteTaskDefinitionByProcessCode maybe more easy to understand
##########
File path: dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskDefinitionServiceImpl.java
##########
@@ -204,5 +211,188 @@ private String getResourceIds(TaskDefinition taskDefinition) {
return result;
}
+ /**
+ * delete task definition
+ *
+ * @param loginUser login user
+ * @param projectName project name
+ * @param taskCode task code
+ */
+ @Transactional(rollbackFor = RuntimeException.class)
+ @Override
+ public Map<String, Object> deleteTaskDefinitionByCode(User loginUser, String projectName, Long taskCode) {
+ Map<String, Object> result = new HashMap<>(5);
+ Project project = projectMapper.queryByName(projectName);
+
+ Map<String, Object> checkResult = projectService.checkProjectAndAuth(loginUser, project, projectName);
+ Status resultEnum = (Status) checkResult.get(Constants.STATUS);
+ if (resultEnum != Status.SUCCESS) {
+ return checkResult;
+ }
+ checkTaskRelation(result, taskCode);
+ resultEnum = (Status) result.get(Constants.STATUS);
+ if (resultEnum != Status.SUCCESS) {
+ return result;
+ }
+ int delete = taskDefinitionMapper.deleteByCode(taskCode);
+ if (delete > 0) {
+ putMsg(result, Status.SUCCESS);
+ } else {
+ putMsg(result, Status.DELETE_PROCESS_DEFINE_BY_ID_ERROR);
+ }
+ return result;
+ }
+
+ /**
+ * update task definition
+ *
+ * @param loginUser login user
+ * @param projectName project name
+ * @param taskCode task code
+ * @param taskDefinitionJson task definition json
+ */
+ @Override
+ public Map<String, Object> updateTaskDefinition(User loginUser, String projectName, Long taskCode, String taskDefinitionJson) {
+ Map<String, Object> result = new HashMap<>(5);
+ Project project = projectMapper.queryByName(projectName);
+
+ Map<String, Object> checkResult = projectService.checkProjectAndAuth(loginUser, project, projectName);
+ Status resultEnum = (Status) checkResult.get(Constants.STATUS);
+ if (resultEnum != Status.SUCCESS) {
+ return checkResult;
+ }
+ checkTaskRelation(result, taskCode);
+ if (result.get(Constants.STATUS) != Status.SUCCESS) {
+ return result;
+ }
+ TaskDefinition taskDefinition = taskDefinitionMapper.queryByDefinitionCode(taskCode);
+ if (taskDefinition == null) {
+ putMsg(result, Status.TASK_DEFINE_NOT_EXIST, taskCode);
+ return result;
+ }
+ TaskNode taskNode = JSONUtils.parseObject(taskDefinitionJson, TaskNode.class);
+ checkTaskNode(result, taskNode, taskDefinitionJson);
+ if (result.get(Constants.STATUS) != Status.SUCCESS) {
+ return result;
+ }
+
+ List<TaskDefinitionLog> taskDefinitionLogs = taskDefinitionLogMapper.queryByDefinitionCode(taskCode);
+ int version = taskDefinitionLogs
+ .stream()
+ .map(TaskDefinitionLog::getVersion)
+ .max((x, y) -> x > y ? x : y)
+ .orElse(0) + 1;
+ Date now = new Date();
+ taskDefinition.setVersion(version);
+ taskDefinition.setCode(taskCode);
+ taskDefinition.setName(taskNode.getName());
+ taskDefinition.setDescription(taskNode.getDesc());
+ taskDefinition.setProjectCode(project.getCode());
+ taskDefinition.setUserId(loginUser.getId());
+ taskDefinition.setTaskType(TaskType.of(taskNode.getType()));
+ taskDefinition.setTaskParams(taskNode.getParams());
+ taskDefinition.setFlag(taskNode.isForbidden() ? Flag.NO : Flag.YES);
+ taskDefinition.setTaskPriority(taskNode.getTaskInstancePriority());
+ taskDefinition.setWorkerGroup(taskNode.getWorkerGroup());
+ taskDefinition.setFailRetryTimes(taskNode.getMaxRetryTimes());
+ taskDefinition.setFailRetryInterval(taskNode.getRetryInterval());
+ taskDefinition.setTimeoutFlag(taskNode.getTaskTimeoutParameter().getEnable() ? TimeoutFlag.OPEN : TimeoutFlag.CLOSE);
+ taskDefinition.setTaskTimeoutStrategy(taskNode.getTaskTimeoutParameter().getStrategy());
+ taskDefinition.setTimeout(taskNode.getTaskTimeoutParameter().getInterval());
+ taskDefinition.setUpdateTime(now);
+ taskDefinition.setResourceIds(getResourceIds(taskDefinition));
+ taskDefinitionMapper.updateById(taskDefinition);
Review comment:
missing setId or updateByCode here?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-dolphinscheduler] JinyLeeChina commented on a change in pull request #4654: [Feature][JsonSplit] add task delete/update/switch method
Posted by GitBox <gi...@apache.org>.
JinyLeeChina commented on a change in pull request #4654:
URL: https://github.com/apache/incubator-dolphinscheduler/pull/4654#discussion_r568484233
##########
File path: dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/ProcessTaskRelationService.java
##########
@@ -48,5 +48,77 @@
String conditionType,
String conditionParams);
+
+ /**
+ * query process task relation
+ *
+ * @param loginUser login user
+ * @param projectName project name
+ * @param processDefinitionCode process definition code
+ */
+ Map<String, Object> queryProcessTaskRelation(User loginUser,
+ String projectName,
+ Long processDefinitionCode);
+
+ /**
+ * delete process task relation
+ *
+ * @param loginUser login user
+ * @param projectName project name
+ * @param processDefinitionCode process definition code
+ */
+ Map<String, Object> deleteTaskDefinitionByProcess(User loginUser,
+ String projectName,
+ Long processDefinitionCode);
+
Review comment:
OK
##########
File path: dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/ProcessTaskRelationService.java
##########
@@ -48,5 +48,77 @@
String conditionType,
String conditionParams);
+
+ /**
+ * query process task relation
+ *
+ * @param loginUser login user
+ * @param projectName project name
+ * @param processDefinitionCode process definition code
+ */
+ Map<String, Object> queryProcessTaskRelation(User loginUser,
+ String projectName,
+ Long processDefinitionCode);
+
+ /**
+ * delete process task relation
+ *
+ * @param loginUser login user
+ * @param projectName project name
+ * @param processDefinitionCode process definition code
+ */
+ Map<String, Object> deleteTaskDefinitionByProcess(User loginUser,
+ String projectName,
+ Long processDefinitionCode);
+
Review comment:
OK, right. Very good
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-dolphinscheduler] JinyLeeChina removed a comment on pull request #4654: [Feature][JsonSplit] add task delete/update/switch method
Posted by GitBox <gi...@apache.org>.
JinyLeeChina removed a comment on pull request #4654:
URL: https://github.com/apache/incubator-dolphinscheduler/pull/4654#issuecomment-771375917
> result is a map, which is passed into the method through the parameter
> result is a map, which is passed into the method through the parameter
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-dolphinscheduler] lenboo commented on a change in pull request #4654: [Feature][JsonSplit] add task delete/update/switch method
Posted by GitBox <gi...@apache.org>.
lenboo commented on a change in pull request #4654:
URL: https://github.com/apache/incubator-dolphinscheduler/pull/4654#discussion_r568299687
##########
File path: dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskDefinitionServiceImpl.java
##########
@@ -100,14 +111,8 @@
}
TaskNode taskNode = JSONUtils.parseObject(taskDefinitionJson, TaskNode.class);
- if (taskNode == null) {
- logger.error("taskDefinitionJson is not valid json");
- putMsg(result, Status.DATA_IS_NOT_VALID, taskDefinitionJson);
- return result;
- }
- if (!CheckUtils.checkTaskNodeParameters(taskNode.getParams(), taskNode.getName())) {
- logger.error("task node {} parameter invalid", taskNode.getName());
- putMsg(result, Status.PROCESS_NODE_S_PARAMETER_INVALID, taskNode.getName());
+ checkTaskNode(result, taskNode, taskDefinitionJson);
Review comment:
missing 'result = ？'
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-dolphinscheduler] lenboo merged pull request #4654: [Feature][JsonSplit] add task delete/update/switch method
Posted by GitBox <gi...@apache.org>.
lenboo merged pull request #4654:
URL: https://github.com/apache/incubator-dolphinscheduler/pull/4654
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-dolphinscheduler] JinyLeeChina commented on a change in pull request #4654: [Feature][JsonSplit] add task delete/update/switch method
Posted by GitBox <gi...@apache.org>.
JinyLeeChina commented on a change in pull request #4654:
URL: https://github.com/apache/incubator-dolphinscheduler/pull/4654#discussion_r568484233
##########
File path: dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/ProcessTaskRelationService.java
##########
@@ -48,5 +48,77 @@
String conditionType,
String conditionParams);
+
+ /**
+ * query process task relation
+ *
+ * @param loginUser login user
+ * @param projectName project name
+ * @param processDefinitionCode process definition code
+ */
+ Map<String, Object> queryProcessTaskRelation(User loginUser,
+ String projectName,
+ Long processDefinitionCode);
+
+ /**
+ * delete process task relation
+ *
+ * @param loginUser login user
+ * @param projectName project name
+ * @param processDefinitionCode process definition code
+ */
+ Map<String, Object> deleteTaskDefinitionByProcess(User loginUser,
+ String projectName,
+ Long processDefinitionCode);
+
Review comment:
OK, right. Very good
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-dolphinscheduler] simon824 commented on a change in pull request #4654: [Feature][JsonSplit] add task delete/update/switch method
Posted by GitBox <gi...@apache.org>.
simon824 commented on a change in pull request #4654:
URL: https://github.com/apache/incubator-dolphinscheduler/pull/4654#discussion_r568455134
##########
File path: dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskDefinitionServiceImpl.java
##########
@@ -204,5 +211,188 @@ private String getResourceIds(TaskDefinition taskDefinition) {
return result;
}
+ /**
+ * delete task definition
+ *
+ * @param loginUser login user
+ * @param projectName project name
+ * @param taskCode task code
+ */
+ @Transactional(rollbackFor = RuntimeException.class)
+ @Override
+ public Map<String, Object> deleteTaskDefinitionByCode(User loginUser, String projectName, Long taskCode) {
+ Map<String, Object> result = new HashMap<>(5);
+ Project project = projectMapper.queryByName(projectName);
+
+ Map<String, Object> checkResult = projectService.checkProjectAndAuth(loginUser, project, projectName);
+ Status resultEnum = (Status) checkResult.get(Constants.STATUS);
+ if (resultEnum != Status.SUCCESS) {
+ return checkResult;
+ }
+ checkTaskRelation(result, taskCode);
+ resultEnum = (Status) result.get(Constants.STATUS);
+ if (resultEnum != Status.SUCCESS) {
+ return result;
+ }
+ int delete = taskDefinitionMapper.deleteByCode(taskCode);
+ if (delete > 0) {
+ putMsg(result, Status.SUCCESS);
+ } else {
+ putMsg(result, Status.DELETE_PROCESS_DEFINE_BY_ID_ERROR);
+ }
+ return result;
+ }
+
+ /**
+ * update task definition
+ *
+ * @param loginUser login user
+ * @param projectName project name
+ * @param taskCode task code
+ * @param taskDefinitionJson task definition json
+ */
+ @Override
+ public Map<String, Object> updateTaskDefinition(User loginUser, String projectName, Long taskCode, String taskDefinitionJson) {
+ Map<String, Object> result = new HashMap<>(5);
+ Project project = projectMapper.queryByName(projectName);
+
+ Map<String, Object> checkResult = projectService.checkProjectAndAuth(loginUser, project, projectName);
+ Status resultEnum = (Status) checkResult.get(Constants.STATUS);
+ if (resultEnum != Status.SUCCESS) {
+ return checkResult;
+ }
+ checkTaskRelation(result, taskCode);
+ if (result.get(Constants.STATUS) != Status.SUCCESS) {
+ return result;
+ }
+ TaskDefinition taskDefinition = taskDefinitionMapper.queryByDefinitionCode(taskCode);
+ if (taskDefinition == null) {
+ putMsg(result, Status.TASK_DEFINE_NOT_EXIST, taskCode);
+ return result;
+ }
+ TaskNode taskNode = JSONUtils.parseObject(taskDefinitionJson, TaskNode.class);
+ checkTaskNode(result, taskNode, taskDefinitionJson);
+ if (result.get(Constants.STATUS) != Status.SUCCESS) {
+ return result;
+ }
+
+ List<TaskDefinitionLog> taskDefinitionLogs = taskDefinitionLogMapper.queryByDefinitionCode(taskCode);
+ int version = taskDefinitionLogs
+ .stream()
+ .map(TaskDefinitionLog::getVersion)
+ .max((x, y) -> x > y ? x : y)
+ .orElse(0) + 1;
+ Date now = new Date();
+ taskDefinition.setVersion(version);
+ taskDefinition.setCode(taskCode);
+ taskDefinition.setName(taskNode.getName());
+ taskDefinition.setDescription(taskNode.getDesc());
+ taskDefinition.setProjectCode(project.getCode());
+ taskDefinition.setUserId(loginUser.getId());
+ taskDefinition.setTaskType(TaskType.of(taskNode.getType()));
+ taskDefinition.setTaskParams(taskNode.getParams());
+ taskDefinition.setFlag(taskNode.isForbidden() ? Flag.NO : Flag.YES);
+ taskDefinition.setTaskPriority(taskNode.getTaskInstancePriority());
+ taskDefinition.setWorkerGroup(taskNode.getWorkerGroup());
+ taskDefinition.setFailRetryTimes(taskNode.getMaxRetryTimes());
+ taskDefinition.setFailRetryInterval(taskNode.getRetryInterval());
+ taskDefinition.setTimeoutFlag(taskNode.getTaskTimeoutParameter().getEnable() ? TimeoutFlag.OPEN : TimeoutFlag.CLOSE);
+ taskDefinition.setTaskTimeoutStrategy(taskNode.getTaskTimeoutParameter().getStrategy());
+ taskDefinition.setTimeout(taskNode.getTaskTimeoutParameter().getInterval());
+ taskDefinition.setUpdateTime(now);
+ taskDefinition.setResourceIds(getResourceIds(taskDefinition));
+ taskDefinitionMapper.updateById(taskDefinition);
Review comment:
missing setId or updateByCode here?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org