[Improve] Enhance complement function transformation (#10376)

docs/docs/en/about/glossary.md

@@ -45,7 +45,7 @@ provided. **Continue** refers to regardless of the status of the task running in
 failure. **End** means that once a failed task is found, Kill will also run the parallel task at the same time, and the
 process fails and ends
 
-**Complement**: Supplement historical data，Supports **interval parallel and serial** two complement methods
+**Complement**: Supplement historical data，supports **interval parallel and serial** two complement methods, and two types of date selection which include **date range** and **date enumeration**.
 
 ### 2.Module introduction
 

docs/docs/zh/about/glossary.md

@@ -30,7 +30,7 @@
 
 **失败策略**：对于并行运行的任务，如果有任务失败，提供两种失败策略处理方式，**继续**是指不管并行运行任务的状态，直到流程失败结束。**结束**是指一旦发现失败任务，则同时Kill掉正在运行的并行任务，流程失败结束
 
-**补数**：补历史数据，支持**区间并行和串行**两种补数方式
+**补数**：补历史数据，支持**区间并行和串行**两种补数方式，其日期选择方式包括**日期范围**和**日期枚举**两种
 
 ### 2.模块介绍
 

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/ExecutorController.java

@@ -87,7 +87,7 @@ public class ExecutorController extends BaseController {
      * @param loginUser login user
      * @param projectCode project code
      * @param processDefinitionCode process definition code
-     * @param scheduleTime schedule time
+     * @param scheduleTime schedule time when CommandType is COMPLEMENT_DATA  there are two ways to transfer parameters 1.date range, for example:{"complementStartDate":"2022-01-01 12:12:12","complementEndDate":"2022-01-6 12:12:12"} 2.manual input,  for example:{"complementScheduleDateList":"2022-01-01 00:00:00,2022-01-02 12:12:12,2022-01-03 12:12:12"}
      * @param failureStrategy failure strategy
      * @param startNodeList start nodes list
      * @param taskDependType task depend type

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/enums/Status.java

@@ -407,6 +407,7 @@ public enum Status {
 
     NO_CURRENT_OPERATING_PERMISSION(1400001, "The current user does not have this permission.", "当前用户无此权限"),
     FUNCTION_DISABLED(1400002, "The current feature is disabled.", "当前功能已被禁用"),
+    SCHEDULE_TIME_NUMBER(1400003, "The number of complement dates exceed 100.", "补数日期个数超过100"),
     ;
 
     private final int code;

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ExecutorServiceImpl.java

@@ -17,14 +17,12 @@
 
 package org.apache.dolphinscheduler.api.service.impl;
 
-import static org.apache.dolphinscheduler.api.constants.ApiFuncIdentificationConstant.WORKFLOW_START;
-import static org.apache.dolphinscheduler.common.Constants.CMDPARAM_COMPLEMENT_DATA_END_DATE;
-import static org.apache.dolphinscheduler.common.Constants.CMDPARAM_COMPLEMENT_DATA_START_DATE;
-import static org.apache.dolphinscheduler.common.Constants.CMD_PARAM_RECOVER_PROCESS_ID_STRING;
-import static org.apache.dolphinscheduler.common.Constants.CMD_PARAM_START_NODES;
-import static org.apache.dolphinscheduler.common.Constants.CMD_PARAM_START_PARAMS;
-import static org.apache.dolphinscheduler.common.Constants.MAX_TASK_TIMEOUT;
-
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.google.common.collect.Lists;
+import org.apache.commons.beanutils.BeanUtils;
+import org.apache.commons.collections.CollectionUtils;
+import org.apache.commons.collections.MapUtils;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.dolphinscheduler.api.constants.ApiFuncIdentificationConstant;
 import org.apache.dolphinscheduler.api.enums.ExecuteType;
 import org.apache.dolphinscheduler.api.enums.Status;
@@ -69,13 +67,13 @@ import org.apache.dolphinscheduler.remote.command.StateEventChangeCommand;
 import org.apache.dolphinscheduler.remote.processor.StateEventCallbackService;
 import org.apache.dolphinscheduler.service.corn.CronUtils;
 import org.apache.dolphinscheduler.service.process.ProcessService;
-
-import org.apache.commons.beanutils.BeanUtils;
-import org.apache.commons.collections.CollectionUtils;
-import org.apache.commons.collections.MapUtils;
-import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
 
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Date;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -84,12 +82,16 @@ import java.util.Map;
 import java.util.Set;
 import java.util.stream.Collectors;
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.stereotype.Service;
-
-import com.fasterxml.jackson.core.type.TypeReference;
+import static org.apache.dolphinscheduler.api.constants.ApiFuncIdentificationConstant.WORKFLOW_START;
+import static org.apache.dolphinscheduler.common.Constants.CMDPARAM_COMPLEMENT_DATA_END_DATE;
+import static org.apache.dolphinscheduler.common.Constants.CMDPARAM_COMPLEMENT_DATA_SCHEDULE_DATE_LIST;
+import static org.apache.dolphinscheduler.common.Constants.CMDPARAM_COMPLEMENT_DATA_START_DATE;
+import static org.apache.dolphinscheduler.common.Constants.CMD_PARAM_RECOVER_PROCESS_ID_STRING;
+import static org.apache.dolphinscheduler.common.Constants.CMD_PARAM_START_NODES;
+import static org.apache.dolphinscheduler.common.Constants.CMD_PARAM_START_PARAMS;
+import static org.apache.dolphinscheduler.common.Constants.COMMA;
+import static org.apache.dolphinscheduler.common.Constants.MAX_TASK_TIMEOUT;
+import static org.apache.dolphinscheduler.common.Constants.SCHEDULE_TIME_MAX_LENGTH;
 
 /**
  * executor service impl
@@ -187,11 +189,15 @@ public class ExecutorServiceImpl extends BaseServiceImpl implements ExecutorServ
             return result;
         }
 
+        if(!checkScheduleTimeNum(commandType,cronTime)){
+            putMsg(result, Status.SCHEDULE_TIME_NUMBER);
+            return result;
+        }
+
         // check master exists
         if (!checkMasterExists(result)) {
             return result;
         }
-
         /**
          * create command
          */
@@ -228,6 +234,29 @@ public class ExecutorServiceImpl extends BaseServiceImpl implements ExecutorServ
         return true;
     }
 
+    /**
+     *
+     * @param complementData
+     * @param cronTime
+     * @return CommandType is COMPLEMENT_DATA and cronTime's number is not greater than 100 return true , otherwise return false
+     */
+    private boolean checkScheduleTimeNum(CommandType complementData,String cronTime) {
+        if (!CommandType.COMPLEMENT_DATA.equals(complementData)) {
+            return true;
+        }
+        if(cronTime == null){
+            return true;
+        }
+        Map<String,String> cronMap =  JSONUtils.toMap(cronTime);
+        if (cronMap.containsKey(CMDPARAM_COMPLEMENT_DATA_SCHEDULE_DATE_LIST)) {
+            String[] stringDates = cronMap.get(CMDPARAM_COMPLEMENT_DATA_SCHEDULE_DATE_LIST).split(COMMA);
+            if (stringDates.length > SCHEDULE_TIME_MAX_LENGTH) {
+                return false;
+            }
+        }
+        return true;
+    }
+
     /**
      * check whether the process definition can be executed
      *
@@ -655,27 +684,16 @@ public class ExecutorServiceImpl extends BaseServiceImpl implements ExecutorServ
         }
         command.setProcessInstanceId(0);
 
-        Date start = null;
-        Date end = null;
-        if (!StringUtils.isEmpty(schedule)) {
-            String[] interval = schedule.split(",");
-            if (interval.length == 2) {
-                start = DateUtils.getScheduleDate(interval[0]);
-                end = DateUtils.getScheduleDate(interval[1]);
-                if (start.after(end)) {
-                    logger.info("complement data error, wrong date start:{} and end date:{} ",
-                            start, end
-                    );
-                    return 0;
-                }
-            }
-        }
         // determine whether to complement
         if (commandType == CommandType.COMPLEMENT_DATA) {
-            if (start == null || end == null) {
+            if (schedule == null || StringUtils.isEmpty(schedule)) {
                 return 0;
             }
-            return createComplementCommandList(start, end, runMode, command, expectedParallelismNumber, complementDependentMode);
+            int check = checkScheduleTime(schedule);
+            if(check == 0){
+                return 0;
+            }
+            return createComplementCommandList(schedule, runMode, command, expectedParallelismNumber, complementDependentMode);
         } else {
             command.setCommandParam(JSONUtils.toJsonString(cmdParam));
             return processService.createCommand(command);
@@ -686,89 +704,117 @@ public class ExecutorServiceImpl extends BaseServiceImpl implements ExecutorServ
      * create complement command
      * close left and close right
      *
-     * @param start
-     * @param end
+     * @param scheduleTimeParam
      * @param runMode
      * @return
      */
-    protected int createComplementCommandList(Date start, Date end, RunMode runMode, Command command,
+    protected int createComplementCommandList(String scheduleTimeParam, RunMode runMode, Command command,
                                             Integer expectedParallelismNumber, ComplementDependentMode complementDependentMode) {
         int createCount = 0;
+        String startDate = null;
+        String endDate = null;
+        String dateList = null;
         int dependentProcessDefinitionCreateCount = 0;
-
         runMode = (runMode == null) ? RunMode.RUN_MODE_SERIAL : runMode;
         Map<String, String> cmdParam = JSONUtils.toMap(command.getCommandParam());
+        Map<String, String> scheduleParam = JSONUtils.toMap(scheduleTimeParam);
+        if(scheduleParam.containsKey(CMDPARAM_COMPLEMENT_DATA_SCHEDULE_DATE_LIST)){
+            dateList = scheduleParam.get(CMDPARAM_COMPLEMENT_DATA_SCHEDULE_DATE_LIST);
+            dateList = removeDuplicates(dateList);
+        }
+        if(scheduleParam.containsKey(CMDPARAM_COMPLEMENT_DATA_START_DATE) && scheduleParam.containsKey(CMDPARAM_COMPLEMENT_DATA_END_DATE)){
+            startDate = scheduleParam.get(CMDPARAM_COMPLEMENT_DATA_START_DATE);
+            endDate = scheduleParam.get(CMDPARAM_COMPLEMENT_DATA_END_DATE);
+        }
         switch (runMode) {
             case RUN_MODE_SERIAL: {
-                if (start.after(end)) {
-                    logger.warn("The startDate {} is later than the endDate {}", start, end);
-                    break;
+                if(StringUtils.isNotEmpty(dateList)){
+                    cmdParam.put(CMDPARAM_COMPLEMENT_DATA_SCHEDULE_DATE_LIST, dateList);
+                    command.setCommandParam(JSONUtils.toJsonString(cmdParam));
+                    createCount = processService.createCommand(command);
                 }
-                cmdParam.put(CMDPARAM_COMPLEMENT_DATA_START_DATE, DateUtils.dateToString(start));
-                cmdParam.put(CMDPARAM_COMPLEMENT_DATA_END_DATE, DateUtils.dateToString(end));
-                command.setCommandParam(JSONUtils.toJsonString(cmdParam));
-                createCount = processService.createCommand(command);
-
-                // dependent process definition
-                List<Schedule> schedules = processService.queryReleaseSchedulerListByProcessDefinitionCode(command.getProcessDefinitionCode());
-
-                if (schedules.isEmpty() || complementDependentMode == ComplementDependentMode.OFF_MODE) {
-                    logger.info("process code: {} complement dependent in off mode or schedule's size is 0, skip "
-                            + "dependent complement data", command.getProcessDefinitionCode());
-                } else {
-                    dependentProcessDefinitionCreateCount += createComplementDependentCommand(schedules, command);
+                if(startDate != null && endDate != null){
+                    cmdParam.put(CMDPARAM_COMPLEMENT_DATA_START_DATE, startDate);
+                    cmdParam.put(CMDPARAM_COMPLEMENT_DATA_END_DATE, endDate);
+                    command.setCommandParam(JSONUtils.toJsonString(cmdParam));
+                    createCount = processService.createCommand(command);
+
+                    // dependent process definition
+                    List<Schedule> schedules = processService.queryReleaseSchedulerListByProcessDefinitionCode(command.getProcessDefinitionCode());
+
+                    if (schedules.isEmpty() || complementDependentMode == ComplementDependentMode.OFF_MODE) {
+                        logger.info("process code: {} complement dependent in off mode or schedule's size is 0, skip "
+                                + "dependent complement data", command.getProcessDefinitionCode());
+                    } else {
+                        dependentProcessDefinitionCreateCount += createComplementDependentCommand(schedules, command);
+                    }
                 }
-
                 break;
             }
             case RUN_MODE_PARALLEL: {
-                if (start.after(end)) {
-                    logger.warn("The startDate {} is later than the endDate {}", start, end);
-                    break;
-                }
-
-                List<Date> listDate = new ArrayList<>();
-                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) {
-                        createCount = Math.min(listDate.size(), expectedParallelismNumber);
-                        if (listDateSize < createCount) {
-                            createCount = listDateSize;
+                if(startDate != null && endDate != null){
+                    List<Date> listDate = new ArrayList<>();
+                    List<Schedule> schedules = processService.queryReleaseSchedulerListByProcessDefinitionCode(command.getProcessDefinitionCode());
+                    listDate.addAll(CronUtils.getSelfFireDateList(DateUtils.getScheduleDate(startDate), DateUtils.getScheduleDate(endDate), schedules));
+                    int listDateSize = listDate.size();
+                    createCount = listDate.size();
+                    if (!CollectionUtils.isEmpty(listDate)) {
+                        if (expectedParallelismNumber != null && expectedParallelismNumber != 0) {
+                            createCount = Math.min(listDate.size(), expectedParallelismNumber);
+                            if (listDateSize < createCount) {
+                                createCount = listDateSize;
+                            }
+                        }
+                        logger.info("In parallel mode, current expectedParallelismNumber:{}", createCount);
+
+                        // Distribute the number of tasks equally to each command.
+                        // The last command with insufficient quantity will be assigned to the remaining tasks.
+                        int itemsPerCommand = (listDateSize / createCount);
+                        int remainingItems = (listDateSize % createCount);
+                        int startDateIndex = 0;
+                        int endDateIndex = 0;
+
+                        for (int i = 1; i <= createCount; i++) {
+                            int extra = (i <= remainingItems) ? 1 : 0;
+                            int singleCommandItems = (itemsPerCommand + extra);
+
+                            if (i == 1) {
+                                endDateIndex += singleCommandItems - 1;
+                            } else {
+                                startDateIndex = endDateIndex + 1;
+                                endDateIndex += singleCommandItems;
+                            }
+
+                            cmdParam.put(CMDPARAM_COMPLEMENT_DATA_START_DATE, DateUtils.dateToString(listDate.get(startDateIndex)));
+                            cmdParam.put(CMDPARAM_COMPLEMENT_DATA_END_DATE, DateUtils.dateToString(listDate.get(endDateIndex)));
+                            command.setCommandParam(JSONUtils.toJsonString(cmdParam));
+                            processService.createCommand(command);
+
+                            if (schedules.isEmpty() || complementDependentMode == ComplementDependentMode.OFF_MODE) {
+                                logger.info("process code: {} complement dependent in off mode or schedule's size is 0, skip "
+                                        + "dependent complement data", command.getProcessDefinitionCode());
+                            } else {
+                                dependentProcessDefinitionCreateCount += createComplementDependentCommand(schedules, command);
+                            }
                         }
                     }
-                    logger.info("In parallel mode, current expectedParallelismNumber:{}", createCount);
-
-                    // Distribute the number of tasks equally to each command.
-                    // The last command with insufficient quantity will be assigned to the remaining tasks.
-                    int itemsPerCommand = (listDateSize / createCount);
-                    int remainingItems = (listDateSize % createCount);
-                    int startDateIndex = 0;
-                    int endDateIndex = 0;
-
-                    for (int i = 1; i <= createCount; i++) {
-                        int extra = (i <= remainingItems) ? 1 : 0;
-                        int singleCommandItems = (itemsPerCommand + extra);
-
-                        if (i == 1) {
-                            endDateIndex += singleCommandItems - 1;
-                        } else {
-                            startDateIndex = endDateIndex + 1;
-                            endDateIndex += singleCommandItems;
+                }
+                if(StringUtils.isNotEmpty(dateList)){
+                    List<String> listDate = Arrays.asList(dateList.split(COMMA));
+                    int listDateSize = listDate.size();
+                    createCount = listDate.size();
+                    if (!CollectionUtils.isEmpty(listDate)) {
+                        if (expectedParallelismNumber != null && expectedParallelismNumber != 0) {
+                            createCount = Math.min(listDate.size(), expectedParallelismNumber);
+                            if (listDateSize < createCount) {
+                                createCount = listDateSize;
+                            }
                         }
-
-                        cmdParam.put(CMDPARAM_COMPLEMENT_DATA_START_DATE, DateUtils.dateToString(listDate.get(startDateIndex)));
-                        cmdParam.put(CMDPARAM_COMPLEMENT_DATA_END_DATE, DateUtils.dateToString(listDate.get(endDateIndex)));
-                        command.setCommandParam(JSONUtils.toJsonString(cmdParam));
-                        processService.createCommand(command);
-
-                        if (schedules.isEmpty() || complementDependentMode == ComplementDependentMode.OFF_MODE) {
-                            logger.info("process code: {} complement dependent in off mode or schedule's size is 0, skip "
-                                    + "dependent complement data", command.getProcessDefinitionCode());
-                        } else {
-                            dependentProcessDefinitionCreateCount += createComplementDependentCommand(schedules, command);
+                        logger.info("In parallel mode, current expectedParallelismNumber:{}", createCount);
+                        for (List<String> stringDate : Lists.partition(listDate, createCount)) {
+                            cmdParam.put(CMDPARAM_COMPLEMENT_DATA_SCHEDULE_DATE_LIST, String.join(COMMA, stringDate));
+                            command.setCommandParam(JSONUtils.toJsonString(cmdParam));
+                            processService.createCommand(command);
                         }
                     }
                 }
@@ -854,4 +900,60 @@ public class ExecutorServiceImpl extends BaseServiceImpl implements ExecutorServ
 
         return validDependentProcessDefinitionList;
     }
+
+    /**
+     *
+     * @param schedule
+     * @return check error return 0 otherwish 1
+     */
+    private int checkScheduleTime(String schedule){
+        Date start = null;
+        Date end = null;
+        Map<String,String> scheduleResult = JSONUtils.toMap(schedule);
+        if(scheduleResult == null){
+            return 0;
+        }
+        if(scheduleResult.containsKey(CMDPARAM_COMPLEMENT_DATA_SCHEDULE_DATE_LIST)){
+            if(scheduleResult.get(CMDPARAM_COMPLEMENT_DATA_SCHEDULE_DATE_LIST) == null){
+                return 0;
+            }
+        }
+        if(scheduleResult.containsKey(CMDPARAM_COMPLEMENT_DATA_START_DATE)){
+            String startDate = scheduleResult.get(CMDPARAM_COMPLEMENT_DATA_START_DATE);
+            String endDate = scheduleResult.get(CMDPARAM_COMPLEMENT_DATA_END_DATE);
+            if (startDate == null || endDate == null) {
+              return 0;
+            }
+            start = DateUtils.getScheduleDate(startDate);
+            end = DateUtils.getScheduleDate(endDate);
+            if(start == null || end == null){
+                return 0;
+            }
+            if (start.after(end)) {
+                logger.error("complement data error, wrong date start:{} and end date:{} ",
+                        start, end
+                );
+                return 0;
+            }
+        }
+        return 1;
+    }
+
+    /**
+     *
+     * @param scheduleTimeList
+     * @return remove duplicate date list
+     */
+    private String removeDuplicates(String scheduleTimeList){
+        HashSet<String> removeDate = new HashSet<String>();
+        List<String> resultList = new ArrayList<String>();
+        if(StringUtils.isNotEmpty(scheduleTimeList)){
+            String[] dateArrays = scheduleTimeList.split(COMMA);
+            List<String> dateList = Arrays.asList(dateArrays);
+            removeDate.addAll(dateList);
+            resultList.addAll(removeDate);
+            return String.join(COMMA, resultList);
+        }
+        return null;
+    }
 }

dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/ExecutorServiceTest.java

@@ -199,7 +199,7 @@ public class ExecutorServiceTest {
 
         Mockito.when(processService.queryReleaseSchedulerListByProcessDefinitionCode(processDefinitionCode)).thenReturn(zeroSchedulerList());
         Map<String, Object> result = executorService.execProcessInstance(loginUser, projectCode,
-                processDefinitionCode, cronTime, CommandType.START_PROCESS,
+                processDefinitionCode, "{\"complementStartDate\":\"2020-01-01 00:00:00\",\"complementEndDate\":\"2020-01-31 23:00:00\"}", CommandType.START_PROCESS,
                 null, null,
                 null, null, 0,
                 RunMode.RUN_MODE_SERIAL,
@@ -218,7 +218,7 @@ public class ExecutorServiceTest {
 
         Mockito.when(processService.queryReleaseSchedulerListByProcessDefinitionCode(processDefinitionCode)).thenReturn(zeroSchedulerList());
         Map<String, Object> result = executorService.execProcessInstance(loginUser, projectCode,
-                processDefinitionCode, cronTime, CommandType.START_PROCESS,
+                processDefinitionCode, "{\"complementStartDate\":\"2020-01-01 00:00:00\",\"complementEndDate\":\"2020-01-31 23:00:00\"}", CommandType.START_PROCESS,
                 null, "n1,n2",
                 null, null, 0,
                 RunMode.RUN_MODE_SERIAL,
@@ -237,7 +237,7 @@ public class ExecutorServiceTest {
 
         Mockito.when(processService.queryReleaseSchedulerListByProcessDefinitionCode(processDefinitionCode)).thenReturn(zeroSchedulerList());
         Map<String, Object> result = executorService.execProcessInstance(loginUser, projectCode,
-                processDefinitionCode, "2020-01-31 23:00:00,2020-01-01 00:00:00", CommandType.COMPLEMENT_DATA,
+                processDefinitionCode, "{\"complementStartDate\":\"2022-01-07 12:12:12\",\"complementEndDate\":\"2022-01-06 12:12:12\"}", CommandType.COMPLEMENT_DATA,
                 null, null,
                 null, null, 0,
                 RunMode.RUN_MODE_SERIAL,
@@ -255,7 +255,7 @@ public class ExecutorServiceTest {
 
         Mockito.when(processService.queryReleaseSchedulerListByProcessDefinitionCode(processDefinitionCode)).thenReturn(zeroSchedulerList());
         Map<String, Object> result = executorService.execProcessInstance(loginUser, projectCode,
-                processDefinitionCode, cronTime, CommandType.COMPLEMENT_DATA,
+                processDefinitionCode, "{\"complementStartDate\":\"2020-01-01 00:00:00\",\"complementEndDate\":\"2020-01-31 23:00:00\"}", CommandType.COMPLEMENT_DATA,
                 null, null,
                 null, null, 0,
                 RunMode.RUN_MODE_SERIAL,
@@ -273,7 +273,7 @@ public class ExecutorServiceTest {
 
         Mockito.when(processService.queryReleaseSchedulerListByProcessDefinitionCode(processDefinitionCode)).thenReturn(zeroSchedulerList());
         Map<String, Object> result = executorService.execProcessInstance(loginUser, projectCode,
-                processDefinitionCode, cronTime, CommandType.COMPLEMENT_DATA,
+                processDefinitionCode, "{\"complementStartDate\":\"2020-01-01 00:00:00\",\"complementEndDate\":\"2020-01-31 23:00:00\"}", CommandType.COMPLEMENT_DATA,
                 null, null,
                 null, null, 0,
                 RunMode.RUN_MODE_PARALLEL,
@@ -292,7 +292,7 @@ public class ExecutorServiceTest {
 
         Mockito.when(processService.queryReleaseSchedulerListByProcessDefinitionCode(processDefinitionCode)).thenReturn(oneSchedulerList());
         Map<String, Object> result = executorService.execProcessInstance(loginUser, projectCode,
-                processDefinitionCode, cronTime, CommandType.COMPLEMENT_DATA,
+                processDefinitionCode, "{\"complementStartDate\":\"2020-01-01 00:00:00\",\"complementEndDate\":\"2020-01-31 23:00:00\"}", CommandType.COMPLEMENT_DATA,
                 null, null,
                 null, null, 0,
                 RunMode.RUN_MODE_PARALLEL,
@@ -308,7 +308,7 @@ public class ExecutorServiceTest {
         Mockito.when(monitorService.getServerListFromRegistry(true)).thenReturn(new ArrayList<>());
 
         Map<String, Object> result = executorService.execProcessInstance(loginUser, projectCode,
-                processDefinitionCode, cronTime, CommandType.COMPLEMENT_DATA,
+                processDefinitionCode, "{\"complementStartDate\":\"2020-01-01 00:00:00\",\"complementEndDate\":\"2020-01-31 23:00:00\"}", CommandType.COMPLEMENT_DATA,
                 null, null,
                 null, null, 0,
                 RunMode.RUN_MODE_PARALLEL,

dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/Constants.java

@@ -361,6 +361,11 @@ public final class Constants {
      */
     public static final String CMDPARAM_COMPLEMENT_DATA_END_DATE = "complementEndDate";
 
+    /**
+     * complement data Schedule date
+     */
+    public static final String CMDPARAM_COMPLEMENT_DATA_SCHEDULE_DATE_LIST = "complementScheduleDateList";
+
     /**
      * complement date default cron string
      */
@@ -809,4 +814,10 @@ public final class Constants {
      * tenant
      */
     public static final int TENANT_FULL_NAME_MAX_LENGTH = 30;
+
+    /**
+     * schedule time  the amount of date data is too large, affecting the memory, so set 100
+     */
+    public static final int SCHEDULE_TIME_MAX_LENGTH = 100;
+
 }

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteRunnable.java

@@ -17,16 +17,10 @@
 
 package org.apache.dolphinscheduler.server.master.runner;
 
-import static org.apache.dolphinscheduler.common.Constants.CMDPARAM_COMPLEMENT_DATA_END_DATE;
-import static org.apache.dolphinscheduler.common.Constants.CMDPARAM_COMPLEMENT_DATA_START_DATE;
-import static org.apache.dolphinscheduler.common.Constants.CMD_PARAM_RECOVERY_START_NODE_STRING;
-import static org.apache.dolphinscheduler.common.Constants.CMD_PARAM_RECOVER_PROCESS_ID_STRING;
-import static org.apache.dolphinscheduler.common.Constants.CMD_PARAM_START_NODES;
-import static org.apache.dolphinscheduler.common.Constants.DEFAULT_WORKER_GROUP;
-import static org.apache.dolphinscheduler.plugin.task.api.TaskConstants.TASK_TYPE_BLOCKING;
-import static org.apache.dolphinscheduler.plugin.task.api.enums.DataType.VARCHAR;
-import static org.apache.dolphinscheduler.plugin.task.api.enums.Direct.IN;
-
+import com.google.common.collect.Lists;
+import org.apache.commons.collections.CollectionUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.math.NumberUtils;
 import org.apache.dolphinscheduler.common.Constants;
 import org.apache.dolphinscheduler.common.enums.CommandType;
 import org.apache.dolphinscheduler.common.enums.FailureStrategy;
@@ -75,10 +69,8 @@ import org.apache.dolphinscheduler.service.alert.ProcessAlertManager;
 import org.apache.dolphinscheduler.service.corn.CronUtils;
 import org.apache.dolphinscheduler.service.process.ProcessService;
 import org.apache.dolphinscheduler.service.queue.PeerTaskInstancePriorityQueue;
-
-import org.apache.commons.collections.CollectionUtils;
-import org.apache.commons.lang3.StringUtils;
-import org.apache.commons.lang3.math.NumberUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -96,10 +88,18 @@ import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.atomic.AtomicBoolean;
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.google.common.collect.Lists;
+import static org.apache.dolphinscheduler.common.Constants.CMDPARAM_COMPLEMENT_DATA_END_DATE;
+import static org.apache.dolphinscheduler.common.Constants.CMDPARAM_COMPLEMENT_DATA_SCHEDULE_DATE_LIST;
+import static org.apache.dolphinscheduler.common.Constants.CMDPARAM_COMPLEMENT_DATA_START_DATE;
+import static org.apache.dolphinscheduler.common.Constants.CMD_PARAM_RECOVERY_START_NODE_STRING;
+import static org.apache.dolphinscheduler.common.Constants.CMD_PARAM_RECOVER_PROCESS_ID_STRING;
+import static org.apache.dolphinscheduler.common.Constants.CMD_PARAM_START_NODES;
+import static org.apache.dolphinscheduler.common.Constants.COMMA;
+import static org.apache.dolphinscheduler.common.Constants.DEFAULT_WORKER_GROUP;
+import static org.apache.dolphinscheduler.common.Constants.YYYY_MM_DD_HH_MM_SS;
+import static org.apache.dolphinscheduler.plugin.task.api.TaskConstants.TASK_TYPE_BLOCKING;
+import static org.apache.dolphinscheduler.plugin.task.api.enums.DataType.VARCHAR;
+import static org.apache.dolphinscheduler.plugin.task.api.enums.Direct.IN;
 
 /**
  * Workflow execute task, used to execute a workflow instance.
@@ -282,14 +282,14 @@ public class WorkflowExecuteRunnable implements Runnable {
 
     public String getKey() {
         if (StringUtils.isNotEmpty(key)
-            || this.processDefinition == null) {
+                || this.processDefinition == null) {
             return key;
         }
 
         key = String.format("%d_%d_%d",
-            this.processDefinition.getCode(),
-            this.processDefinition.getVersion(),
-            this.processInstance.getId());
+                this.processDefinition.getCode(),
+                this.processDefinition.getVersion(),
+                this.processInstance.getId());
         return key;
     }
 
@@ -503,7 +503,7 @@ public class WorkflowExecuteRunnable implements Runnable {
                 } else {
                     ProcessInstance processInstance = this.processService.findProcessInstanceById(nextTaskInstance.getProcessInstanceId());
                     this.processService.sendStartTask2Master(processInstance, nextTaskInstance.getId(),
-                        org.apache.dolphinscheduler.remote.command.CommandType.TASK_WAKEUP_EVENT_REQUEST);
+                            org.apache.dolphinscheduler.remote.command.CommandType.TASK_WAKEUP_EVENT_REQUEST);
                 }
             }
         }
@@ -526,12 +526,12 @@ public class WorkflowExecuteRunnable implements Runnable {
         waitToRetryTaskInstanceMap.put(newTaskInstance.getTaskCode(), newTaskInstance);
         if (!taskInstance.retryTaskIntervalOverTime()) {
             logger.info("failure task will be submitted: process id: {}, task instance code: {} state:{} retry times:{} / {}, interval:{}",
-                processInstance.getId(),
-                newTaskInstance.getTaskCode(),
-                newTaskInstance.getState(),
-                newTaskInstance.getRetryTimes(),
-                newTaskInstance.getMaxRetryTimes(),
-                newTaskInstance.getRetryInterval());
+                    processInstance.getId(),
+                    newTaskInstance.getTaskCode(),
+                    newTaskInstance.getState(),
+                    newTaskInstance.getRetryTimes(),
+                    newTaskInstance.getMaxRetryTimes(),
+                    newTaskInstance.getRetryInterval());
             stateWheelExecuteThread.addTask4TimeoutCheck(processInstance, newTaskInstance);
             stateWheelExecuteThread.addTask4RetryCheck(processInstance, newTaskInstance);
         } else {
@@ -562,7 +562,7 @@ public class WorkflowExecuteRunnable implements Runnable {
         logger.info("process instance update: {}", processInstanceId);
         processInstance = processService.findProcessInstanceById(processInstanceId);
         processDefinition = processService.findProcessDefinition(processInstance.getProcessDefinitionCode(),
-            processInstance.getProcessDefinitionVersion());
+                processInstance.getProcessDefinitionVersion());
         processInstance.setProcessDefinition(processDefinition);
     }
 
@@ -591,8 +591,8 @@ public class WorkflowExecuteRunnable implements Runnable {
     public boolean checkProcessInstance(StateEvent stateEvent) {
         if (this.processInstance.getId() != stateEvent.getProcessInstanceId()) {
             logger.error("mismatch process instance id: {}, state event:{}",
-                this.processInstance.getId(),
-                stateEvent);
+                    this.processInstance.getId(),
+                    stateEvent);
             return false;
         }
         return true;
@@ -774,7 +774,12 @@ public class WorkflowExecuteRunnable implements Runnable {
         if (cmdParam.containsKey(Constants.CMD_PARAM_RECOVERY_START_NODE_STRING)) {
             cmdParam.remove(Constants.CMD_PARAM_RECOVERY_START_NODE_STRING);
         }
-        cmdParam.replace(CMDPARAM_COMPLEMENT_DATA_START_DATE, DateUtils.format(scheduleDate, "yyyy-MM-dd HH:mm:ss", null));
+        if(cmdParam.containsKey(CMDPARAM_COMPLEMENT_DATA_SCHEDULE_DATE_LIST)){
+            cmdParam.replace(CMDPARAM_COMPLEMENT_DATA_SCHEDULE_DATE_LIST, cmdParam.get(CMDPARAM_COMPLEMENT_DATA_SCHEDULE_DATE_LIST).substring(cmdParam.get(CMDPARAM_COMPLEMENT_DATA_SCHEDULE_DATE_LIST).indexOf(COMMA)+1));
+        }
+        if(cmdParam.containsKey(CMDPARAM_COMPLEMENT_DATA_START_DATE)){
+            cmdParam.replace(CMDPARAM_COMPLEMENT_DATA_START_DATE, DateUtils.format(scheduleDate, YYYY_MM_DD_HH_MM_SS, null));
+        }
         command.setCommandParam(JSONUtils.toJsonString(cmdParam));
         command.setTaskDependType(processInstance.getTaskDependType());
         command.setFailureStrategy(processInstance.getFailureStrategy());
@@ -794,7 +799,7 @@ public class WorkflowExecuteRunnable implements Runnable {
 
     private boolean needComplementProcess() {
         if (processInstance.isComplementData()
-            && Flag.NO == processInstance.getIsSubProcess()) {
+                && Flag.NO == processInstance.getIsSubProcess()) {
             return true;
         }
         return false;
@@ -874,7 +879,7 @@ public class WorkflowExecuteRunnable implements Runnable {
             return;
         }
         processDefinition = processService.findProcessDefinition(processInstance.getProcessDefinitionCode(),
-            processInstance.getProcessDefinitionVersion());
+                processInstance.getProcessDefinitionVersion());
         processInstance.setProcessDefinition(processDefinition);
 
         List<TaskInstance> recoverNodeList = getStartTaskInstanceList(processInstance.getCommandParam());
@@ -894,7 +899,7 @@ public class WorkflowExecuteRunnable implements Runnable {
         List<String> recoveryNodeCodeList = getRecoveryNodeCodeList(recoverNodeList);
         List<String> startNodeNameList = parseStartNodeName(processInstance.getCommandParam());
         ProcessDag processDag = generateFlowDag(taskNodeList,
-            startNodeNameList, recoveryNodeCodeList, processInstance.getTaskDependType());
+                startNodeNameList, recoveryNodeCodeList, processInstance.getTaskDependType());
         if (processDag == null) {
             logger.error("processDag is null");
             return;
@@ -956,15 +961,24 @@ public class WorkflowExecuteRunnable implements Runnable {
 
         if (processInstance.isComplementData() && complementListDate.isEmpty()) {
             Map<String, String> cmdParam = JSONUtils.toMap(processInstance.getCommandParam());
-            if (cmdParam != null && cmdParam.containsKey(CMDPARAM_COMPLEMENT_DATA_START_DATE)) {
+            if (cmdParam != null) {
                 // reset global params while there are start parameters
                 setGlobalParamIfCommanded(processDefinition, cmdParam);
 
-                Date start = DateUtils.stringToDate(cmdParam.get(CMDPARAM_COMPLEMENT_DATA_START_DATE));
-                Date end = DateUtils.stringToDate(cmdParam.get(CMDPARAM_COMPLEMENT_DATA_END_DATE));
+                Date start = null;
+                Date end = null;
+                if(cmdParam.containsKey(CMDPARAM_COMPLEMENT_DATA_START_DATE) && cmdParam.containsKey(CMDPARAM_COMPLEMENT_DATA_END_DATE)){
+                    start = DateUtils.stringToDate(cmdParam.get(CMDPARAM_COMPLEMENT_DATA_START_DATE));
+                    end = DateUtils.stringToDate(cmdParam.get(CMDPARAM_COMPLEMENT_DATA_END_DATE));
+                }
                 List<Schedule> schedules = processService.queryReleaseSchedulerListByProcessDefinitionCode(processInstance.getProcessDefinitionCode());
                 if (complementListDate.isEmpty() && needComplementProcess()) {
-                    complementListDate = CronUtils.getSelfFireDateList(start, end, schedules);
+                    if(start != null && end != null){
+                        complementListDate = CronUtils.getSelfFireDateList(start, end, schedules);
+                    }
+                    if(cmdParam.containsKey(CMDPARAM_COMPLEMENT_DATA_SCHEDULE_DATE_LIST)){
+                        complementListDate = CronUtils.getSelfScheduleDateList(cmdParam);
+                    }
                     logger.info(" process definition code:{} complement data: {}",
                             processInstance.getProcessDefinitionCode(), complementListDate);
 
@@ -996,15 +1010,15 @@ public class WorkflowExecuteRunnable implements Runnable {
             taskProcessor.init(taskInstance, processInstance);
 
             if (taskInstance.getState() == ExecutionStatus.RUNNING_EXECUTION
-                && taskProcessor.getType().equalsIgnoreCase(Constants.COMMON_TASK_TYPE)) {
+                    && taskProcessor.getType().equalsIgnoreCase(Constants.COMMON_TASK_TYPE)) {
                 notifyProcessHostUpdate(taskInstance);
             }
 
             boolean submit = taskProcessor.action(TaskAction.SUBMIT);
             if (!submit) {
                 logger.error("process id:{} name:{} submit standby task id:{} name:{} failed!",
-                    processInstance.getId(), processInstance.getName(),
-                    taskInstance.getId(), taskInstance.getName());
+                        processInstance.getId(), processInstance.getName(),
+                        taskInstance.getId(), taskInstance.getName());
                 return null;
             }
 
@@ -1475,10 +1489,10 @@ public class WorkflowExecuteRunnable implements Runnable {
      */
     private ExecutionStatus runningState(ExecutionStatus state) {
         if (state == ExecutionStatus.READY_STOP
-            || state == ExecutionStatus.READY_PAUSE
-            || state == ExecutionStatus.WAITING_THREAD
-            || state == ExecutionStatus.READY_BLOCK
-            || state == ExecutionStatus.DELAY_EXECUTION) {
+                || state == ExecutionStatus.READY_PAUSE
+                || state == ExecutionStatus.WAITING_THREAD
+                || state == ExecutionStatus.READY_BLOCK
+                || state == ExecutionStatus.DELAY_EXECUTION) {
             // if the running task is not completed, the state remains unchanged
             return state;
         } else {
@@ -1514,8 +1528,8 @@ public class WorkflowExecuteRunnable implements Runnable {
             }
             if (processInstance.getFailureStrategy() == FailureStrategy.CONTINUE) {
                 return readyToSubmitTaskQueue.size() == 0
-                    && activeTaskProcessorMaps.size() == 0
-                    && waitToRetryTaskInstanceMap.size() == 0;
+                        && activeTaskProcessorMaps.size() == 0
+                        && waitToRetryTaskInstanceMap.size() == 0;
             }
         }
         return false;
@@ -1546,9 +1560,9 @@ public class WorkflowExecuteRunnable implements Runnable {
 
         List<TaskInstance> pauseList = getCompleteTaskByState(ExecutionStatus.PAUSE);
         if (CollectionUtils.isNotEmpty(pauseList)
-            || processInstance.isBlocked()
-            || !isComplementEnd()
-            || readyToSubmitTaskQueue.size() > 0) {
+                || processInstance.isBlocked()
+                || !isComplementEnd()
+                || readyToSubmitTaskQueue.size() > 0) {
             return ExecutionStatus.PAUSE;
         } else {
             return ExecutionStatus.SUCCESS;
@@ -1613,9 +1627,9 @@ public class WorkflowExecuteRunnable implements Runnable {
             List<TaskInstance> killList = getCompleteTaskByState(ExecutionStatus.KILL);
             List<TaskInstance> failList = getCompleteTaskByState(ExecutionStatus.FAILURE);
             if (CollectionUtils.isNotEmpty(stopList)
-                || CollectionUtils.isNotEmpty(killList)
-                || CollectionUtils.isNotEmpty(failList)
-                || !isComplementEnd()) {
+                    || CollectionUtils.isNotEmpty(killList)
+                    || CollectionUtils.isNotEmpty(failList)
+                    || !isComplementEnd()) {
                 return ExecutionStatus.STOP;
             } else {
                 return ExecutionStatus.SUCCESS;
@@ -1673,10 +1687,10 @@ public class WorkflowExecuteRunnable implements Runnable {
         ExecutionStatus state = getProcessInstanceState(processInstance);
         if (processInstance.getState() != state) {
             logger.info(
-                "work flow process instance [id: {}, name:{}], state change from {} to {}, cmd type: {}",
-                processInstance.getId(), processInstance.getName(),
-                processInstance.getState(), state,
-                processInstance.getCommandType());
+                    "work flow process instance [id: {}, name:{}], state change from {} to {}, cmd type: {}",
+                    processInstance.getId(), processInstance.getName(),
+                    processInstance.getState(), state,
+                    processInstance.getCommandType());
 
             processInstance.setState(state);
             if (state.typeIsFinished()) {
@@ -1701,10 +1715,10 @@ public class WorkflowExecuteRunnable implements Runnable {
         ExecutionStatus state = stateEvent.getExecutionStatus();
         if (processInstance.getState() != state) {
             logger.info(
-                "work flow process instance [id: {}, name:{}], state change from {} to {}, cmd type: {}",
-                processInstance.getId(), processInstance.getName(),
-                processInstance.getState(), state,
-                processInstance.getCommandType());
+                    "work flow process instance [id: {}, name:{}], state change from {} to {}, cmd type: {}",
+                    processInstance.getId(), processInstance.getName(),
+                    processInstance.getState(), state,
+                    processInstance.getCommandType());
 
             processInstance.setState(state);
             if (state.typeIsFinished()) {
@@ -1751,14 +1765,14 @@ public class WorkflowExecuteRunnable implements Runnable {
      */
     private void removeTaskFromStandbyList(TaskInstance taskInstance) {
         logger.info("remove task from stand by list, id: {} name:{}",
-            taskInstance.getId(),
-            taskInstance.getName());
+                taskInstance.getId(),
+                taskInstance.getName());
         try {
             readyToSubmitTaskQueue.remove(taskInstance);
         } catch (Exception e) {
             logger.error("remove task instance from readyToSubmitTaskQueue error, task id:{}, Name: {}",
-                taskInstance.getId(),
-                taskInstance.getName(), e);
+                    taskInstance.getId(),
+                    taskInstance.getName(), e);
         }
     }
 
@@ -1781,7 +1795,7 @@ public class WorkflowExecuteRunnable implements Runnable {
      */
     private void killAllTasks() {
         logger.info("kill called on process instance id: {}, num: {}", processInstance.getId(),
-            activeTaskProcessorMaps.size());
+                activeTaskProcessorMaps.size());
 
         if (readyToSubmitTaskQueue.size() > 0) {
             readyToSubmitTaskQueue.clear();
@@ -2070,4 +2084,4 @@ public class WorkflowExecuteRunnable implements Runnable {
                 break;
         }
     }
-}
+}

dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/corn/CronUtils.java

@@ -17,22 +17,19 @@
 
 package org.apache.dolphinscheduler.service.corn;
 
-import static org.apache.dolphinscheduler.service.corn.CycleFactory.day;
-import static org.apache.dolphinscheduler.service.corn.CycleFactory.hour;
-import static org.apache.dolphinscheduler.service.corn.CycleFactory.min;
-import static org.apache.dolphinscheduler.service.corn.CycleFactory.month;
-import static org.apache.dolphinscheduler.service.corn.CycleFactory.week;
-import static org.apache.dolphinscheduler.service.corn.CycleFactory.year;
-
-import static com.cronutils.model.CronType.QUARTZ;
-
+import com.cronutils.model.Cron;
+import com.cronutils.model.definition.CronDefinitionBuilder;
+import com.cronutils.parser.CronParser;
+import org.apache.commons.collections.CollectionUtils;
 import org.apache.dolphinscheduler.common.Constants;
 import org.apache.dolphinscheduler.common.enums.CycleEnum;
 import org.apache.dolphinscheduler.common.thread.Stopper;
 import org.apache.dolphinscheduler.common.utils.DateUtils;
 import org.apache.dolphinscheduler.dao.entity.Schedule;
-
-import org.apache.commons.collections.CollectionUtils;
+import org.apache.dolphinscheduler.spi.utils.StringUtils;
+import org.quartz.CronExpression;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.text.ParseException;
 import java.util.ArrayList;
@@ -41,14 +38,17 @@ import java.util.Collections;
 import java.util.Date;
 import java.util.GregorianCalendar;
 import java.util.List;
+import java.util.Map;
 
-import org.quartz.CronExpression;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.cronutils.model.Cron;
-import com.cronutils.model.definition.CronDefinitionBuilder;
-import com.cronutils.parser.CronParser;
+import static com.cronutils.model.CronType.QUARTZ;
+import static org.apache.dolphinscheduler.common.Constants.CMDPARAM_COMPLEMENT_DATA_SCHEDULE_DATE_LIST;
+import static org.apache.dolphinscheduler.common.Constants.COMMA;
+import static org.apache.dolphinscheduler.service.corn.CycleFactory.day;
+import static org.apache.dolphinscheduler.service.corn.CycleFactory.hour;
+import static org.apache.dolphinscheduler.service.corn.CycleFactory.min;
+import static org.apache.dolphinscheduler.service.corn.CycleFactory.month;
+import static org.apache.dolphinscheduler.service.corn.CycleFactory.week;
+import static org.apache.dolphinscheduler.service.corn.CycleFactory.year;
 
 /**
  * // todo: this utils is heavy, it rely on quartz and corn-utils.
@@ -283,4 +283,21 @@ public class CronUtils {
         return end.getTime();
     }
 
+    /**
+     * get Schedule Date
+     * @param param
+     * @return  date list
+     */
+    public static List<Date> getSelfScheduleDateList(Map<String, String> param){
+        List<Date> result = new ArrayList<>();
+        String scheduleDates = param.get(CMDPARAM_COMPLEMENT_DATA_SCHEDULE_DATE_LIST);
+        if(StringUtils.isNotEmpty(scheduleDates)){
+            for (String stringDate : scheduleDates.split(COMMA)) {
+                result.add(DateUtils.stringToDate(stringDate));
+            }
+            return result;
+        }
+        return null;
+    }
+
 }