diff --git a/docs/docs/en/guide/upgrade/incompatible.md b/docs/docs/en/guide/upgrade/incompatible.md index 0f60fcb9986a..026d9b88bbcb 100644 --- a/docs/docs/en/guide/upgrade/incompatible.md +++ b/docs/docs/en/guide/upgrade/incompatible.md @@ -44,3 +44,7 @@ This document records the incompatible updates between each version. You need to * Remove import and export of workflow definition. ([#17940])(https://github.com/apache/dolphinscheduler/issues/17940) +## 3.5.0 + +* Add the `missed_fire_policy` column to `t_ds_schedules`. Existing schedules default to `FIRE_ALL_MISSED` to preserve the previous Quartz `IgnoreMisfires` behavior. ([#18464](https://github.com/apache/dolphinscheduler/pull/18464)) + diff --git a/docs/docs/zh/guide/upgrade/incompatible.md b/docs/docs/zh/guide/upgrade/incompatible.md index e6309beac9d1..db0ea2d4c5f1 100644 --- a/docs/docs/zh/guide/upgrade/incompatible.md +++ b/docs/docs/zh/guide/upgrade/incompatible.md @@ -44,3 +44,7 @@ * 移除导入导出工作流([#17940])(https://github.com/apache/dolphinscheduler/issues/17940) +## 3.5.0 + +* 为 `t_ds_schedules` 表新增 `missed_fire_policy` 字段。现有定时默认使用 `FIRE_ALL_MISSED`，以保持原有 Quartz `IgnoreMisfires` 行为。([#18464](https://github.com/apache/dolphinscheduler/pull/18464)) + diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/dto/ScheduleParam.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/dto/ScheduleParam.java index 88fd4ea9664b..7d70c77b9c3a 100644 --- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/dto/ScheduleParam.java +++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/dto/ScheduleParam.java @@ -17,10 +17,14 @@ package org.apache.dolphinscheduler.api.dto; +import org.apache.dolphinscheduler.common.enums.ScheduleMissedFirePolicy; + import java.util.Date; import lombok.Data; +import com.fasterxml.jackson.annotation.JsonIgnore; + /** * schedule parameters */ @@ -31,6 +35,10 @@ public class ScheduleParam { private Date endTime; private String crontab; private String timezoneId; + private ScheduleMissedFirePolicy missedFirePolicy = ScheduleMissedFirePolicy.FIRE_ALL_MISSED; + + @JsonIgnore + private boolean missedFirePolicySet; public ScheduleParam() { } @@ -42,6 +50,15 @@ public ScheduleParam(Date startTime, Date endTime, String timezoneId, String cro this.crontab = crontab; } + public void setMissedFirePolicy(ScheduleMissedFirePolicy missedFirePolicy) { + this.missedFirePolicy = missedFirePolicy; + this.missedFirePolicySet = true; + } + + public boolean isMissedFirePolicySet() { + return missedFirePolicySet; + } + @Override public String toString() { return "ScheduleParam{" diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/SchedulerServiceImpl.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/SchedulerServiceImpl.java index 12df89e9fccf..7b0ae201c947 100644 --- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/SchedulerServiceImpl.java +++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/SchedulerServiceImpl.java @@ -172,6 +172,8 @@ public Schedule insertSchedule(User loginUser, throw new ServiceException(Status.REQUEST_PARAMS_NOT_VALID_ERROR, scheduleParam.getCrontab()); } scheduleObj.setCrontab(scheduleParam.getCrontab()); + validateMissedFirePolicy(scheduleParam); + scheduleObj.setMissedFirePolicy(scheduleParam.getMissedFirePolicy()); scheduleObj.setTimezoneId(scheduleParam.getTimezoneId()); scheduleObj.setWarningType(warningType); scheduleObj.setWarningGroupId(warningGroupId); @@ -557,6 +559,10 @@ private Schedule updateSchedule(Schedule schedule, WorkflowDefinition workflowDe throw new ServiceException(Status.SCHEDULE_CRON_CHECK_FAILED, scheduleParam.getCrontab()); } schedule.setCrontab(scheduleParam.getCrontab()); + validateMissedFirePolicy(scheduleParam); + if (scheduleParam.isMissedFirePolicySet() && scheduleParam.getMissedFirePolicy() != null) { + schedule.setMissedFirePolicy(scheduleParam.getMissedFirePolicy()); + } schedule.setTimezoneId(scheduleParam.getTimezoneId()); } @@ -585,4 +591,11 @@ private Schedule updateSchedule(Schedule schedule, WorkflowDefinition workflowDe return schedule; } + private void validateMissedFirePolicy(ScheduleParam scheduleParam) { + if (scheduleParam.isMissedFirePolicySet() && scheduleParam.getMissedFirePolicy() == null) { + log.warn("Schedule missed fire policy is invalid."); + throw new ServiceException(Status.REQUEST_PARAMS_NOT_VALID_ERROR, "missedFirePolicy"); + } + } + } diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/vo/ScheduleVO.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/vo/ScheduleVO.java index fc4b1d966859..8ddf9c3ade64 100644 --- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/vo/ScheduleVO.java +++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/vo/ScheduleVO.java @@ -20,6 +20,7 @@ import org.apache.dolphinscheduler.common.enums.FailureStrategy; import org.apache.dolphinscheduler.common.enums.Priority; import org.apache.dolphinscheduler.common.enums.ReleaseState; +import org.apache.dolphinscheduler.common.enums.ScheduleMissedFirePolicy; import org.apache.dolphinscheduler.common.enums.WarningType; import org.apache.dolphinscheduler.common.utils.DateUtils; import org.apache.dolphinscheduler.dao.entity.Schedule; @@ -54,6 +55,8 @@ public class ScheduleVO { private String crontab; + private ScheduleMissedFirePolicy missedFirePolicy; + private FailureStrategy failureStrategy; private WarningType warningType; @@ -83,6 +86,7 @@ public class ScheduleVO { public ScheduleVO(Schedule schedule) { this.setId(schedule.getId()); this.setCrontab(schedule.getCrontab()); + this.setMissedFirePolicy(schedule.getMissedFirePolicy()); this.setProjectName(schedule.getProjectName()); this.setUserName(schedule.getUserName()); this.setWorkerGroup(schedule.getWorkerGroup()); diff --git a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/SchedulerServiceTest.java b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/SchedulerServiceTest.java index 20ea9e3ea4a1..3c9a9dedc696 100644 --- a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/SchedulerServiceTest.java +++ b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/SchedulerServiceTest.java @@ -17,10 +17,17 @@ package org.apache.dolphinscheduler.api.service; +import org.apache.dolphinscheduler.api.dto.ScheduleParam; import org.apache.dolphinscheduler.api.enums.Status; import org.apache.dolphinscheduler.api.exceptions.ServiceException; import org.apache.dolphinscheduler.api.service.impl.SchedulerServiceImpl; +import org.apache.dolphinscheduler.api.validator.TenantExistValidator; +import org.apache.dolphinscheduler.common.enums.FailureStrategy; +import org.apache.dolphinscheduler.common.enums.Priority; import org.apache.dolphinscheduler.common.enums.ReleaseState; +import org.apache.dolphinscheduler.common.enums.ScheduleMissedFirePolicy; +import org.apache.dolphinscheduler.common.enums.WarningType; +import org.apache.dolphinscheduler.common.utils.JSONUtils; import org.apache.dolphinscheduler.dao.entity.Project; import org.apache.dolphinscheduler.dao.entity.Schedule; import org.apache.dolphinscheduler.dao.entity.User; @@ -28,6 +35,7 @@ import org.apache.dolphinscheduler.dao.repository.ProjectDao; import org.apache.dolphinscheduler.dao.repository.ScheduleDao; import org.apache.dolphinscheduler.dao.repository.WorkflowDefinitionDao; +import org.apache.dolphinscheduler.scheduler.api.SchedulerApi; import java.util.Optional; @@ -35,6 +43,9 @@ import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.EnumSource; +import org.mockito.ArgumentCaptor; import org.mockito.InjectMocks; import org.mockito.Mock; import org.mockito.Mockito; @@ -61,6 +72,15 @@ public class SchedulerServiceTest extends BaseServiceTestTool { @Mock private ProjectService projectService; + @Mock + private ExecutorService executorService; + + @Mock + private TenantExistValidator tenantExistValidator; + + @Mock + private SchedulerApi schedulerApi; + protected static User user; protected Exception exception; private static final String userName = "userName"; @@ -80,6 +100,173 @@ public void setUp() { user.setId(userId); } + @Test + public void testScheduleParamMissedFirePolicyPresence() { + String scheduleWithoutPolicy = "{\"startTime\":\"2019-12-16 00:00:00\"," + + "\"endTime\":\"2019-12-17 00:00:00\",\"crontab\":\"0 0 6 * * ? *\"}"; + String scheduleWithPolicy = "{\"startTime\":\"2019-12-16 00:00:00\"," + + "\"endTime\":\"2019-12-17 00:00:00\",\"crontab\":\"0 0 6 * * ? *\"," + + "\"missedFirePolicy\":\"SKIP_MISSED\"}"; + + ScheduleParam withoutPolicy = JSONUtils.parseObject(scheduleWithoutPolicy, ScheduleParam.class); + ScheduleParam withPolicy = JSONUtils.parseObject(scheduleWithPolicy, ScheduleParam.class); + + Assertions.assertEquals(ScheduleMissedFirePolicy.FIRE_ALL_MISSED, withoutPolicy.getMissedFirePolicy()); + Assertions.assertFalse(withoutPolicy.isMissedFirePolicySet()); + Assertions.assertEquals(ScheduleMissedFirePolicy.SKIP_MISSED, withPolicy.getMissedFirePolicy()); + Assertions.assertTrue(withPolicy.isMissedFirePolicySet()); + } + + @ParameterizedTest + @EnumSource(ScheduleMissedFirePolicy.class) + public void testInsertScheduleWithMissedFirePolicy(ScheduleMissedFirePolicy missedFirePolicy) { + Project project = this.getProject(); + WorkflowDefinition workflowDefinition = this.getProcessDefinition(); + Schedule insertedSchedule = new Schedule(); + insertedSchedule.setId(scheduleId); + Mockito.when(projectDao.queryByCode(projectCode)).thenReturn(project); + Mockito.when(scheduleDao.queryByWorkflowDefinitionCode(processDefinitionCode)).thenReturn(null); + Mockito.when(workflowDefinitionDao.queryByCode(processDefinitionCode)) + .thenReturn(Optional.of(workflowDefinition)); + Mockito.when(scheduleDao.queryById(Mockito.any())).thenReturn(insertedSchedule); + + Schedule result = schedulerService.insertSchedule( + user, projectCode, processDefinitionCode, scheduleExpression(missedFirePolicy), WarningType.NONE, 0, + FailureStrategy.CONTINUE, Priority.MEDIUM, "default", "tenantCode", environmentCode); + + ArgumentCaptor scheduleCaptor = ArgumentCaptor.forClass(Schedule.class); + Mockito.verify(scheduleDao).insert(scheduleCaptor.capture()); + Assertions.assertEquals(missedFirePolicy, scheduleCaptor.getValue().getMissedFirePolicy()); + Assertions.assertSame(insertedSchedule, result); + } + + @Test + public void testInsertScheduleDefaultsMissedFirePolicy() { + Mockito.when(projectDao.queryByCode(projectCode)).thenReturn(this.getProject()); + Mockito.when(scheduleDao.queryByWorkflowDefinitionCode(processDefinitionCode)).thenReturn(null); + Mockito.when(workflowDefinitionDao.queryByCode(processDefinitionCode)) + .thenReturn(Optional.of(this.getProcessDefinition())); + Mockito.when(scheduleDao.queryById(Mockito.anyInt())).thenReturn(new Schedule()); + + schedulerService.insertSchedule( + user, projectCode, processDefinitionCode, scheduleExpression(null), WarningType.NONE, 0, + FailureStrategy.CONTINUE, Priority.MEDIUM, "default", "tenantCode", environmentCode); + + ArgumentCaptor scheduleCaptor = ArgumentCaptor.forClass(Schedule.class); + Mockito.verify(scheduleDao).insert(scheduleCaptor.capture()); + Assertions.assertEquals(ScheduleMissedFirePolicy.FIRE_ALL_MISSED, + scheduleCaptor.getValue().getMissedFirePolicy()); + } + + @Test + public void testInsertScheduleRejectsExplicitNullMissedFirePolicy() { + assertInsertScheduleRejectsInvalidMissedFirePolicy("null"); + } + + @Test + public void testInsertScheduleRejectsUnknownMissedFirePolicy() { + assertInsertScheduleRejectsInvalidMissedFirePolicy("\"FIRE_ONCE_NWO\""); + } + + @ParameterizedTest + @EnumSource(ScheduleMissedFirePolicy.class) + public void testUpdateScheduleWithMissedFirePolicy(ScheduleMissedFirePolicy missedFirePolicy) { + Schedule schedule = this.getSchedule(); + schedule.setReleaseState(ReleaseState.OFFLINE); + WorkflowDefinition workflowDefinition = this.getProcessDefinition(); + Mockito.when(projectDao.queryByCode(projectCode)).thenReturn(this.getProject()); + Mockito.when(scheduleDao.queryById(scheduleId)).thenReturn(schedule); + Mockito.when(workflowDefinitionDao.queryByCode(processDefinitionCode)) + .thenReturn(Optional.of(workflowDefinition)); + + schedulerService.updateSchedule( + user, projectCode, scheduleId, scheduleExpression(missedFirePolicy), WarningType.NONE, 0, + FailureStrategy.CONTINUE, Priority.MEDIUM, "default", "tenantCode", environmentCode); + + Assertions.assertEquals(missedFirePolicy, schedule.getMissedFirePolicy()); + } + + @Test + public void testUpdateSchedulePreservesMissedFirePolicyWhenOmitted() { + Schedule schedule = this.getSchedule(); + schedule.setReleaseState(ReleaseState.OFFLINE); + schedule.setMissedFirePolicy(ScheduleMissedFirePolicy.SKIP_MISSED); + WorkflowDefinition workflowDefinition = this.getProcessDefinition(); + Mockito.when(projectDao.queryByCode(projectCode)).thenReturn(this.getProject()); + Mockito.when(scheduleDao.queryById(scheduleId)).thenReturn(schedule); + Mockito.when(workflowDefinitionDao.queryByCode(processDefinitionCode)) + .thenReturn(Optional.of(workflowDefinition)); + + schedulerService.updateSchedule( + user, projectCode, scheduleId, scheduleExpression(null), WarningType.NONE, 0, + FailureStrategy.CONTINUE, Priority.MEDIUM, "default", "tenantCode", environmentCode); + + Assertions.assertEquals(ScheduleMissedFirePolicy.SKIP_MISSED, schedule.getMissedFirePolicy()); + } + + @Test + public void testUpdateScheduleRejectsExplicitNullMissedFirePolicy() { + assertUpdateScheduleRejectsInvalidMissedFirePolicy("null"); + } + + @Test + public void testUpdateScheduleRejectsUnknownMissedFirePolicy() { + assertUpdateScheduleRejectsInvalidMissedFirePolicy("\"FIRE_ONCE_NWO\""); + } + + private void assertInsertScheduleRejectsInvalidMissedFirePolicy(String missedFirePolicyValue) { + Mockito.when(projectDao.queryByCode(projectCode)).thenReturn(this.getProject()); + Mockito.when(scheduleDao.queryByWorkflowDefinitionCode(processDefinitionCode)).thenReturn(null); + Mockito.when(workflowDefinitionDao.queryByCode(processDefinitionCode)) + .thenReturn(Optional.of(this.getProcessDefinition())); + + exception = Assertions.assertThrows(ServiceException.class, + () -> schedulerService.insertSchedule( + user, projectCode, processDefinitionCode, + scheduleExpressionWithPolicyValue(missedFirePolicyValue), + WarningType.NONE, 0, FailureStrategy.CONTINUE, Priority.MEDIUM, "default", "tenantCode", + environmentCode)); + + Assertions.assertEquals(Status.REQUEST_PARAMS_NOT_VALID_ERROR.getCode(), + ((ServiceException) exception).getCode()); + Mockito.verify(scheduleDao, Mockito.never()).insert(Mockito.any()); + } + + private void assertUpdateScheduleRejectsInvalidMissedFirePolicy(String missedFirePolicyValue) { + Schedule schedule = this.getSchedule(); + schedule.setReleaseState(ReleaseState.OFFLINE); + schedule.setMissedFirePolicy(ScheduleMissedFirePolicy.SKIP_MISSED); + Mockito.when(projectDao.queryByCode(projectCode)).thenReturn(this.getProject()); + Mockito.when(scheduleDao.queryById(scheduleId)).thenReturn(schedule); + Mockito.when(workflowDefinitionDao.queryByCode(processDefinitionCode)) + .thenReturn(Optional.of(this.getProcessDefinition())); + + exception = Assertions.assertThrows(ServiceException.class, + () -> schedulerService.updateSchedule( + user, projectCode, scheduleId, scheduleExpressionWithPolicyValue(missedFirePolicyValue), + WarningType.NONE, 0, FailureStrategy.CONTINUE, Priority.MEDIUM, "default", "tenantCode", + environmentCode)); + + Assertions.assertEquals(Status.REQUEST_PARAMS_NOT_VALID_ERROR.getCode(), + ((ServiceException) exception).getCode()); + Assertions.assertEquals(ScheduleMissedFirePolicy.SKIP_MISSED, schedule.getMissedFirePolicy()); + Mockito.verify(scheduleDao, Mockito.never()).updateById(Mockito.any()); + } + + private String scheduleExpression(ScheduleMissedFirePolicy missedFirePolicy) { + String policy = missedFirePolicy == null ? "" : ",\"missedFirePolicy\":\"" + missedFirePolicy.name() + "\""; + return scheduleExpressionWithPolicy(policy); + } + + private String scheduleExpressionWithPolicyValue(String missedFirePolicyValue) { + return scheduleExpressionWithPolicy(",\"missedFirePolicy\":" + missedFirePolicyValue); + } + + private String scheduleExpressionWithPolicy(String policy) { + return "{\"startTime\":\"2019-12-16 00:00:00\",\"endTime\":\"2019-12-17 00:00:00\"," + + "\"crontab\":\"0 0 6 * * ? *\",\"timezoneId\":\"Asia/Shanghai\"" + policy + "}"; + } + @Test public void testDeleteSchedules() { Schedule schedule = this.getSchedule(); diff --git a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/ScheduleMissedFirePolicy.java b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/ScheduleMissedFirePolicy.java new file mode 100644 index 000000000000..7959d636191e --- /dev/null +++ b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/ScheduleMissedFirePolicy.java @@ -0,0 +1,37 @@ +/* + * 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.common.enums; + +import lombok.Getter; + +import com.baomidou.mybatisplus.annotation.EnumValue; + +@Getter +public enum ScheduleMissedFirePolicy { + + SKIP_MISSED(0), + FIRE_ONCE_NOW(1), + FIRE_ALL_MISSED(2); + + @EnumValue + private final int code; + + ScheduleMissedFirePolicy(int code) { + this.code = code; + } +} diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/Schedule.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/Schedule.java index a55c8d1ad529..965a5c919477 100644 --- a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/Schedule.java +++ b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/Schedule.java @@ -20,6 +20,7 @@ import org.apache.dolphinscheduler.common.enums.FailureStrategy; import org.apache.dolphinscheduler.common.enums.Priority; import org.apache.dolphinscheduler.common.enums.ReleaseState; +import org.apache.dolphinscheduler.common.enums.ScheduleMissedFirePolicy; import org.apache.dolphinscheduler.common.enums.WarningType; import java.util.Date; @@ -67,6 +68,8 @@ public class Schedule { private String crontab; + private ScheduleMissedFirePolicy missedFirePolicy; + private FailureStrategy failureStrategy; private WarningType warningType; diff --git a/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/ScheduleMapper.xml b/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/ScheduleMapper.xml index 99c1a59ef6eb..78a401c9d1b2 100644 --- a/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/ScheduleMapper.xml +++ b/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/ScheduleMapper.xml @@ -19,12 +19,12 @@ - id, workflow_definition_code, start_time, end_time, timezone_id, crontab, failure_strategy, user_id, release_state, + id, workflow_definition_code, start_time, end_time, timezone_id, crontab, missed_fire_policy, failure_strategy, user_id, release_state, warning_type, warning_group_id, workflow_instance_priority, worker_group, tenant_code, environment_code, create_time, update_time ${alias}.id, ${alias}.workflow_definition_code, ${alias}.start_time, ${alias}.end_time, ${alias}.timezone_id, - ${alias}.crontab, ${alias}.failure_strategy, ${alias}.user_id, ${alias}.release_state, ${alias}.warning_type, + ${alias}.crontab, ${alias}.missed_fire_policy, ${alias}.failure_strategy, ${alias}.user_id, ${alias}.release_state, ${alias}.warning_type, ${alias}.warning_group_id, ${alias}.workflow_instance_priority, ${alias}.worker_group, ${alias}.tenant_code, ${alias}.environment_code, ${alias}.create_time, ${alias}.update_time diff --git a/dolphinscheduler-dao/src/main/resources/sql/dolphinscheduler_h2.sql b/dolphinscheduler-dao/src/main/resources/sql/dolphinscheduler_h2.sql index 1725b5c2df36..d69728be6ef3 100644 --- a/dolphinscheduler-dao/src/main/resources/sql/dolphinscheduler_h2.sql +++ b/dolphinscheduler-dao/src/main/resources/sql/dolphinscheduler_h2.sql @@ -858,6 +858,7 @@ CREATE TABLE t_ds_schedules end_time datetime NOT NULL, timezone_id varchar(40) DEFAULT NULL, crontab varchar(255) NOT NULL, + missed_fire_policy tinyint NOT NULL DEFAULT 2, failure_strategy tinyint(4) NOT NULL, user_id int(11) NOT NULL, release_state tinyint(4) NOT NULL, diff --git a/dolphinscheduler-dao/src/main/resources/sql/dolphinscheduler_mysql.sql b/dolphinscheduler-dao/src/main/resources/sql/dolphinscheduler_mysql.sql index f6dfa61fc122..361cdc3ff7fd 100644 --- a/dolphinscheduler-dao/src/main/resources/sql/dolphinscheduler_mysql.sql +++ b/dolphinscheduler-dao/src/main/resources/sql/dolphinscheduler_mysql.sql @@ -859,6 +859,7 @@ CREATE TABLE `t_ds_schedules` ( `end_time` datetime NOT NULL COMMENT 'end time', `timezone_id` varchar(40) DEFAULT NULL COMMENT 'schedule timezone id', `crontab` varchar(255) NOT NULL COMMENT 'crontab description', + `missed_fire_policy` tinyint NOT NULL DEFAULT '2' COMMENT 'missed fire policy: 0 skip missed, 1 fire once now, 2 fire all missed', `failure_strategy` tinyint(4) NOT NULL COMMENT 'failure strategy. 0:end,1:continue', `user_id` int(11) NOT NULL COMMENT 'user id', `release_state` tinyint(4) NOT NULL COMMENT 'release state. 0:offline,1:online ', diff --git a/dolphinscheduler-dao/src/main/resources/sql/dolphinscheduler_postgresql.sql b/dolphinscheduler-dao/src/main/resources/sql/dolphinscheduler_postgresql.sql index 698b35768d94..387716298493 100644 --- a/dolphinscheduler-dao/src/main/resources/sql/dolphinscheduler_postgresql.sql +++ b/dolphinscheduler-dao/src/main/resources/sql/dolphinscheduler_postgresql.sql @@ -785,6 +785,7 @@ CREATE TABLE t_ds_schedules ( end_time timestamp NOT NULL , timezone_id varchar(40) default NULL , crontab varchar(255) NOT NULL , + missed_fire_policy smallint NOT NULL DEFAULT 2, failure_strategy int NOT NULL , user_id int NOT NULL , release_state int NOT NULL , diff --git a/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/mysql/dolphinscheduler_ddl.sql b/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/mysql/dolphinscheduler_ddl.sql index 3f7d3175979f..afeb9917b86b 100644 --- a/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/mysql/dolphinscheduler_ddl.sql +++ b/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/mysql/dolphinscheduler_ddl.sql @@ -17,4 +17,6 @@ ALTER TABLE `t_ds_task_instance` ADD INDEX idx_project_submit_time (project_code ASC, submit_time DESC); ALTER TABLE `t_ds_workflow_instance` ADD INDEX idx_project_start_time (project_code ASC, start_time DESC); +ALTER TABLE `t_ds_schedules` + ADD COLUMN `missed_fire_policy` tinyint NOT NULL DEFAULT '2' COMMENT 'missed fire policy: 0 skip missed, 1 fire once now, 2 fire all missed' AFTER `crontab`; diff --git a/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/postgresql/dolphinscheduler_ddl.sql b/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/postgresql/dolphinscheduler_ddl.sql index 61a5ae809a13..1709fe57e737 100644 --- a/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/postgresql/dolphinscheduler_ddl.sql +++ b/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/postgresql/dolphinscheduler_ddl.sql @@ -17,3 +17,5 @@ CREATE INDEX idx_project_submit_time ON t_ds_task_instance (project_code ASC, submit_time DESC); CREATE INDEX idx_project_start_time ON t_ds_workflow_instance (project_code ASC, start_time DESC); +ALTER TABLE t_ds_schedules + ADD COLUMN missed_fire_policy smallint NOT NULL DEFAULT 2; diff --git a/dolphinscheduler-scheduler-plugin/dolphinscheduler-scheduler-quartz/src/main/java/org/apache/dolphinscheduler/scheduler/quartz/CronScheduleBuilderFactory.java b/dolphinscheduler-scheduler-plugin/dolphinscheduler-scheduler-quartz/src/main/java/org/apache/dolphinscheduler/scheduler/quartz/CronScheduleBuilderFactory.java new file mode 100644 index 000000000000..4702995a973a --- /dev/null +++ b/dolphinscheduler-scheduler-plugin/dolphinscheduler-scheduler-quartz/src/main/java/org/apache/dolphinscheduler/scheduler/quartz/CronScheduleBuilderFactory.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.scheduler.quartz; + +import org.apache.dolphinscheduler.common.enums.ScheduleMissedFirePolicy; +import org.apache.dolphinscheduler.dao.entity.Schedule; + +import org.quartz.CronScheduleBuilder; + +interface CronScheduleBuilderFactory { + + CronScheduleBuilder createCronScheduleBuilder(Schedule schedule); + + static CronScheduleBuilderFactory getFactory(ScheduleMissedFirePolicy missedFirePolicy) { + ScheduleMissedFirePolicy effectivePolicy = missedFirePolicy == null + ? ScheduleMissedFirePolicy.FIRE_ALL_MISSED + : missedFirePolicy; + switch (effectivePolicy) { + case SKIP_MISSED: + return new SkipMissedCronScheduleBuilderFactory(); + case FIRE_ONCE_NOW: + return new FireOnceNowCronScheduleBuilderFactory(); + case FIRE_ALL_MISSED: + return new FireAllMissedCronScheduleBuilderFactory(); + default: + throw new IllegalArgumentException("Unsupported schedule missed fire policy: " + missedFirePolicy); + } + } +} diff --git a/dolphinscheduler-scheduler-plugin/dolphinscheduler-scheduler-quartz/src/main/java/org/apache/dolphinscheduler/scheduler/quartz/FireAllMissedCronScheduleBuilderFactory.java b/dolphinscheduler-scheduler-plugin/dolphinscheduler-scheduler-quartz/src/main/java/org/apache/dolphinscheduler/scheduler/quartz/FireAllMissedCronScheduleBuilderFactory.java new file mode 100644 index 000000000000..b3022d1cb569 --- /dev/null +++ b/dolphinscheduler-scheduler-plugin/dolphinscheduler-scheduler-quartz/src/main/java/org/apache/dolphinscheduler/scheduler/quartz/FireAllMissedCronScheduleBuilderFactory.java @@ -0,0 +1,33 @@ +/* + * 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.scheduler.quartz; + +import org.apache.dolphinscheduler.common.utils.DateUtils; +import org.apache.dolphinscheduler.dao.entity.Schedule; + +import org.quartz.CronScheduleBuilder; + +final class FireAllMissedCronScheduleBuilderFactory implements CronScheduleBuilderFactory { + + @Override + public CronScheduleBuilder createCronScheduleBuilder(Schedule schedule) { + return CronScheduleBuilder.cronSchedule(schedule.getCrontab()) + .withMisfireHandlingInstructionIgnoreMisfires() + .inTimeZone(DateUtils.getTimezone(schedule.getTimezoneId())); + } +} diff --git a/dolphinscheduler-scheduler-plugin/dolphinscheduler-scheduler-quartz/src/main/java/org/apache/dolphinscheduler/scheduler/quartz/FireOnceNowCronScheduleBuilderFactory.java b/dolphinscheduler-scheduler-plugin/dolphinscheduler-scheduler-quartz/src/main/java/org/apache/dolphinscheduler/scheduler/quartz/FireOnceNowCronScheduleBuilderFactory.java new file mode 100644 index 000000000000..9990f429fd63 --- /dev/null +++ b/dolphinscheduler-scheduler-plugin/dolphinscheduler-scheduler-quartz/src/main/java/org/apache/dolphinscheduler/scheduler/quartz/FireOnceNowCronScheduleBuilderFactory.java @@ -0,0 +1,33 @@ +/* + * 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.scheduler.quartz; + +import org.apache.dolphinscheduler.common.utils.DateUtils; +import org.apache.dolphinscheduler.dao.entity.Schedule; + +import org.quartz.CronScheduleBuilder; + +final class FireOnceNowCronScheduleBuilderFactory implements CronScheduleBuilderFactory { + + @Override + public CronScheduleBuilder createCronScheduleBuilder(Schedule schedule) { + return CronScheduleBuilder.cronSchedule(schedule.getCrontab()) + .withMisfireHandlingInstructionFireAndProceed() + .inTimeZone(DateUtils.getTimezone(schedule.getTimezoneId())); + } +} diff --git a/dolphinscheduler-scheduler-plugin/dolphinscheduler-scheduler-quartz/src/main/java/org/apache/dolphinscheduler/scheduler/quartz/QuartzCornTriggerBuilder.java b/dolphinscheduler-scheduler-plugin/dolphinscheduler-scheduler-quartz/src/main/java/org/apache/dolphinscheduler/scheduler/quartz/QuartzCornTriggerBuilder.java index b7177e74db84..270f06eb970a 100644 --- a/dolphinscheduler-scheduler-plugin/dolphinscheduler-scheduler-quartz/src/main/java/org/apache/dolphinscheduler/scheduler/quartz/QuartzCornTriggerBuilder.java +++ b/dolphinscheduler-scheduler-plugin/dolphinscheduler-scheduler-quartz/src/main/java/org/apache/dolphinscheduler/scheduler/quartz/QuartzCornTriggerBuilder.java @@ -81,14 +81,14 @@ public CronTrigger build() { JobKey jobKey = QuartzJobKey.of(projectId, schedule.getId()).toJobKey(); TriggerKey triggerKey = TriggerKey.triggerKey(jobKey.getName(), jobKey.getGroup()); + CronScheduleBuilder scheduleBuilder = CronScheduleBuilderFactory.getFactory(schedule.getMissedFirePolicy()) + .createCronScheduleBuilder(schedule); + return TriggerBuilder.newTrigger() .withIdentity(triggerKey) .startAt(startDate) .endAt(endDate) - .withSchedule( - CronScheduleBuilder.cronSchedule(schedule.getCrontab()) - .withMisfireHandlingInstructionIgnoreMisfires() - .inTimeZone(DateUtils.getTimezone(schedule.getTimezoneId()))) + .withSchedule(scheduleBuilder) .build(); } diff --git a/dolphinscheduler-scheduler-plugin/dolphinscheduler-scheduler-quartz/src/main/java/org/apache/dolphinscheduler/scheduler/quartz/SkipMissedCronScheduleBuilderFactory.java b/dolphinscheduler-scheduler-plugin/dolphinscheduler-scheduler-quartz/src/main/java/org/apache/dolphinscheduler/scheduler/quartz/SkipMissedCronScheduleBuilderFactory.java new file mode 100644 index 000000000000..2d948085acca --- /dev/null +++ b/dolphinscheduler-scheduler-plugin/dolphinscheduler-scheduler-quartz/src/main/java/org/apache/dolphinscheduler/scheduler/quartz/SkipMissedCronScheduleBuilderFactory.java @@ -0,0 +1,33 @@ +/* + * 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.scheduler.quartz; + +import org.apache.dolphinscheduler.common.utils.DateUtils; +import org.apache.dolphinscheduler.dao.entity.Schedule; + +import org.quartz.CronScheduleBuilder; + +final class SkipMissedCronScheduleBuilderFactory implements CronScheduleBuilderFactory { + + @Override + public CronScheduleBuilder createCronScheduleBuilder(Schedule schedule) { + return CronScheduleBuilder.cronSchedule(schedule.getCrontab()) + .withMisfireHandlingInstructionDoNothing() + .inTimeZone(DateUtils.getTimezone(schedule.getTimezoneId())); + } +} diff --git a/dolphinscheduler-scheduler-plugin/dolphinscheduler-scheduler-quartz/src/test/java/org/apache/dolphinscheduler/scheduler/quartz/CronScheduleBuilderFactoryTest.java b/dolphinscheduler-scheduler-plugin/dolphinscheduler-scheduler-quartz/src/test/java/org/apache/dolphinscheduler/scheduler/quartz/CronScheduleBuilderFactoryTest.java new file mode 100644 index 000000000000..f006472188b0 --- /dev/null +++ b/dolphinscheduler-scheduler-plugin/dolphinscheduler-scheduler-quartz/src/test/java/org/apache/dolphinscheduler/scheduler/quartz/CronScheduleBuilderFactoryTest.java @@ -0,0 +1,82 @@ +/* + * 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.scheduler.quartz; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; + +import org.apache.dolphinscheduler.common.enums.ScheduleMissedFirePolicy; +import org.apache.dolphinscheduler.dao.entity.Schedule; + +import org.junit.jupiter.api.Test; +import org.quartz.CronTrigger; +import org.quartz.Trigger; + +class CronScheduleBuilderFactoryTest { + + private static final String CRON_EXPRESSION = "0 0 * * * ?"; + + private static final String TIMEZONE_ID = "Asia/Shanghai"; + + @Test + void shouldCreateSkipMissedCronScheduleBuilder() { + assertFactoryAndMisfireInstruction( + ScheduleMissedFirePolicy.SKIP_MISSED, + SkipMissedCronScheduleBuilderFactory.class, + CronTrigger.MISFIRE_INSTRUCTION_DO_NOTHING); + } + + @Test + void shouldCreateFireOnceNowCronScheduleBuilder() { + assertFactoryAndMisfireInstruction( + ScheduleMissedFirePolicy.FIRE_ONCE_NOW, + FireOnceNowCronScheduleBuilderFactory.class, + CronTrigger.MISFIRE_INSTRUCTION_FIRE_ONCE_NOW); + } + + @Test + void shouldCreateFireAllMissedCronScheduleBuilder() { + assertFactoryAndMisfireInstruction( + ScheduleMissedFirePolicy.FIRE_ALL_MISSED, + FireAllMissedCronScheduleBuilderFactory.class, + Trigger.MISFIRE_INSTRUCTION_IGNORE_MISFIRE_POLICY); + } + + @Test + void shouldCreateFireAllMissedCronScheduleBuilderByDefault() { + assertFactoryAndMisfireInstruction( + null, + FireAllMissedCronScheduleBuilderFactory.class, + Trigger.MISFIRE_INSTRUCTION_IGNORE_MISFIRE_POLICY); + } + + private void assertFactoryAndMisfireInstruction( + ScheduleMissedFirePolicy policy, + Class expectedFactoryClass, + int expectedMisfireInstruction) { + Schedule schedule = new Schedule(); + schedule.setCrontab(CRON_EXPRESSION); + schedule.setTimezoneId(TIMEZONE_ID); + CronScheduleBuilderFactory factory = CronScheduleBuilderFactory.getFactory(policy); + assertInstanceOf(expectedFactoryClass, factory); + Trigger trigger = factory.createCronScheduleBuilder(schedule).build(); + CronTrigger cronTrigger = assertInstanceOf(CronTrigger.class, trigger); + assertEquals(expectedMisfireInstruction, cronTrigger.getMisfireInstruction()); + assertEquals(TIMEZONE_ID, cronTrigger.getTimeZone().getID()); + } +} diff --git a/dolphinscheduler-ui/src/locales/en_US/project.ts b/dolphinscheduler-ui/src/locales/en_US/project.ts index 40e553d4dd1d..534c11915b84 100644 --- a/dolphinscheduler-ui/src/locales/en_US/project.ts +++ b/dolphinscheduler-ui/src/locales/en_US/project.ts @@ -149,6 +149,10 @@ export default { start_time: 'Start Time', end_time: 'End Time', crontab: 'Crontab', + missed_fire_policy: 'Missed Fire Policy', + skip_missed: 'Skip missed executions', + fire_once_now: 'Fire once immediately', + fire_all_missed: 'Fire all missed executions', delete_confirm: 'Delete?', delete_confirm_with_name: 'Delete "{name}"?', delete_irreversible: diff --git a/dolphinscheduler-ui/src/locales/zh_CN/project.ts b/dolphinscheduler-ui/src/locales/zh_CN/project.ts index abaa84a34b1a..429fd4990bf7 100644 --- a/dolphinscheduler-ui/src/locales/zh_CN/project.ts +++ b/dolphinscheduler-ui/src/locales/zh_CN/project.ts @@ -148,6 +148,10 @@ export default { start_time: '开始时间', end_time: '结束时间', crontab: 'Crontab', + missed_fire_policy: '定时错过策略', + skip_missed: '跳过错过的执行，等待下一次调度', + fire_once_now: '立即补触发一次，之后按正常节奏继续调度', + fire_all_missed: '补触发所有错过的执行，之后按正常节奏继续调度', delete_confirm: '确定删除吗?', delete_confirm_with_name: '确定删除“{name}”吗?', delete_irreversible: '此操作不可撤销。工作流及其关联数据将被永久删除。', diff --git a/dolphinscheduler-ui/src/views/projects/workflow/definition/components/timing-modal.tsx b/dolphinscheduler-ui/src/views/projects/workflow/definition/components/timing-modal.tsx index c30230581a65..5d15f06962a2 100644 --- a/dolphinscheduler-ui/src/views/projects/workflow/definition/components/timing-modal.tsx +++ b/dolphinscheduler-ui/src/views/projects/workflow/definition/components/timing-modal.tsx @@ -295,6 +295,8 @@ export default defineComponent({ ] timingState.timingForm.crontab = props.row.crontab timingState.timingForm.timezoneId = props.row.timezoneId + timingState.timingForm.missedFirePolicy = + props.row.missedFirePolicy || 'FIRE_ALL_MISSED' timingState.timingForm.failureStrategy = props.row.failureStrategy timingState.timingForm.warningType = props.row.warningType timingState.timingForm.workflowInstancePriority = @@ -410,6 +412,28 @@ export default defineComponent({ ) : null} + + + { ], crontab: '0 0 * * * ? *', timezoneId: Intl.DateTimeFormat().resolvedOptions().timeZone, + missedFirePolicy: 'FIRE_ALL_MISSED', failureStrategy: 'CONTINUE', warningType: 'NONE', workflowInstancePriority: 'MEDIUM', diff --git a/dolphinscheduler-ui/src/views/projects/workflow/definition/components/use-modal.ts b/dolphinscheduler-ui/src/views/projects/workflow/definition/components/use-modal.ts index 4baac0f1d1dc..203ae5bc447e 100644 --- a/dolphinscheduler-ui/src/views/projects/workflow/definition/components/use-modal.ts +++ b/dolphinscheduler-ui/src/views/projects/workflow/definition/components/use-modal.ts @@ -180,7 +180,8 @@ export function useModal( startTime: start, endTime: end, crontab: state.timingForm.crontab, - timezoneId: state.timingForm.timezoneId + timezoneId: state.timingForm.timezoneId, + missedFirePolicy: state.timingForm.missedFirePolicy }), failureStrategy: state.timingForm.failureStrategy, warningType: state.timingForm.warningType,