diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/ExecutorController.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/ExecutorController.java index b131909a76..4df65a991b 100644 --- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/ExecutorController.java +++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/ExecutorController.java @@ -23,6 +23,7 @@ import static org.apache.dolphinscheduler.api.enums.Status.START_PROCESS_INSTANC import org.apache.dolphinscheduler.api.aspect.AccessLogAnnotation; import org.apache.dolphinscheduler.api.enums.ExecuteType; +import org.apache.dolphinscheduler.api.enums.Status; import org.apache.dolphinscheduler.api.exceptions.ApiException; import org.apache.dolphinscheduler.api.service.ExecutorService; import org.apache.dolphinscheduler.api.utils.Result; @@ -36,8 +37,6 @@ import org.apache.dolphinscheduler.common.enums.WarningType; import org.apache.dolphinscheduler.common.utils.JSONUtils; import org.apache.dolphinscheduler.dao.entity.User; -import java.util.Map; - import org.springframework.beans.factory.annotation.Autowired; import org.springframework.http.HttpStatus; import org.springframework.web.bind.annotation.PathVariable; @@ -55,6 +54,13 @@ import io.swagger.annotations.ApiOperation; import io.swagger.annotations.ApiParam; import springfox.documentation.annotations.ApiIgnore; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; + /** * executor controller */ @@ -138,6 +144,100 @@ public class ExecutorController extends BaseController { return returnDataList(result); } + /** + * batch execute process instance + * If any processDefinitionCode cannot be found, the failure information is returned and the status is set to + * failed. The successful task will run normally and will not stop + * + * @param loginUser login user + * @param projectCode project code + * @param processDefinitionCodes process definition codes + * @param scheduleTime schedule time + * @param failureStrategy failure strategy + * @param startNodeList start nodes list + * @param taskDependType task depend type + * @param execType execute type + * @param warningType warning type + * @param warningGroupId warning group id + * @param runMode run mode + * @param processInstancePriority process instance priority + * @param workerGroup worker group + * @param timeout timeout + * @param expectedParallelismNumber the expected parallelism number when execute complement in parallel mode + * @return start process result code + */ + @ApiOperation(value = "batchStartProcessInstance", notes = "BATCH_RUN_PROCESS_INSTANCE_NOTES") + @ApiImplicitParams({ + @ApiImplicitParam(name = "processDefinitionCodes", value = "PROCESS_DEFINITION_CODES", required = true, dataType = "String", example = "1,2,3"), + @ApiImplicitParam(name = "scheduleTime", value = "SCHEDULE_TIME", required = true, dataType = "String"), + @ApiImplicitParam(name = "failureStrategy", value = "FAILURE_STRATEGY", required = true, dataType = "FailureStrategy"), + @ApiImplicitParam(name = "startNodeList", value = "START_NODE_LIST", dataType = "String"), + @ApiImplicitParam(name = "taskDependType", value = "TASK_DEPEND_TYPE", dataType = "TaskDependType"), + @ApiImplicitParam(name = "execType", value = "COMMAND_TYPE", dataType = "CommandType"), + @ApiImplicitParam(name = "warningType", value = "WARNING_TYPE", required = true, dataType = "WarningType"), + @ApiImplicitParam(name = "warningGroupId", value = "WARNING_GROUP_ID", required = true, dataType = "Int", example = "100"), + @ApiImplicitParam(name = "runMode", value = "RUN_MODE", dataType = "RunMode"), + @ApiImplicitParam(name = "processInstancePriority", value = "PROCESS_INSTANCE_PRIORITY", required = true, dataType = "Priority"), + @ApiImplicitParam(name = "workerGroup", value = "WORKER_GROUP", dataType = "String", example = "default"), + @ApiImplicitParam(name = "environmentCode", value = "ENVIRONMENT_CODE", dataType = "Long", example = "-1"), + @ApiImplicitParam(name = "timeout", value = "TIMEOUT", dataType = "Int", example = "100"), + @ApiImplicitParam(name = "expectedParallelismNumber", value = "EXPECTED_PARALLELISM_NUMBER", dataType = "Int", example = "8") + }) + @PostMapping(value = "batch-start-process-instance") + @ResponseStatus(HttpStatus.OK) + @ApiException(START_PROCESS_INSTANCE_ERROR) + @AccessLogAnnotation(ignoreRequestArgs = "loginUser") + public Result batchStartProcessInstance(@ApiIgnore @RequestAttribute(value = Constants.SESSION_USER) User loginUser, + @ApiParam(name = "projectCode", value = "PROJECT_CODE", required = true) @PathVariable long projectCode, + @RequestParam(value = "processDefinitionCodes") String processDefinitionCodes, + @RequestParam(value = "scheduleTime", required = false) String scheduleTime, + @RequestParam(value = "failureStrategy", required = true) FailureStrategy failureStrategy, + @RequestParam(value = "startNodeList", required = false) String startNodeList, + @RequestParam(value = "taskDependType", required = false) TaskDependType taskDependType, + @RequestParam(value = "execType", required = false) CommandType execType, + @RequestParam(value = "warningType", required = true) WarningType warningType, + @RequestParam(value = "warningGroupId", required = false) int warningGroupId, + @RequestParam(value = "runMode", required = false) RunMode runMode, + @RequestParam(value = "processInstancePriority", required = false) Priority processInstancePriority, + @RequestParam(value = "workerGroup", required = false, defaultValue = "default") String workerGroup, + @RequestParam(value = "environmentCode", required = false, defaultValue = "-1") Long environmentCode, + @RequestParam(value = "timeout", required = false) Integer timeout, + @RequestParam(value = "startParams", required = false) String startParams, + @RequestParam(value = "expectedParallelismNumber", required = false) Integer expectedParallelismNumber, + @RequestParam(value = "dryRun", defaultValue = "0", required = false) int dryRun) { + + if (timeout == null) { + timeout = Constants.MAX_TASK_TIMEOUT; + } + + Map startParamMap = null; + if (startParams != null) { + startParamMap = JSONUtils.toMap(startParams); + } + + Map result = new HashMap<>(); + List processDefinitionCodeArray = Arrays.asList(processDefinitionCodes.split(Constants.COMMA)); + List startFailedProcessDefinitionCodeList = new ArrayList<>(); + + processDefinitionCodeArray = processDefinitionCodeArray.stream().distinct().collect(Collectors.toList()); + + for (String strProcessDefinitionCode : processDefinitionCodeArray) { + long processDefinitionCode = Long.parseLong(strProcessDefinitionCode); + result = execService.execProcessInstance(loginUser, projectCode, processDefinitionCode, scheduleTime, execType, failureStrategy, + startNodeList, taskDependType, warningType, warningGroupId, runMode, processInstancePriority, workerGroup, environmentCode, timeout, startParamMap, expectedParallelismNumber, dryRun); + + if (!Status.SUCCESS.equals(result.get(Constants.STATUS))) { + startFailedProcessDefinitionCodeList.add(String.valueOf(processDefinitionCode)); + } + } + + if (!startFailedProcessDefinitionCodeList.isEmpty()) { + putMsg(result, Status.BATCH_START_PROCESS_INSTANCE_ERROR, String.join(Constants.COMMA, startFailedProcessDefinitionCodeList)); + } + + return returnDataList(result); + } + /** * do action to process instance:pause, stop, repeat, recover from pause, recover from stop * 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 413a328081..824d0d81fa 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 @@ -166,7 +166,7 @@ public enum Status { NAME_EXIST(10135, "name {0} already exists", "名称[{0}]已存在"), SAVE_ERROR(10136, "save error", "保存错误"), DELETE_PROJECT_ERROR_DEFINES_NOT_NULL(10137, "please delete the process definitions in project first!", "请先删除全部工作流定义"), - BATCH_DELETE_PROCESS_INSTANCE_BY_IDS_ERROR(10117, "batch delete process instance by ids {0} error", "批量删除工作流实例错误"), + BATCH_DELETE_PROCESS_INSTANCE_BY_IDS_ERROR(10117, "batch delete process instance by ids {0} error", "批量删除工作流实例错误: {0}"), PREVIEW_SCHEDULE_ERROR(10139, "preview schedule error", "预览调度配置错误"), PARSE_TO_CRON_EXPRESSION_ERROR(10140, "parse cron to cron expression error", "解析调度表达式错误"), SCHEDULE_START_TIME_END_TIME_SAME(10141, "The start time must not be the same as the end", "开始时间不能和结束时间一样"), @@ -250,6 +250,7 @@ public enum Status { COUNT_PROCESS_INSTANCE_STATE_ERROR(50012, "count process instance state error", "查询各状态流程实例数错误"), COUNT_PROCESS_DEFINITION_USER_ERROR(50013, "count process definition user error", "查询各用户流程定义数错误"), START_PROCESS_INSTANCE_ERROR(50014, "start process instance error", "运行工作流实例错误"), + BATCH_START_PROCESS_INSTANCE_ERROR(50014, "batch start process instance error: {0}", "批量运行工作流实例错误: {0}"), EXECUTE_PROCESS_INSTANCE_ERROR(50015, "execute process instance error", "操作工作流实例错误"), CHECK_PROCESS_DEFINITION_ERROR(50016, "check process definition error", "工作流定义错误"), QUERY_RECIPIENTS_AND_COPYERS_BY_PROCESS_DEFINITION_ERROR(50017, "query recipients and copyers by process definition error", "查询收件人和抄送人错误"), 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 18eba02dc7..1b4cee852e 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 @@ -571,6 +571,10 @@ public class ExecutorServiceImpl extends BaseServiceImpl implements ExecutorServ Map cmdParam = JSONUtils.toMap(command.getCommandParam()); switch (runMode) { case RUN_MODE_SERIAL: { + if (start.after(end)) { + logger.warn("The startDate {} is later than the endDate {}", start, end); + break; + } 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)); @@ -578,26 +582,45 @@ public class ExecutorServiceImpl extends BaseServiceImpl implements ExecutorServ break; } case RUN_MODE_PARALLEL: { + if (start.after(end)) { + logger.warn("The startDate {} is later than the endDate {}", start, end); + 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) { createCount = Math.min(listDate.size(), expectedParallelismNumber); + if (listDateSize < createCount) { + createCount = listDateSize; + } } logger.info("In parallel mode, current expectedParallelismNumber:{}", createCount); - listDate.addLast(end); - int chunkSize = listDate.size() / 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 = 0; i < createCount; i++) { - int rangeStart = i == 0 ? i : (i * chunkSize); - int rangeEnd = i == createCount - 1 ? listDate.size() - 1 - : rangeStart + chunkSize; + for (int i = 1; i <= createCount; i++) { + int extra = (i <= remainingItems) ? 1 : 0; + int singleCommandItems = (itemsPerCommand + extra); - cmdParam.put(CMDPARAM_COMPLEMENT_DATA_START_DATE, DateUtils.dateToString(listDate.get(rangeStart))); - cmdParam.put(CMDPARAM_COMPLEMENT_DATA_END_DATE, DateUtils.dateToString(listDate.get(rangeEnd))); + 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); } diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ProcessDefinitionServiceImpl.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ProcessDefinitionServiceImpl.java index 9cfb9f9488..ee53c164e6 100644 --- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ProcessDefinitionServiceImpl.java +++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ProcessDefinitionServiceImpl.java @@ -377,7 +377,7 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro processDefinitionNode.put("id", processDefinition.getId()); processDefinitionNode.put("code", processDefinition.getCode()); processDefinitionNode.put("name", processDefinition.getName()); - processDefinitionNode.put("projectCode", processDefinition.getCode()); + processDefinitionNode.put("projectCode", processDefinition.getProjectCode()); arrayNode.add(processDefinitionNode); } result.put(Constants.DATA_LIST, arrayNode); diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ProcessInstanceServiceImpl.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ProcessInstanceServiceImpl.java index 2b98e14578..bec9fd8cf4 100644 --- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ProcessInstanceServiceImpl.java +++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ProcessInstanceServiceImpl.java @@ -629,13 +629,13 @@ public class ProcessInstanceServiceImpl extends BaseServiceImpl implements Proce } ProcessInstance processInstance = processService.findProcessInstanceDetailById(processInstanceId); if (null == processInstance) { - putMsg(result, Status.PROCESS_INSTANCE_NOT_EXIST, processInstanceId); + putMsg(result, Status.PROCESS_INSTANCE_NOT_EXIST, String.valueOf(processInstanceId)); return result; } ProcessDefinition processDefinition = processDefineMapper.queryByCode(processInstance.getProcessDefinitionCode()); if (processDefinition != null && projectCode != processDefinition.getProjectCode()) { - putMsg(result, Status.PROCESS_INSTANCE_NOT_EXIST, processInstanceId); + putMsg(result, Status.PROCESS_INSTANCE_NOT_EXIST, String.valueOf(processInstanceId)); return result; } @@ -649,6 +649,7 @@ public class ProcessInstanceServiceImpl extends BaseServiceImpl implements Proce processService.deleteAllSubWorkProcessByParentId(processInstanceId); processService.deleteWorkProcessMapByParentId(processInstanceId); + processService.deleteWorkTaskInstanceByProcessInstanceId(processInstanceId); if (delete > 0) { putMsg(result, Status.SUCCESS); diff --git a/dolphinscheduler-api/src/main/resources/i18n/messages.properties b/dolphinscheduler-api/src/main/resources/i18n/messages.properties index 7a4282a63c..928e608880 100644 --- a/dolphinscheduler-api/src/main/resources/i18n/messages.properties +++ b/dolphinscheduler-api/src/main/resources/i18n/messages.properties @@ -18,13 +18,14 @@ QUERY_SCHEDULE_LIST_NOTES=query schedule list EXECUTE_PROCESS_TAG=execute process related operation PROCESS_INSTANCE_EXECUTOR_TAG=process instance executor related operation -RUN_PROCESS_INSTANCE_NOTES=run process instance +RUN_PROCESS_INSTANCE_NOTES=run process instance +BATCH_RUN_PROCESS_INSTANCE_NOTES=batch run process instance START_NODE_LIST=start node list(node name) TASK_DEPEND_TYPE=task depend type COMMAND_TYPE=command type RUN_MODE=run mode TIMEOUT=timeout -EXECUTE_ACTION_TO_PROCESS_INSTANCE_NOTES=execute action to process instance +EXECUTE_ACTION_TO_PROCESS_INSTANCE_NOTES=execute action to process instance EXECUTE_TYPE=execute type START_CHECK_PROCESS_DEFINITION_NOTES=start check process definition GET_RECEIVER_CC_NOTES=query receiver cc diff --git a/dolphinscheduler-api/src/main/resources/i18n/messages_en_US.properties b/dolphinscheduler-api/src/main/resources/i18n/messages_en_US.properties index 19dc343c59..7e9a7e62bc 100644 --- a/dolphinscheduler-api/src/main/resources/i18n/messages_en_US.properties +++ b/dolphinscheduler-api/src/main/resources/i18n/messages_en_US.properties @@ -17,7 +17,9 @@ QUERY_SCHEDULE_LIST_NOTES=query schedule list EXECUTE_PROCESS_TAG=execute process related operation PROCESS_INSTANCE_EXECUTOR_TAG=process instance executor related operation -RUN_PROCESS_INSTANCE_NOTES=run process instance +RUN_PROCESS_INSTANCE_NOTES=run process instance +BATCH_RUN_PROCESS_INSTANCE_NOTES=batch run process instance(If any processDefinitionCode cannot be found, the failure\ + \ information is returned and the status is set to failed. The successful task will run normally and will not stop) START_NODE_LIST=start node list(node name) TASK_DEPEND_TYPE=task depend type COMMAND_TYPE=command type diff --git a/dolphinscheduler-api/src/main/resources/i18n/messages_zh_CN.properties b/dolphinscheduler-api/src/main/resources/i18n/messages_zh_CN.properties index a0f1b8d47a..d5df6cd18b 100644 --- a/dolphinscheduler-api/src/main/resources/i18n/messages_zh_CN.properties +++ b/dolphinscheduler-api/src/main/resources/i18n/messages_zh_CN.properties @@ -19,6 +19,7 @@ PROCESS_INSTANCE_EXECUTOR_TAG=流程实例执行相关操作 UI_PLUGINS_TAG=UI插件相关操作 WORK_FLOW_LINEAGE_TAG=工作流血缘相关操作 RUN_PROCESS_INSTANCE_NOTES=运行流程实例 +BATCH_RUN_PROCESS_INSTANCE_NOTES=批量运行流程实例(其中有任意一个processDefinitionCode找不到,则返回失败信息并且状态置为失败,成功的任务会正常运行,不会停止) START_NODE_LIST=开始节点列表(节点name) TASK_DEPEND_TYPE=任务依赖类型 COMMAND_TYPE=指令类型 diff --git a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/ExecutorServiceTest.java b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/ExecutorServiceTest.java index da942fbedf..90fb173017 100644 --- a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/ExecutorServiceTest.java +++ b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/ExecutorServiceTest.java @@ -17,6 +17,8 @@ package org.apache.dolphinscheduler.api.service; +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.ArgumentMatchers.any; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; @@ -32,6 +34,8 @@ 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.common.model.Server; +import org.apache.dolphinscheduler.common.utils.DateUtils; +import org.apache.dolphinscheduler.common.utils.JSONUtils; import org.apache.dolphinscheduler.dao.entity.Command; import org.apache.dolphinscheduler.dao.entity.ProcessDefinition; import org.apache.dolphinscheduler.dao.entity.ProcessInstance; @@ -340,22 +344,35 @@ public class ExecutorServiceTest { listDate.add(1); listDate.add(2); listDate.add(3); + listDate.add(4); + int listDateSize = listDate.size(); int createCount = Math.min(listDate.size(), expectedParallelismNumber); logger.info("In parallel mode, current expectedParallelismNumber:{}", createCount); - listDate.addLast(4); - int chunkSize = listDate.size() / createCount; - for (int i = 0; i < createCount; i++) { - int rangeStart = i == 0 ? i : (i * chunkSize); - int rangeEnd = i == createCount - 1 ? listDate.size() - 1 : rangeStart + chunkSize; - logger.info("rangeStart:{}, rangeEnd:{}",rangeStart, rangeEnd); - result.add(listDate.get(rangeStart) + "," + listDate.get(rangeEnd)); + 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; + } + + logger.info("startDate:{}, endDate:{}", listDate.get(startDateIndex), listDate.get(endDateIndex)); + result.add(listDate.get(startDateIndex) + "," + listDate.get(endDateIndex)); } Assert.assertEquals("0,1", result.get(0)); - Assert.assertEquals("1,2", result.get(1)); - Assert.assertEquals("2,4", result.get(2)); - + Assert.assertEquals("2,3", result.get(1)); + Assert.assertEquals("4,4", result.get(2)); } + } diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/upgrade/ResourceDao.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/upgrade/ResourceDao.java index c9ff149306..49c0e80c48 100644 --- a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/upgrade/ResourceDao.java +++ b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/upgrade/ResourceDao.java @@ -18,7 +18,6 @@ package org.apache.dolphinscheduler.dao.upgrade; import org.apache.dolphinscheduler.common.utils.ConnectionUtils; - import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -32,7 +31,7 @@ import java.util.Map; * resource dao */ public class ResourceDao { - public static final Logger logger = LoggerFactory.getLogger(ProcessDefinitionDao.class); + public static final Logger logger = LoggerFactory.getLogger(ResourceDao.class); /** * list all resources diff --git a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessService.java b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessService.java index 72abed76e3..8215ed31fa 100644 --- a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessService.java +++ b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessService.java @@ -451,6 +451,25 @@ public class ProcessService { } } + /** + * recursive delete all task instance by process instance id + * @param processInstanceId + */ + public void deleteWorkTaskInstanceByProcessInstanceId(int processInstanceId) { + List taskInstanceList = findValidTaskListByProcessId(processInstanceId); + if (CollectionUtils.isEmpty(taskInstanceList)) { + return; + } + + List taskInstanceIdList = new ArrayList<>(); + + for (TaskInstance taskInstance : taskInstanceList) { + taskInstanceIdList.add(taskInstance.getId()); + } + + taskInstanceMapper.deleteBatchIds(taskInstanceIdList); + } + /** * recursive query sub process definition id by parent id. * 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 3e7007a02a..d7847229ed 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 @@ -191,9 +191,9 @@ public class CronUtils { return result; } - // support left closed and right open time interval (startDate <= N < endDate) + // support left closed and right closed time interval (startDate <= N <= endDate) Date from = new Date(startTime.getTime() - Constants.SECOND_TIME_MILLIS); - Date to = new Date(endTime.getTime() - Constants.SECOND_TIME_MILLIS); + Date to = new Date(endTime.getTime() + Constants.SECOND_TIME_MILLIS); List listSchedule = new ArrayList<>(); listSchedule.addAll(schedules); diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-sqoop/src/main/java/org/apache/dolphinscheduler/plugin/task/sqoop/generator/targets/HiveTargetGenerator.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-sqoop/src/main/java/org/apache/dolphinscheduler/plugin/task/sqoop/generator/targets/HiveTargetGenerator.java index 05ba68fe9e..5e768f3850 100644 --- a/dolphinscheduler-task-plugin/dolphinscheduler-task-sqoop/src/main/java/org/apache/dolphinscheduler/plugin/task/sqoop/generator/targets/HiveTargetGenerator.java +++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-sqoop/src/main/java/org/apache/dolphinscheduler/plugin/task/sqoop/generator/targets/HiveTargetGenerator.java @@ -27,6 +27,7 @@ import static org.apache.dolphinscheduler.plugin.task.sqoop.SqoopConstants.HIVE_ import static org.apache.dolphinscheduler.plugin.task.sqoop.SqoopConstants.HIVE_PARTITION_KEY; import static org.apache.dolphinscheduler.plugin.task.sqoop.SqoopConstants.HIVE_PARTITION_VALUE; import static org.apache.dolphinscheduler.plugin.task.sqoop.SqoopConstants.HIVE_TABLE; +import static org.apache.dolphinscheduler.plugin.task.sqoop.SqoopConstants.TARGET_DIR; import static org.apache.dolphinscheduler.spi.task.TaskConstants.SPACE; import org.apache.dolphinscheduler.plugin.task.sqoop.generator.ITargetGenerator; @@ -91,6 +92,11 @@ public class HiveTargetGenerator implements ITargetGenerator { .append(SPACE).append(targetHiveParameter.getHivePartitionValue()); } + if (StringUtils.isNotEmpty(targetHiveParameter.getHiveTargetDir())) { + hiveTargetSb.append(SPACE).append(TARGET_DIR) + .append(SPACE).append(targetHiveParameter.getHiveTargetDir()); + } + } } catch (Exception e) { logger.error(String.format("Sqoop hive target params build failed: [%s]", e.getMessage())); diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-sqoop/src/main/java/org/apache/dolphinscheduler/plugin/task/sqoop/parameter/targets/TargetHiveParameter.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-sqoop/src/main/java/org/apache/dolphinscheduler/plugin/task/sqoop/parameter/targets/TargetHiveParameter.java index 7358de7546..9f2579fb26 100644 --- a/dolphinscheduler-task-plugin/dolphinscheduler-task-sqoop/src/main/java/org/apache/dolphinscheduler/plugin/task/sqoop/parameter/targets/TargetHiveParameter.java +++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-sqoop/src/main/java/org/apache/dolphinscheduler/plugin/task/sqoop/parameter/targets/TargetHiveParameter.java @@ -54,6 +54,10 @@ public class TargetHiveParameter { * hive partition value */ private String hivePartitionValue; + /** + * hive target dir + */ + private String hiveTargetDir; public String getHiveDatabase() { return hiveDatabase; @@ -118,4 +122,12 @@ public class TargetHiveParameter { public void setHivePartitionValue(String hivePartitionValue) { this.hivePartitionValue = hivePartitionValue; } + + public String getHiveTargetDir() { + return hiveTargetDir; + } + + public void setHiveTargetDir(String hiveTargetDir) { + this.hiveTargetDir = hiveTargetDir; + } } diff --git a/dolphinscheduler-ui/src/js/conf/home/pages/dag/_source/dag.vue b/dolphinscheduler-ui/src/js/conf/home/pages/dag/_source/dag.vue index 1489020eef..60729086aa 100644 --- a/dolphinscheduler-ui/src/js/conf/home/pages/dag/_source/dag.vue +++ b/dolphinscheduler-ui/src/js/conf/home/pages/dag/_source/dag.vue @@ -529,7 +529,7 @@ this.$router.push({ name: 'task-instance', query: { - processInstanceId: this.$route.params.code, + processInstanceId: this.instanceId, taskName: taskName } }) diff --git a/dolphinscheduler-ui/src/js/conf/home/pages/dag/_source/formModel/_source/dependentTimeout.vue b/dolphinscheduler-ui/src/js/conf/home/pages/dag/_source/formModel/_source/dependentTimeout.vue index 8d8f13cfee..d4cd8efd2f 100644 --- a/dolphinscheduler-ui/src/js/conf/home/pages/dag/_source/formModel/_source/dependentTimeout.vue +++ b/dolphinscheduler-ui/src/js/conf/home/pages/dag/_source/formModel/_source/dependentTimeout.vue @@ -28,6 +28,7 @@ +
{{$t('Waiting Dependent start')}} @@ -40,25 +41,44 @@
+
+
+ {{$t('Timeout period')}} +
+ +
+
+ {{$t('Check interval')}} +
+
+ +
+
+ +
+
+ {{$t('Timeout strategy')}} +
+
+
+
{{$t('Waiting Dependent complete')}} @@ -78,19 +99,29 @@
+
+
+ {{$t('Timeout period')}} +
+ +
+
+ {{$t('Timeout strategy')}} +
+
+