You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@dolphinscheduler.apache.org by ji...@apache.org on 2023/01/11 07:20:39 UTC
[dolphinscheduler] branch 2.0.8-prepare updated: [Fix-13292] [Api] When serial complement is a process configured with timing scheduling, an error will occur when selecting a time that is not within the timing range (#13356)
This is an automated email from the ASF dual-hosted git repository.
jinyleechina pushed a commit to branch 2.0.8-prepare
in repository https://gitbox.apache.org/repos/asf/dolphinscheduler.git
The following commit(s) were added to refs/heads/2.0.8-prepare by this push:
new a7ddc9047b [Fix-13292] [Api] When serial complement is a process configured with timing scheduling, an error will occur when selecting a time that is not within the timing range (#13356)
a7ddc9047b is described below
commit a7ddc9047b460bb03d5e62d8831db83ba1525ecf
Author: Molin Wang <wc...@gmail.com>
AuthorDate: Wed Jan 11 15:20:32 2023 +0800
[Fix-13292] [Api] When serial complement is a process configured with timing scheduling, an error will occur when selecting a time that is not within the timing range (#13356)
* [Fix-13292] [Api] When serial complement is a process configured with timing scheduling, an error will occur when selecting a time that is not within the timing range
* [Fix-13292] [Api] When serial complement is a process configured with timing scheduling, an error will occur when selecting a time that is not within the timing range
---
.../apache/dolphinscheduler/api/enums/Status.java | 2 ++
.../api/service/impl/ExecutorServiceImpl.java | 39 ++++++++++++++--------
.../service/quartz/cron/CronUtils.java | 4 ---
3 files changed, 27 insertions(+), 18 deletions(-)
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/enums/Status.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/enums/Status.java
index 0dd6cbf36d..90dbe771d4 100644
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/enums/Status.java
+++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/enums/Status.java
@@ -295,6 +295,8 @@ public enum Status {
DELETE_EDGE_ERROR(50055, "delete edge error", "删除工作流任务连接线错误"),
NOT_SUPPORT_UPDATE_TASK_DEFINITION(50056, "task state does not support modification", "当前任务不支持修改"),
NOT_SUPPORT_COPY_TASK_TYPE(50057, "task type [{0}] does not support copy", "不支持复制的任务类型[{0}]"),
+ COMPLEMENT_PROCESS_INSTANCE_DATE_RANGE_ERROR(50059, "complement instances cannot be generated within the complement time range according to the schedule configuration",
+ "根据调度配置在补数时间范围内无法生成补数实例"),
HDFS_NOT_STARTUP(60001, "hdfs not startup", "hdfs未启用"),
/**
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ExecutorServiceImpl.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ExecutorServiceImpl.java
index 1b4cee852e..fe21842f72 100644
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ExecutorServiceImpl.java
+++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ExecutorServiceImpl.java
@@ -23,6 +23,7 @@ import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.dolphinscheduler.api.enums.ExecuteType;
import org.apache.dolphinscheduler.api.enums.Status;
+import org.apache.dolphinscheduler.api.exceptions.ServiceException;
import org.apache.dolphinscheduler.api.service.ExecutorService;
import org.apache.dolphinscheduler.api.service.MonitorService;
import org.apache.dolphinscheduler.api.service.ProjectService;
@@ -144,17 +145,23 @@ public class ExecutorServiceImpl extends BaseServiceImpl implements ExecutorServ
/**
* create command
*/
- int create = this.createCommand(commandType, processDefinition.getCode(),
- taskDependType, failureStrategy, startNodeList, cronTime, warningType, loginUser.getId(),
- warningGroupId, runMode, processInstancePriority, workerGroup, environmentCode, startParams, expectedParallelismNumber, dryRun);
-
- if (create > 0) {
- processDefinition.setWarningGroupId(warningGroupId);
- processDefinitionMapper.updateById(processDefinition);
- putMsg(result, Status.SUCCESS);
- } else {
- putMsg(result, Status.START_PROCESS_INSTANCE_ERROR);
+ try {
+ int create = this.createCommand(commandType, processDefinition.getCode(),
+ taskDependType, failureStrategy, startNodeList, cronTime, warningType, loginUser.getId(),
+ warningGroupId, runMode, processInstancePriority, workerGroup, environmentCode, startParams, expectedParallelismNumber, dryRun);
+
+ if (create > 0) {
+ processDefinition.setWarningGroupId(warningGroupId);
+ processDefinitionMapper.updateById(processDefinition);
+ putMsg(result, Status.SUCCESS);
+ } else {
+ putMsg(result, Status.START_PROCESS_INSTANCE_ERROR);
+ }
+ } catch (ServiceException e) {
+ Optional<Status> status = Status.findStatusBy(e.getCode());
+ putMsg(result, status.orElse(Status.START_PROCESS_INSTANCE_ERROR));
}
+
return result;
}
@@ -569,6 +576,14 @@ public class ExecutorServiceImpl extends BaseServiceImpl implements ExecutorServ
int createCount = 0;
runMode = (runMode == null) ? RunMode.RUN_MODE_SERIAL : runMode;
Map<String, String> cmdParam = JSONUtils.toMap(command.getCommandParam());
+ List<Schedule> schedules = processService.queryReleaseSchedulerListByProcessDefinitionCode(command.getProcessDefinitionCode());
+ LinkedList<Date> listDate = new LinkedList<>(CronUtils.getSelfFireDateList(start, end, schedules));
+ final int listDateSize = listDate.size();
+ if (listDateSize == 0) {
+ logger.warn("can't create complement command, because the fire date cannot be created, scope: {} ~ {}",
+ DateUtils.dateToString(start), DateUtils.dateToString(end));
+ throw new ServiceException(Status.COMPLEMENT_PROCESS_INSTANCE_DATE_RANGE_ERROR);
+ }
switch (runMode) {
case RUN_MODE_SERIAL: {
if (start.after(end)) {
@@ -587,10 +602,6 @@ public class ExecutorServiceImpl extends BaseServiceImpl implements ExecutorServ
break;
}
- LinkedList<Date> listDate = new LinkedList<>();
- List<Schedule> schedules = processService.queryReleaseSchedulerListByProcessDefinitionCode(command.getProcessDefinitionCode());
- listDate.addAll(CronUtils.getSelfFireDateList(start, end, schedules));
- int listDateSize = listDate.size();
createCount = listDate.size();
if (!CollectionUtils.isEmpty(listDate)) {
if (expectedParallelismNumber != null && expectedParallelismNumber != 0) {
diff --git a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/quartz/cron/CronUtils.java b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/quartz/cron/CronUtils.java
index 49810cd1f4..d2e1479427 100644
--- a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/quartz/cron/CronUtils.java
+++ b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/quartz/cron/CronUtils.java
@@ -187,10 +187,6 @@ public class CronUtils {
*/
public static List<Date> getSelfFireDateList(final Date startTime, final Date endTime, final List<Schedule> schedules) {
List<Date> result = new ArrayList<>();
- if (startTime.equals(endTime)) {
- result.add(startTime);
- return result;
- }
// support left closed and right closed time interval (startDate <= N <= endDate)
Date from = new Date(startTime.getTime() - Constants.SECOND_TIME_MILLIS);