test/DolphinScheduler
1
0
Fork 0
mirror of https://gitee.com/dolphinscheduler/DolphinScheduler.git synced 2024-12-02 04:08:31 +08:00
Code Issues Actions 10 Packages Projects Releases Wiki Activity

feat: Python add online_schedule for workflow control schedule state (#13551)

Browse Source 
related: apache/dolphinscheduler-sdk-python#73
This commit is contained in:
Jay Chung 2023-03-07 11:35:01 +08:00
parent c41154642f
commit 9d2140a651
1 changed files with 12 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/python/PythonGateway.java
View File

@ -215,6 +215,7 @@ public class PythonGateway {
* @param globalParams global params
* @param schedule schedule for workflow, will not set schedule if null,
* and if would always fresh exists schedule if not null
* @param onlineSchedule Whether set the workflow's schedule to online state
* @param warningType warning type
* @param warningGroupId warning group id
* @param timeout timeout for workflow working, if running time longer than timeout,
@ -231,6 +232,7 @@ public class PythonGateway {
String description,
String globalParams,
String schedule,
boolean onlineSchedule,
String warningType,
int warningGroupId,
int timeout,
@ -272,7 +274,9 @@ public class PythonGateway {
// Fresh workflow schedule
if (schedule != null) {
createOrUpdateSchedule(user, projectCode, processDefinitionCode, schedule, workerGroup, warningType, warningGroupId);
createOrUpdateSchedule(user, projectCode, processDefinitionCode, schedule, onlineSchedule, workerGroup,
warningType,
warningGroupId);
}
processDefinitionService.releaseProcessDefinition(user, projectCode, processDefinitionCode, ReleaseState.getEnum(releaseState));
return processDefinitionCode;
@ -312,6 +316,7 @@ public class PythonGateway {
* @param projectCode project which workflow belongs to
* @param workflowCode workflow code
* @param schedule schedule expression
* @param onlineSchedule Whether set the workflow's schedule to online state
* @param workerGroup work group
* @param warningType warning type
* @param warningGroupId warning group id
@ -320,6 +325,7 @@ public class PythonGateway {
long projectCode,
long workflowCode,
String schedule,
boolean onlineSchedule,
String workerGroup,
String warningType,
int warningGroupId) {
@ -340,9 +346,11 @@ public class PythonGateway {
schedulerService.updateSchedule(user, projectCode, scheduleId, schedule, WarningType.valueOf(warningType),
warningGroupId, DEFAULT_FAILURE_STRATEGY, DEFAULT_PRIORITY, workerGroup, DEFAULT_ENVIRONMENT_CODE);
}
// Always set workflow online to set schedule online
processDefinitionService.releaseProcessDefinition(user, projectCode, workflowCode, ReleaseState.ONLINE);
schedulerService.setScheduleState(user, projectCode, scheduleId, ReleaseState.ONLINE);
if (onlineSchedule) {
// set workflow online to make sure we can set schedule online
processDefinitionService.releaseProcessDefinition(user, projectCode, workflowCode, ReleaseState.ONLINE);
schedulerService.setScheduleState(user, projectCode, scheduleId, ReleaseState.ONLINE);
}
}
public void execWorkflowInstance(String userName,