Skip to content

[Improvement-18662][API]Aligns insertSchedule with updateSchedule (which already allows offline workflows to save) #18664

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
SbloodyS merged 7 commits into from
Sep 30, 2026
Merged

[Improvement-18662][API]Aligns insertSchedule with updateSchedule (which already allows offline workflows to save) #18664

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
4fddb92
Allow creating schedule for offline workflow definition
Sep 23, 2026
d7773ca
Align insertSchedule validation with updateSchedule to allow scheduli…
Sep 23, 2026
b6ae144
Merge branch 'Improvement-18662' of github.com:njnu-seafish/dolphinsc…
Sep 23, 2026
30352e5
Merge branch 'apache:dev' into Improvement-18662
njnu-seafish Sep 24, 2026
00c1da0
Tighten tests to lock the saved-but-not-scheduled contract for offlin…
Sep 29, 2026
484cb79
Preserve subworkflow validation by checking subworkflow readiness whe…
Sep 29, 2026
54397c1
Merge branch 'dev' into Improvement-18662
SbloodyS Sep 30, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 9 additions & 4 deletions ...-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/SchedulerServiceImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -129,10 +129,12 @@ public Schedule insertSchedule(User loginUser,

projectService.checkHasProjectWritePermissionThrowException(loginUser, project);

// check workflow define release state
WorkflowDefinition workflowDefinition = workflowDefinitionDao.queryByCode(workflowDefinitionCode).orElse(null);
executorService.checkWorkflowDefinitionValid(projectCode, workflowDefinition, workflowDefinitionCode,
workflowDefinition.getVersion());
// check workflow definition exists
WorkflowDefinition workflowDefinition = workflowDefinitionDao.queryByCode(workflowDefinitionCode)
.orElseThrow(() -> new ServiceException(Status.WORKFLOW_DEFINITION_NOT_EXIST, workflowDefinitionCode));
if (projectCode != workflowDefinition.getProjectCode()) {
throw new ServiceException(Status.WORKFLOW_DEFINITION_NOT_EXIST, workflowDefinitionCode);
}

Schedule scheduleExists =
scheduleDao.queryByWorkflowDefinitionCode(workflowDefinitionCode);
Expand Down Expand Up @@ -476,6 +478,9 @@ private void doOnlineScheduler(Schedule schedule) {
if (!ReleaseState.ONLINE.equals(workflowDefinition.getReleaseState())) {
throw new ServiceException(Status.WORKFLOW_DEFINITION_NOT_RELEASE, workflowDefinition.getName());
}
if (!executorService.checkSubWorkflowDefinitionValid(workflowDefinition)) {
throw new ServiceException(Status.SUB_WORKFLOW_DEFINITION_NOT_RELEASE);
}

schedule.setReleaseState(ReleaseState.ONLINE);
schedule.setUpdateTime(new Date());
Expand Down
113 changes: 113 additions & 0 deletions ...duler-api/src/test/java/org/apache/dolphinscheduler/api/service/SchedulerServiceTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -324,6 +324,119 @@ public void testDeleteSchedules() {
Assertions.assertDoesNotThrow(() -> schedulerService.deleteSchedulesById(user, scheduleId));
}

@Test
public void testInsertScheduleWorkflowNotExists() {
Mockito.when(projectDao.queryByCode(projectCode)).thenReturn(this.getProject());
Mockito.when(workflowDefinitionDao.queryByCode(processDefinitionCode))
.thenReturn(Optional.empty());

exception = Assertions.assertThrows(ServiceException.class,
() -> schedulerService.insertSchedule(
user, projectCode, processDefinitionCode, scheduleExpression(null), WarningType.NONE, 0,
FailureStrategy.CONTINUE, Priority.MEDIUM, "default", "tenantCode", environmentCode));
Assertions.assertEquals(Status.WORKFLOW_DEFINITION_NOT_EXIST.getCode(),
((ServiceException) exception).getCode());
}

@Test
public void testInsertScheduleWorkflowFromAnotherProject() {
Project project = this.getProject();
WorkflowDefinition workflowDefinition = this.getProcessDefinition();
workflowDefinition.setProjectCode(999L);
Mockito.when(projectDao.queryByCode(projectCode)).thenReturn(project);
Mockito.when(workflowDefinitionDao.queryByCode(processDefinitionCode))
.thenReturn(Optional.of(workflowDefinition));

exception = Assertions.assertThrows(ServiceException.class,
() -> schedulerService.insertSchedule(
user, projectCode, processDefinitionCode, scheduleExpression(null), WarningType.NONE, 0,
FailureStrategy.CONTINUE, Priority.MEDIUM, "default", "tenantCode", environmentCode));
Assertions.assertEquals(Status.WORKFLOW_DEFINITION_NOT_EXIST.getCode(),
((ServiceException) exception).getCode());
}

@Test
public void testInsertScheduleOfflineWorkflow() {
Project project = this.getProject();
WorkflowDefinition workflowDefinition = this.getProcessDefinition();
workflowDefinition.setReleaseState(ReleaseState.OFFLINE);
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(null), WarningType.NONE, 0,
FailureStrategy.CONTINUE, Priority.MEDIUM, "default", "tenantCode", environmentCode);

ArgumentCaptor<Schedule> scheduleCaptor = ArgumentCaptor.forClass(Schedule.class);
Mockito.verify(scheduleDao).insert(scheduleCaptor.capture());
Assertions.assertSame(insertedSchedule, result);
Assertions.assertEquals(ReleaseState.OFFLINE, scheduleCaptor.getValue().getReleaseState());
Mockito.verifyNoInteractions(schedulerApi);
}

@Test
public void testOnlineSchedulerRejectsOfflineWorkflow() {
Schedule schedule = this.getSchedule();
schedule.setReleaseState(ReleaseState.OFFLINE);
WorkflowDefinition workflowDefinition = this.getProcessDefinition();
workflowDefinition.setReleaseState(ReleaseState.OFFLINE);
Mockito.when(projectDao.queryByCode(projectCode)).thenReturn(this.getProject());
Mockito.when(scheduleDao.queryById(scheduleId)).thenReturn(schedule);
Mockito.when(workflowDefinitionDao.queryByCode(processDefinitionCode))
.thenReturn(Optional.of(workflowDefinition));

ServiceException ex = Assertions.assertThrows(ServiceException.class,
() -> schedulerService.onlineScheduler(user, projectCode, scheduleId));
Assertions.assertEquals(Status.WORKFLOW_DEFINITION_NOT_RELEASE.getCode(), ex.getCode());
Mockito.verify(scheduleDao, Mockito.never()).updateById(Mockito.any());
Mockito.verifyNoInteractions(schedulerApi);
}

@Test
public void testOnlineSchedulerRejectsOfflineSubWorkflow() {
Schedule schedule = this.getSchedule();
schedule.setReleaseState(ReleaseState.OFFLINE);
WorkflowDefinition workflowDefinition = this.getProcessDefinition();
workflowDefinition.setReleaseState(ReleaseState.ONLINE);
Mockito.when(projectDao.queryByCode(projectCode)).thenReturn(this.getProject());
Mockito.when(scheduleDao.queryById(scheduleId)).thenReturn(schedule);
Mockito.when(workflowDefinitionDao.queryByCode(processDefinitionCode))
.thenReturn(Optional.of(workflowDefinition));
Mockito.when(executorService.checkSubWorkflowDefinitionValid(workflowDefinition)).thenReturn(false);

ServiceException ex = Assertions.assertThrows(ServiceException.class,
() -> schedulerService.onlineScheduler(user, projectCode, scheduleId));
Assertions.assertEquals(Status.SUB_WORKFLOW_DEFINITION_NOT_RELEASE.getCode(), ex.getCode());
Assertions.assertEquals(ReleaseState.OFFLINE, schedule.getReleaseState());
Mockito.verify(scheduleDao, Mockito.never()).updateById(Mockito.any());
Mockito.verifyNoInteractions(schedulerApi);
}

@Test
public void testOnlineSchedulerSucceedsWhenSubWorkflowOnline() {
Schedule schedule = this.getSchedule();
schedule.setReleaseState(ReleaseState.OFFLINE);
WorkflowDefinition workflowDefinition = this.getProcessDefinition();
workflowDefinition.setReleaseState(ReleaseState.ONLINE);
Project project = this.getProject();
Mockito.when(projectDao.queryByCode(projectCode)).thenReturn(project);
Mockito.when(scheduleDao.queryById(scheduleId)).thenReturn(schedule);
Mockito.when(workflowDefinitionDao.queryByCode(processDefinitionCode))
.thenReturn(Optional.of(workflowDefinition));
Mockito.when(executorService.checkSubWorkflowDefinitionValid(workflowDefinition)).thenReturn(true);

schedulerService.onlineScheduler(user, projectCode, scheduleId);

Assertions.assertEquals(ReleaseState.ONLINE, schedule.getReleaseState());
Mockito.verify(scheduleDao).updateById(schedule);
Mockito.verify(schedulerApi).insertOrUpdateScheduleTask(project.getId(), schedule);
}

@Test
public void testReadOnlyUserCannotChangeSchedule() {
Project project = this.getProject();
Expand Down
Loading