natas_hw
/
dolphinscheduler0701
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

[feat] Support set execute type to pydolphinscheduler (#12871) 

Up to now, we can only submit workflow with parallel mode. this patch give users ability specific execute type to workflow

(cherry picked from commit 87a88e3662)
This commit is contained in:
Jay Chung 2022-11-12 10:31:08 +08:00
parent 682829e4ea
commit 696d8ae7f1
1 changed files with 13 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -234,28 +234,34 @@ public class PythonGateway {
String taskRelationJson,
String taskDefinitionJson,
String otherParamsJson,
ProcessExecutionTypeEnum executionType) {
String executionType) {
User user = usersService.queryUser(userName);
Project project = projectMapper.queryByName(projectName);
long projectCode = project.getCode();
ProcessDefinition processDefinition = getProcessDefinition(user, projectCode, name);
ProcessExecutionTypeEnum executionTypeEnum = ProcessExecutionTypeEnum.valueOf(executionType);
long processDefinitionCode;
// create or update process definition
if (processDefinition != null) {
processDefinitionCode = processDefinition.getCode();
// make sure process definition offline which could edit
processDefinitionService.releaseProcessDefinition(user, projectCode, processDefinitionCode, ReleaseState.OFFLINE);
Map<String, Object> result = processDefinitionService.updateProcessDefinition(user, projectCode, name, processDefinitionCode, description, globalParams,
null, timeout, tenantCode, taskRelationJson, taskDefinitionJson, otherParamsJson, executionType);
processDefinitionService.releaseProcessDefinition(user, projectCode, processDefinitionCode,
ReleaseState.OFFLINE);
processDefinitionService.updateProcessDefinition(user, projectCode, name,
processDefinitionCode, description, globalParams,
null, timeout, tenantCode, taskRelationJson, taskDefinitionJson, otherParamsJson,
executionTypeEnum);
} else {
Map<String, Object> result = processDefinitionService.createProcessDefinition(user, projectCode, name, description, globalParams,
null, timeout, tenantCode, taskRelationJson, taskDefinitionJson, otherParamsJson, executionType);
Map<String, Object> result = processDefinitionService.createProcessDefinition(user, projectCode, name,
description, globalParams,
null, timeout, tenantCode, taskRelationJson, taskDefinitionJson, otherParamsJson,
executionTypeEnum);
processDefinition = (ProcessDefinition) result.get(Constants.DATA_LIST);
processDefinitionCode = processDefinition.getCode();
}
// Fresh process definition schedule
// Fresh process definition schedule
if (schedule != null) {
createOrUpdateSchedule(user, projectCode, processDefinitionCode, schedule, workerGroup, warningType, warningGroupId);
}