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); + 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); + 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.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 cmdParam = JSONUtils.toMap(command.getCommandParam()); + List schedules = processService.queryReleaseSchedulerListByProcessDefinitionCode(command.getProcessDefinitionCode()); + LinkedList 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 listDate = new LinkedList<>(); - List 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 getSelfFireDateList(final Date startTime, final Date endTime, final List schedules) { List 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);