Supplementary data by schedule (#1830)

* Supplementary data by schedule

* fix sonar check bug

* fix code duplicated blocks

* ut

* loop by day

* MasterExecThread test

* test add licene

Co-authored-by: dailidong <dailidong66@gmail.com>

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/ExecutorService.java

@@ -21,6 +21,7 @@ import org.apache.dolphinscheduler.api.enums.ExecuteType;
 import org.apache.dolphinscheduler.api.enums.Status;
 import org.apache.dolphinscheduler.common.Constants;
 import org.apache.dolphinscheduler.common.enums.*;
+import org.apache.dolphinscheduler.common.utils.CollectionUtils;
 import org.apache.dolphinscheduler.common.utils.DateUtils;
 import org.apache.dolphinscheduler.common.utils.JSONUtils;
 import org.apache.dolphinscheduler.common.utils.StringUtils;
@@ -29,6 +30,7 @@ import org.apache.dolphinscheduler.dao.entity.*;
 import org.apache.dolphinscheduler.dao.mapper.ProcessDefinitionMapper;
 import org.apache.dolphinscheduler.dao.mapper.ProcessInstanceMapper;
 import org.apache.dolphinscheduler.dao.mapper.ProjectMapper;
+import org.apache.dolphinscheduler.server.utils.ScheduleUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -499,14 +501,34 @@ public class ExecutorService extends BaseService{
 
         if(commandType == CommandType.COMPLEMENT_DATA){
             runMode = (runMode == null) ? RunMode.RUN_MODE_SERIAL : runMode;
+            if(null != start && null != end && start.before(end)){
                 if(runMode == RunMode.RUN_MODE_SERIAL){
                     cmdParam.put(CMDPARAM_COMPLEMENT_DATA_START_DATE, DateUtils.dateToString(start));
                     cmdParam.put(CMDPARAM_COMPLEMENT_DATA_END_DATE, DateUtils.dateToString(end));
                     command.setCommandParam(JSONUtils.toJson(cmdParam));
                     return processDao.createCommand(command);
                 }else if (runMode == RunMode.RUN_MODE_PARALLEL){
+                    List<Schedule> schedules = processDao.queryReleaseSchedulerListByProcessDefinitionId(processDefineId);
+                    List<Date> listDate = new LinkedList<>();
+                    if(!CollectionUtils.isEmpty(schedules)){
+                        for (Schedule item : schedules) {
+                            List<Date> list = ScheduleUtils.getRecentTriggerTime(item.getCrontab(), start, end);
+                            listDate.addAll(list);
+                        }
+                    }
+                    if(!CollectionUtils.isEmpty(listDate)){
+                        // loop by schedule date
+                        for (Date date : listDate) {
+                            cmdParam.put(CMDPARAM_COMPLEMENT_DATA_START_DATE, DateUtils.dateToString(date));
+                            cmdParam.put(CMDPARAM_COMPLEMENT_DATA_END_DATE, DateUtils.dateToString(date));
+                            command.setCommandParam(JSONUtils.toJson(cmdParam));
+                            processDao.createCommand(command);
+                        }
+                        return listDate.size();
+                    }else{
+                        // loop by day
                         int runCunt = 0;
-                while(!start.after(end)){
+                        while(!start.after(end)) {
                             runCunt += 1;
                             cmdParam.put(CMDPARAM_COMPLEMENT_DATA_START_DATE, DateUtils.dateToString(start));
                             cmdParam.put(CMDPARAM_COMPLEMENT_DATA_END_DATE, DateUtils.dateToString(start));
@@ -516,6 +538,11 @@ public class ExecutorService extends BaseService{
                         }
                         return runCunt;
                     }
+                }
+            }else{
+                logger.error("there is not vaild schedule date for the process definition: id:{},date:{}",
+                        processDefineId, schedule);
+            }
         }else{
             command.setCommandParam(JSONUtils.toJson(cmdParam));
             return processDao.createCommand(command);

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

@@ -0,0 +1,229 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.dolphinscheduler.api.service;
+
+import org.apache.dolphinscheduler.api.enums.Status;
+import org.apache.dolphinscheduler.common.Constants;
+import org.apache.dolphinscheduler.common.enums.CommandType;
+import org.apache.dolphinscheduler.common.enums.Priority;
+import org.apache.dolphinscheduler.common.enums.ReleaseState;
+import org.apache.dolphinscheduler.common.enums.RunMode;
+import org.apache.dolphinscheduler.dao.ProcessDao;
+import org.apache.dolphinscheduler.dao.entity.*;
+import org.apache.dolphinscheduler.dao.mapper.ProcessDefinitionMapper;
+import org.apache.dolphinscheduler.dao.mapper.ProjectMapper;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.junit.MockitoJUnitRunner;
+
+import java.text.ParseException;
+import java.util.*;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.times;
+
+/**
+ * test for ExecutorService
+ */
+@RunWith(MockitoJUnitRunner.Silent.class)
+public class ExecutorService2Test {
+
+    @InjectMocks
+    private ExecutorService executorService;
+
+    @Mock
+    private ProcessDao processDao;
+
+    @Mock
+    private ProcessDefinitionMapper processDefinitionMapper;
+
+    @Mock
+    private ProjectMapper projectMapper;
+
+    @Mock
+    private ProjectService projectService;
+
+    private int processDefinitionId = 1;
+
+    private int tenantId = 1;
+
+    private int userId = 1;
+
+    private ProcessDefinition processDefinition = new ProcessDefinition();
+
+    private User loginUser = new User();
+
+    private String projectName = "projectName";
+
+    private Project project = new Project();
+
+    private String cronTime;
+
+    @Before
+    public void init(){
+        // user
+        loginUser.setId(userId);
+
+        // processDefinition
+        processDefinition.setId(processDefinitionId);
+        processDefinition.setReleaseState(ReleaseState.ONLINE);
+        processDefinition.setTenantId(tenantId);
+        processDefinition.setUserId(userId);
+
+        // project
+        project.setName(projectName);
+
+        // cronRangeTime
+        cronTime = "2020-01-01 00:00:00,2020-01-31 23:00:00";
+
+        // mock
+        Mockito.when(projectMapper.queryByName(projectName)).thenReturn(project);
+        Mockito.when(projectService.checkProjectAndAuth(loginUser, project, projectName)).thenReturn(checkProjectAndAuth());
+        Mockito.when(processDefinitionMapper.selectById(processDefinitionId)).thenReturn(processDefinition);
+        Mockito.when(processDao.getTenantForProcess(tenantId, userId)).thenReturn(new Tenant());
+        Mockito.when(processDao.createCommand(any(Command.class))).thenReturn(1);
+    }
+
+    /**
+     * not complement
+     * @throws ParseException
+     */
+    @Test
+    public void testNoComplement() throws ParseException {
+        try {
+            Mockito.when(processDao.queryReleaseSchedulerListByProcessDefinitionId(processDefinitionId)).thenReturn(zeroSchedulerList());
+            Map<String, Object> result = executorService.execProcessInstance(loginUser, projectName,
+                    processDefinitionId, cronTime, CommandType.START_PROCESS,
+                    null, null,
+                    null, null, 0,
+                    "", "", RunMode.RUN_MODE_SERIAL,
+                    Priority.LOW, 0, 110);
+            Assert.assertEquals(Status.SUCCESS, result.get(Constants.STATUS));
+            verify(processDao, times(1)).createCommand(any(Command.class));
+        }catch (Exception e){
+            Assert.assertTrue(false);
+        }
+    }
+
+    /**
+     * date error
+     * @throws ParseException
+     */
+    @Test
+    public void testDateError() throws ParseException {
+        try {
+            Mockito.when(processDao.queryReleaseSchedulerListByProcessDefinitionId(processDefinitionId)).thenReturn(zeroSchedulerList());
+            Map<String, Object> result = executorService.execProcessInstance(loginUser, projectName,
+                    processDefinitionId, "2020-01-31 23:00:00,2020-01-01 00:00:00", CommandType.COMPLEMENT_DATA,
+                    null, null,
+                    null, null, 0,
+                    "", "", RunMode.RUN_MODE_SERIAL,
+                    Priority.LOW, 0, 110);
+            Assert.assertEquals(Status.START_PROCESS_INSTANCE_ERROR, result.get(Constants.STATUS));
+            verify(processDao, times(0)).createCommand(any(Command.class));
+        }catch (Exception e){
+            Assert.assertTrue(false);
+        }
+    }
+
+    /**
+     * serial
+     * @throws ParseException
+     */
+    @Test
+    public void testSerial() throws ParseException {
+        try {
+            Mockito.when(processDao.queryReleaseSchedulerListByProcessDefinitionId(processDefinitionId)).thenReturn(zeroSchedulerList());
+            Map<String, Object> result = executorService.execProcessInstance(loginUser, projectName,
+                    processDefinitionId, cronTime, CommandType.COMPLEMENT_DATA,
+                    null, null,
+                    null, null, 0,
+                    "", "", RunMode.RUN_MODE_SERIAL,
+                    Priority.LOW, 0, 110);
+            Assert.assertEquals(Status.SUCCESS, result.get(Constants.STATUS));
+            verify(processDao, times(1)).createCommand(any(Command.class));
+        }catch (Exception e){
+            Assert.assertTrue(false);
+        }
+    }
+
+    /**
+     * without schedule
+     * @throws ParseException
+     */
+    @Test
+    public void testParallelWithOutSchedule() throws ParseException {
+        try{
+            Mockito.when(processDao.queryReleaseSchedulerListByProcessDefinitionId(processDefinitionId)).thenReturn(zeroSchedulerList());
+            Map<String, Object> result = executorService.execProcessInstance(loginUser, projectName,
+                    processDefinitionId, cronTime, CommandType.COMPLEMENT_DATA,
+                    null, null,
+                    null, null, 0,
+                    "", "", RunMode.RUN_MODE_PARALLEL,
+                    Priority.LOW, 0, 110);
+            Assert.assertEquals(Status.SUCCESS, result.get(Constants.STATUS));
+            verify(processDao, times(31)).createCommand(any(Command.class));
+        }catch (Exception e){
+            Assert.assertTrue(false);
+        }
+    }
+
+    /**
+     * with schedule
+     * @throws ParseException
+     */
+    @Test
+    public void testParallelWithSchedule() throws ParseException {
+        try{
+            Mockito.when(processDao.queryReleaseSchedulerListByProcessDefinitionId(processDefinitionId)).thenReturn(oneSchedulerList());
+            Map<String, Object> result = executorService.execProcessInstance(loginUser, projectName,
+                    processDefinitionId, cronTime, CommandType.COMPLEMENT_DATA,
+                    null, null,
+                    null, null, 0,
+                    "", "", RunMode.RUN_MODE_PARALLEL,
+                    Priority.LOW, 0, 110);
+            Assert.assertEquals(Status.SUCCESS, result.get(Constants.STATUS));
+            verify(processDao, times(16)).createCommand(any(Command.class));
+        }catch (Exception e){
+            Assert.assertTrue(false);
+        }
+    }
+
+    private List<Schedule> zeroSchedulerList(){
+        return Collections.EMPTY_LIST;
+    }
+
+    private List<Schedule> oneSchedulerList(){
+        List<Schedule> schedulerList = new LinkedList<>();
+        Schedule schedule = new Schedule();
+        schedule.setCrontab("0 0 0 1/2 * ?");
+        schedulerList.add(schedule);
+        return schedulerList;
+    }
+
+    private Map<String, Object> checkProjectAndAuth(){
+        Map<String, Object> result = new HashMap<>();
+        result.put(Constants.STATUS, Status.SUCCESS);
+        return result;
+    }
+}

dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/ProcessDao.java

@@ -1461,6 +1461,15 @@ public class ProcessDao {
         return scheduleMapper.selectById(id);
     }
 
+    /**
+     * query Schedule by processDefinitionId
+     * @param processDefinitionId processDefinitionId
+     * @see Schedule
+     */
+    public List<Schedule> queryReleaseSchedulerListByProcessDefinitionId(int processDefinitionId) {
+        return scheduleMapper.queryReleaseSchedulerListByProcessDefinitionId(processDefinitionId);
+    }
+
     /**
      * query need failover process instance
      * @param host host

dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/ScheduleMapper.java

@@ -60,4 +60,11 @@ public interface ScheduleMapper extends BaseMapper<Schedule> {
      */
     List<Schedule> queryByProcessDefinitionId(@Param("processDefinitionId") int processDefinitionId);
 
+    /**
+     * query schedule list by process definition id
+     * @param processDefinitionId
+     * @return
+     */
+    List<Schedule> queryReleaseSchedulerListByProcessDefinitionId(@Param("processDefinitionId") int processDefinitionId);
+
 }

dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/ScheduleMapper.xml

@@ -55,4 +55,9 @@
         from t_ds_schedules
         where process_definition_id =#{processDefinitionId}
     </select>
+    <select id="queryReleaseSchedulerListByProcessDefinitionId" resultType="org.apache.dolphinscheduler.dao.entity.Schedule">
+        select *
+        from t_ds_schedules
+        where process_definition_id =#{processDefinitionId} and release_state = 1
+    </select>
 </mapper>

dolphinscheduler-server/pom.xml

@@ -111,6 +111,27 @@
 			<artifactId>dolphinscheduler-alert</artifactId>
 		</dependency>
 
+		<dependency>
+			<groupId>org.powermock</groupId>
+			<artifactId>powermock-module-junit4</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.powermock</groupId>
+			<artifactId>powermock-api-mockito2</artifactId>
+			<scope>test</scope>
+			<exclusions>
+				<exclusion>
+					<groupId>org.mockito</groupId>
+					<artifactId>mockito-core</artifactId>
+				</exclusion>
+			</exclusions>
+		</dependency>
+		<dependency>
+			<groupId>org.mockito</groupId>
+			<artifactId>mockito-core</artifactId>
+			<scope>test</scope>
+		</dependency>
     </dependencies>
 
 

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

@@ -17,6 +17,7 @@
 package org.apache.dolphinscheduler.server.master.runner;
 
 import com.alibaba.fastjson.JSONObject;
+import com.google.common.collect.Lists;
 import org.apache.commons.io.FileUtils;
 import org.apache.dolphinscheduler.common.Constants;
 import org.apache.dolphinscheduler.common.enums.*;
@@ -29,10 +30,12 @@ import org.apache.dolphinscheduler.common.thread.ThreadUtils;
 import org.apache.dolphinscheduler.common.utils.*;
 import org.apache.dolphinscheduler.dao.ProcessDao;
 import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
+import org.apache.dolphinscheduler.dao.entity.Schedule;
 import org.apache.dolphinscheduler.dao.entity.TaskInstance;
 import org.apache.dolphinscheduler.dao.utils.DagHelper;
 import org.apache.dolphinscheduler.server.master.config.MasterConfig;
 import org.apache.dolphinscheduler.server.utils.AlertManager;
+import org.apache.dolphinscheduler.server.utils.ScheduleUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -203,11 +206,31 @@ public class MasterExecThread implements Runnable {
         Date startDate = DateUtils.getScheduleDate(cmdParam.get(CMDPARAM_COMPLEMENT_DATA_START_DATE));
         Date endDate = DateUtils.getScheduleDate(cmdParam.get(CMDPARAM_COMPLEMENT_DATA_END_DATE));
         processDao.saveProcessInstance(processInstance);
-        Date scheduleDate = processInstance.getScheduleTime();
 
+        // get schedules
+        int processDefinitionId = processInstance.getProcessDefinitionId();
+        List<Schedule> schedules = processDao.queryReleaseSchedulerListByProcessDefinitionId(processDefinitionId);
+        List<Date> listDate = Lists.newLinkedList();
+        if(!CollectionUtils.isEmpty(schedules)){
+            for (Schedule schedule : schedules) {
+                List<Date> list = ScheduleUtils.getRecentTriggerTime(schedule.getCrontab(), startDate, endDate);
+                listDate.addAll(list);
+            }
+        }
+        // get first fire date
+        Iterator<Date> iterator = null;
+        Date scheduleDate = null;
+        if(!CollectionUtils.isEmpty(listDate)) {
+            iterator = listDate.iterator();
+            scheduleDate = iterator.next();
+            processInstance.setScheduleTime(scheduleDate);
+            processDao.updateProcessInstance(processInstance);
+        }else{
+            scheduleDate = processInstance.getScheduleTime();
             if(scheduleDate == null){
                 scheduleDate = startDate;
             }
+        }
 
         while(Stopper.isRunning()){
             // prepare dag and other info
@@ -232,12 +255,23 @@ public class MasterExecThread implements Runnable {
             }
 
             //  current process instance sucess ，next execute
+            if(null == iterator){
+                // loop by day
                 scheduleDate = DateUtils.getSomeDay(scheduleDate, 1);
                 if(scheduleDate.after(endDate)){
                     // all success
                     logger.info("process {} complement completely!", processInstance.getId());
                     break;
                 }
+            }else{
+                // loop by schedule date
+                if(!iterator.hasNext()){
+                    // all success
+                    logger.info("process {} complement completely!", processInstance.getId());
+                    break;
+                }
+                scheduleDate = iterator.next();
+            }
 
             logger.info("process {} start to complement {} data",
                     processInstance.getId(), DateUtils.dateToString(scheduleDate));

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/utils/ScheduleUtils.java

@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.dolphinscheduler.server.utils;
+
+import org.quartz.impl.triggers.CronTriggerImpl;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.text.ParseException;
+import java.util.Date;
+import java.util.LinkedList;
+import java.util.List;
+
+/**
+ * ScheduleUtils
+ */
+public class ScheduleUtils {
+
+    private static final Logger logger = LoggerFactory.getLogger(ScheduleUtils.class);
+
+    /**
+     * Get the execution time of the time interval
+     * @param cron
+     * @param from
+     * @param to
+     * @return
+     */
+    public static List<Date> getRecentTriggerTime(String cron, Date from, Date to) {
+        return getRecentTriggerTime(cron, Integer.MAX_VALUE, from, to);
+    }
+
+    /**
+     * Get the execution time of the time interval
+     * @param cron
+     * @param size
+     * @param from
+     * @param to
+     * @return
+     */
+    public static List<Date> getRecentTriggerTime(String cron, int size, Date from, Date to) {
+        List list = new LinkedList<Date>();
+        if(to.before(from)){
+            logger.error("schedule date from:{} must before date to:{}!", from, to);
+            return list;
+        }
+        try {
+            CronTriggerImpl trigger = new CronTriggerImpl();
+            trigger.setCronExpression(cron);
+            trigger.setStartTime(from);
+            trigger.setEndTime(to);
+            trigger.computeFirstFireTime(null);
+            for (int i = 0; i < size; i++) {
+                Date schedule = trigger.getNextFireTime();
+                if(null == schedule){
+                    break;
+                }
+                list.add(schedule);
+                trigger.triggered(null);
+            }
+        } catch (ParseException e) {
+            logger.error("cron:{} error:{}", cron, e.getMessage());
+        }
+        return java.util.Collections.unmodifiableList(list);
+    }
+}

dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/master/MasterExecThreadTest.java

@@ -0,0 +1,154 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.dolphinscheduler.server.master;
+
+import com.alibaba.fastjson.JSONObject;
+import org.apache.dolphinscheduler.common.enums.*;
+import org.apache.dolphinscheduler.common.graph.DAG;
+import org.apache.dolphinscheduler.common.utils.DateUtils;
+import org.apache.dolphinscheduler.common.utils.SpringApplicationContext;
+import org.apache.dolphinscheduler.dao.ProcessDao;
+import org.apache.dolphinscheduler.dao.entity.ProcessDefinition;
+import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
+import org.apache.dolphinscheduler.dao.entity.Schedule;
+import org.apache.dolphinscheduler.server.master.config.MasterConfig;
+import org.apache.dolphinscheduler.server.master.runner.MasterExecThread;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mockito;
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+import org.springframework.context.ApplicationContext;
+import java.lang.reflect.Field;
+import java.lang.reflect.Method;
+import java.text.ParseException;
+import java.util.*;
+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.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.powermock.api.mockito.PowerMockito.mock;
+
+/**
+ * test for MasterExecThread
+ */
+@RunWith(PowerMockRunner.class)
+@PrepareForTest({MasterExecThread.class})
+public class MasterExecThreadTest {
+
+    private MasterExecThread masterExecThread;
+
+    private ProcessInstance processInstance;
+
+    private ProcessDao processDao;
+
+    private int processDefinitionId = 1;
+
+    private MasterConfig config;
+
+    private ApplicationContext applicationContext;
+
+    @Before
+    public void init() throws Exception{
+        processDao = mock(ProcessDao.class);
+
+        applicationContext = mock(ApplicationContext.class);
+        config = new MasterConfig();
+        config.setMasterExecTaskNum(1);
+        SpringApplicationContext springApplicationContext = new SpringApplicationContext();
+        springApplicationContext.setApplicationContext(applicationContext);
+        Mockito.when(applicationContext.getBean(MasterConfig.class)).thenReturn(config);
+
+        processInstance = mock(ProcessInstance.class);
+        Mockito.when(processInstance.getProcessDefinitionId()).thenReturn(processDefinitionId);
+        Mockito.when(processInstance.getState()).thenReturn(ExecutionStatus.SUCCESS);
+        Mockito.when(processInstance.getHistoryCmd()).thenReturn(CommandType.COMPLEMENT_DATA.toString());
+        Mockito.when(processInstance.getIsSubProcess()).thenReturn(Flag.NO);
+        Mockito.when(processInstance.getScheduleTime()).thenReturn(DateUtils.stringToDate("2020-01-01 00:00:00"));
+        Map<String, String> cmdParam = new HashMap<>();
+        cmdParam.put(CMDPARAM_COMPLEMENT_DATA_START_DATE, "2020-01-01 00:00:00");
+        cmdParam.put(CMDPARAM_COMPLEMENT_DATA_END_DATE, "2020-01-31 23:00:00");
+        Mockito.when(processInstance.getCommandParam()).thenReturn(JSONObject.toJSONString(cmdParam));
+        ProcessDefinition processDefinition = new ProcessDefinition();
+        processDefinition.setGlobalParamMap(Collections.EMPTY_MAP);
+        processDefinition.setGlobalParamList(Collections.EMPTY_LIST);
+        Mockito.when(processInstance.getProcessDefinition()).thenReturn(processDefinition);
+
+        masterExecThread = PowerMockito.spy(new MasterExecThread(processInstance, processDao));
+        // prepareProcess init dag
+        Field dag = MasterExecThread.class.getDeclaredField("dag");
+        dag.setAccessible(true);
+        dag.set(masterExecThread, new DAG());
+        PowerMockito.doNothing().when(masterExecThread, "executeProcess");
+        PowerMockito.doNothing().when(masterExecThread, "postHandle");
+        PowerMockito.doNothing().when(masterExecThread, "prepareProcess");
+        PowerMockito.doNothing().when(masterExecThread, "runProcess");
+        PowerMockito.doNothing().when(masterExecThread, "endProcess");
+    }
+
+    /**
+     * without schedule
+     * @throws ParseException
+     */
+    @Test
+    public void testParallelWithOutSchedule() throws ParseException {
+        try{
+            Mockito.when(processDao.queryReleaseSchedulerListByProcessDefinitionId(processDefinitionId)).thenReturn(zeroSchedulerList());
+            Method method = MasterExecThread.class.getDeclaredMethod("executeComplementProcess");
+            method.setAccessible(true);
+            method.invoke(masterExecThread);
+            // one create save, and 1-30 for next save, and last day 31 no save
+            verify(processDao, times(31)).saveProcessInstance(processInstance);
+        }catch (Exception e){
+            e.printStackTrace();
+            Assert.assertTrue(false);
+        }
+    }
+
+    /**
+     * with schedule
+     * @throws ParseException
+     */
+    @Test
+    public void testParallelWithSchedule() throws ParseException {
+        try{
+            Mockito.when(processDao.queryReleaseSchedulerListByProcessDefinitionId(processDefinitionId)).thenReturn(oneSchedulerList());
+            Method method = MasterExecThread.class.getDeclaredMethod("executeComplementProcess");
+            method.setAccessible(true);
+            method.invoke(masterExecThread);
+            // one create save, and 15(1 to 31 step 2) for next save, and last day 31 no save
+            verify(processDao, times(16)).saveProcessInstance(processInstance);
+        }catch (Exception e){
+            Assert.assertTrue(false);
+        }
+    }
+
+    private List<Schedule> zeroSchedulerList(){
+        return Collections.EMPTY_LIST;
+    }
+
+    private List<Schedule> oneSchedulerList(){
+        List<Schedule> schedulerList = new LinkedList<>();
+        Schedule schedule = new Schedule();
+        schedule.setCrontab("0 0 0 1/2 * ?");
+        schedulerList.add(schedule);
+        return schedulerList;
+    }
+}

dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/utils/ScheduleUtilsTest.java

@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.dolphinscheduler.server.utils;
+
+import org.apache.dolphinscheduler.common.utils.DateUtils;
+import org.junit.Test;
+import java.util.Date;
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Test ScheduleUtils
+ */
+public class ScheduleUtilsTest {
+
+    /**
+     * Test the getRecentTriggerTime method
+     */
+    @Test
+    public void testGetRecentTriggerTime() {
+        Date from = DateUtils.stringToDate("2020-01-01 00:00:00");
+        Date to = DateUtils.stringToDate("2020-01-31 01:00:00");
+        // test date
+        assertEquals(0, ScheduleUtils.getRecentTriggerTime("0 0 0 * * ? ", to, from).size());
+        // test error cron
+        assertEquals(0, ScheduleUtils.getRecentTriggerTime("0 0 0 * *", from, to).size());
+        // test cron
+        assertEquals(31, ScheduleUtils.getRecentTriggerTime("0 0 0 * * ? ", from, to).size());
+    }
+}

dolphinscheduler-ui/src/js/conf/home/pages/projects/pages/definition/pages/list/_source/start.vue

@@ -141,7 +141,7 @@
       </div>
       <div class="clearfix list">
         <div class="text">
-          {{$t('Date')}}
+          {{$t('Schedule date')}}
         </div>
         <div class="cont">
           <x-datepicker

dolphinscheduler-ui/src/js/module/i18n/locale/en_US.js

@@ -361,6 +361,7 @@ export default {
   'Recipient': 'Recipient',
   'Cc': 'Cc',
   'Whether it is a complement process?': 'Whether it is a complement process?',
+  'Schedule date': 'Schedule date',
   'Mode of execution': 'Mode of execution',
   'Serial execution': 'Serial execution',
   'Parallel execution': 'Parallel execution',

dolphinscheduler-ui/src/js/module/i18n/locale/zh_CN.js

@@ -356,6 +356,7 @@ export default {
   'Recipient': '收件人',
   'Cc': '抄送人',
   'Whether it is a complement process?': '是否补数',
+  'Schedule date': '调度日期',
   'Mode of execution': '执行方式',
   'Serial execution': '串行执行',
   'Parallel execution': '并行执行',

pom.xml

@@ -694,6 +694,7 @@
 						<include>**/api/service/ProcessDefinitionServiceTest.java</include>
 						<include>**/api/service/UdfFuncServiceTest.java</include>
 						<include>**/api/service/ResourcesServiceTest.java</include>
+						<include>**/api/service/ExecutorService2Test.java</include>
 						<include>**/api/service/BaseServiceTest.java</include>
 						<include>**/api/service/BaseDAGServiceTest.java</include>
 						<include>**/alert/utils/ExcelUtilsTest.java</include>
@@ -703,6 +704,8 @@
 						<include>**/server/utils/SparkArgsUtilsTest.java</include>
 						<include>**/server/utils/FlinkArgsUtilsTest.java</include>
 						<include>**/server/utils/ParamUtilsTest.java</include>
+						<include>**/server/utils/ScheduleUtilsTest.java</include>
+						<include>**/server/master/MasterExecThreadTest.java</include>
 						<include>**/dao/mapper/AccessTokenMapperTest.java</include>
 						<include>**/dao/mapper/AlertGroupMapperTest.java</include>
 						<include>**/dao/mapper/AlertMapperTest.java</include>