github
/
DolphinScheduler
mirror of https://gitee.com/dolphinscheduler/DolphinScheduler.git
1
0
Fork 0
Code Issues Releases Wiki Activity
Browse Source

[Feature][master] change sub process id to code. (#6846) 

* 6597 change sub process id to code.

* 6597 change sub process id to code.

* 6597 change sub process id to code.

* 6597 change sub process id to code.

* 6597 change sub process id to code.

* 6597 change sub process id to code.

* code style

* fix ut process

* fix ut process
3.0.0/version-upgrade
OS 3 years ago committed by GitHub
parent
7f3b999e5c
commit
880c1d28ff
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
12 changed files with 60 additions and 91 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 14
      dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/dto/treeview/Instance.java
  2. 9
      dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ExecutorServiceImpl.java
  3. 14
      dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ProcessDefinitionServiceImpl.java
  4. 9
      dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/SchedulerServiceImpl.java
  5. 6
      dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/ExecutorServiceTest.java
  6. 2
      dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/utils/CheckUtilsTest.java
  7. 2
      dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/Constants.java
  8. 13
      dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/task/subprocess/SubProcessParameters.java
  9. 2
      dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/TaskInstanceMapperTest.java
  10. 49
      dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessService.java
  11. 17
      dolphinscheduler-service/src/test/java/org/apache/dolphinscheduler/service/process/ProcessServiceTest.java
  12. 14
      dolphinscheduler-ui/src/js/conf/home/pages/dag/_source/formModel/tasks/sub_process.vue

14
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/dto/treeview/Instance.java
Unescape View File

@ -71,7 +71,7 @@ public class Instance {
*/
private String duration;
private int subflowId;
private long subflowCode;
public Instance() {
}
@ -83,7 +83,7 @@ public class Instance {
this.type = type;
}
public Instance(int id, String name, long code, String type, String state, Date startTime, Date endTime, String host, String duration, int subflowId) {
public Instance(int id, String name, long code, String type, String state, Date startTime, Date endTime, String host, String duration, long subflowCode) {
this.id = id;
this.name = name;
this.code = code;
@ -93,7 +93,7 @@ public class Instance {
this.endTime = endTime;
this.host = host;
this.duration = duration;
this.subflowId = subflowId;
this.subflowCode = subflowCode;
}
public Instance(int id, String name, long code, String type, String state, Date startTime, Date endTime, String host, String duration) {
@ -173,11 +173,11 @@ public class Instance {
this.duration = duration;
}
public int getSubflowId() {
return subflowId;
public long getSubflowCode() {
return subflowCode;
}
public void setSubflowId(int subflowId) {
this.subflowId = subflowId;
public void setSubflowCode(long subflowCode) {
this.subflowCode = subflowCode;
}
}

9
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ExecutorServiceImpl.java
Unescape View File

@ -465,11 +465,10 @@ public class ExecutorServiceImpl extends BaseServiceImpl implements ExecutorServ
return result;
}
List<Integer> ids = new ArrayList<>();
processService.recurseFindSubProcessId(processDefinition.getId(), ids);
Integer[] idArray = ids.toArray(new Integer[ids.size()]);
if (!ids.isEmpty()) {
List<ProcessDefinition> processDefinitionList = processDefinitionMapper.queryDefinitionListByIdList(idArray);
List<Long> codes = new ArrayList<>();
processService.recurseFindSubProcess(processDefinition.getCode(), codes);
if (!codes.isEmpty()) {
List<ProcessDefinition> processDefinitionList = processDefinitionMapper.queryByCodes(codes);
if (processDefinitionList != null) {
for (ProcessDefinition processDefinitionTmp : processDefinitionList) {
/**

14
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ProcessDefinitionServiceImpl.java
Unescape View File

@ -17,7 +17,7 @@
package org.apache.dolphinscheduler.api.service.impl;
import static org.apache.dolphinscheduler.common.Constants.CMD_PARAM_SUB_PROCESS_DEFINE_ID;
import static org.apache.dolphinscheduler.common.Constants.CMD_PARAM_SUB_PROCESS_DEFINE_CODE;
import org.apache.dolphinscheduler.api.dto.DagDataSchedule;
import org.apache.dolphinscheduler.api.dto.treeview.Instance;
@ -33,7 +33,6 @@ import org.apache.dolphinscheduler.api.utils.FileUtils;
import org.apache.dolphinscheduler.api.utils.PageInfo;
import org.apache.dolphinscheduler.api.utils.Result;
import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.common.enums.AuthorizationType;
import org.apache.dolphinscheduler.common.enums.ProcessExecutionTypeEnum;
import org.apache.dolphinscheduler.common.enums.ReleaseState;
import org.apache.dolphinscheduler.common.enums.UserType;
@ -69,7 +68,6 @@ import org.apache.dolphinscheduler.dao.mapper.TaskDefinitionMapper;
import org.apache.dolphinscheduler.dao.mapper.TaskInstanceMapper;
import org.apache.dolphinscheduler.dao.mapper.TenantMapper;
import org.apache.dolphinscheduler.dao.mapper.UserMapper;
import org.apache.dolphinscheduler.service.permission.PermissionCheck;
import org.apache.dolphinscheduler.service.process.ProcessService;
import org.apache.commons.collections4.CollectionUtils;
@ -1202,16 +1200,16 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
Date startTime = taskInstance.getStartTime() == null ? new Date() : taskInstance.getStartTime();
Date endTime = taskInstance.getEndTime() == null ? new Date() : taskInstance.getEndTime();
int subProcessId = 0;
long subProcessCode = 0L;
// if process is sub process, the return sub id, or sub id=0
if (taskInstance.isSubProcess()) {
TaskDefinition taskDefinition = taskDefinitionMap.get(taskInstance.getTaskCode());
subProcessId = Integer.parseInt(JSONUtils.parseObject(
taskDefinition.getTaskParams()).path(CMD_PARAM_SUB_PROCESS_DEFINE_ID).asText());
subProcessCode = Integer.parseInt(JSONUtils.parseObject(
taskDefinition.getTaskParams()).path(CMD_PARAM_SUB_PROCESS_DEFINE_CODE).asText());
}
treeViewDto.getInstances().add(new Instance(taskInstance.getId(), taskInstance.getName(), taskInstance.getTaskCode(),
taskInstance.getTaskType(), taskInstance.getState().toString(), taskInstance.getStartTime(), taskInstance.getEndTime(),
taskInstance.getHost(), DateUtils.format2Readable(endTime.getTime() - startTime.getTime()), subProcessId));
taskInstance.getTaskType(), taskInstance.getState().toString(), taskInstance.getStartTime(), taskInstance.getEndTime(),
taskInstance.getHost(), DateUtils.format2Readable(endTime.getTime() - startTime.getTime()), subProcessCode));
}
}
for (TreeViewDto pTreeViewDto : parentTreeViewDtoList) {

9
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/SchedulerServiceImpl.java
Unescape View File

@ -355,12 +355,11 @@ public class SchedulerServiceImpl extends BaseServiceImpl implements SchedulerSe
return result;
}
// check sub process definition release state
List<Integer> subProcessDefineIds = new ArrayList<>();
processService.recurseFindSubProcessId(processDefinition.getId(), subProcessDefineIds);
Integer[] idArray = subProcessDefineIds.toArray(new Integer[subProcessDefineIds.size()]);
if (!subProcessDefineIds.isEmpty()) {
List<Long> subProcessDefineCodes = new ArrayList<>();
processService.recurseFindSubProcess(processDefinition.getCode(), subProcessDefineCodes);
if (!subProcessDefineCodes.isEmpty()) {
List<ProcessDefinition> subProcessDefinitionList =
processDefinitionMapper.queryDefinitionListByIdList(idArray);
processDefinitionMapper.queryByCodes(subProcessDefineCodes);
if (subProcessDefinitionList != null && !subProcessDefinitionList.isEmpty()) {
for (ProcessDefinition subProcessDefinition : subProcessDefinitionList) {
/**

6
dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/ExecutorServiceTest.java
Unescape View File

@ -279,9 +279,9 @@ public class ExecutorServiceTest {
@Test
public void testStartCheckByProcessDefinedCode() {
List<Integer> ids = new ArrayList<>();
ids.add(1);
Mockito.doNothing().when(processService).recurseFindSubProcessId(1, ids);
List<Long> ids = new ArrayList<>();
ids.add(1L);
Mockito.doNothing().when(processService).recurseFindSubProcess(1, ids);
List<ProcessDefinition> processDefinitionList = new ArrayList<>();
ProcessDefinition processDefinition = new ProcessDefinition();

2
dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/utils/CheckUtilsTest.java
Unescape View File

@ -128,7 +128,7 @@ public class CheckUtilsTest {
taskNode.setType(TaskType.SUB_PROCESS.getDesc());
assertFalse(CheckUtils.checkTaskNodeParameters(taskNode));
subProcessParameters.setProcessDefinitionId(1234);
subProcessParameters.setProcessDefinitionCode(1234L);
taskNode.setParams(JSONUtils.toJsonString(subProcessParameters));
taskNode.setType(TaskType.SUB_PROCESS.getDesc());
assertTrue(CheckUtils.checkTaskNodeParameters(taskNode));

2
dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/Constants.java
Unescape View File

@ -323,7 +323,7 @@ public final class Constants {
public static final String CMD_PARAM_SUB_PROCESS_PARENT_INSTANCE_ID = "parentProcessInstanceId";
public static final String CMD_PARAM_SUB_PROCESS_DEFINE_ID = "processDefinitionId";
public static final String CMD_PARAM_SUB_PROCESS_DEFINE_CODE = "processDefinitionCode";
public static final String CMD_PARAM_START_NODES = "StartNodeList";

13
dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/task/subprocess/SubProcessParameters.java
Unescape View File

@ -15,6 +15,7 @@
* limitations under the License.
*/
package org.apache.dolphinscheduler.common.task.subprocess;
import org.apache.dolphinscheduler.common.process.ResourceInfo;
import org.apache.dolphinscheduler.common.task.AbstractParameters;
@ -27,19 +28,19 @@ public class SubProcessParameters extends AbstractParameters {
/**
* process definition id
*/
private Integer processDefinitionId;
private long processDefinitionCode;
public void setProcessDefinitionId(Integer processDefinitionId){
this.processDefinitionId = processDefinitionId;
public void setProcessDefinitionCode(long processDefinitionCode) {
this.processDefinitionCode = processDefinitionCode;
}
public Integer getProcessDefinitionId(){
return this.processDefinitionId;
public long getProcessDefinitionCode() {
return this.processDefinitionCode;
}
@Override
public boolean checkParameters() {
return this.processDefinitionId != null && this.processDefinitionId != 0;
return this.processDefinitionCode != 0;
}
@Override

2
dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/TaskInstanceMapperTest.java
Unescape View File

@ -304,7 +304,6 @@ public class TaskInstanceMapperTest extends BaseDaoTest {
definition.setCreateTime(new Date());
definition.setUpdateTime(new Date());
processDefinitionMapper.insert(definition);
//task.setProcessDefinitionId(definition.getId());
taskInstanceMapper.updateById(task);
int countTask = taskInstanceMapper.countTask(
@ -340,7 +339,6 @@ public class TaskInstanceMapperTest extends BaseDaoTest {
definition.setCreateTime(new Date());
definition.setUpdateTime(new Date());
processDefinitionMapper.insert(definition);
//task.setProcessDefinitionId(definition.getId());
taskInstanceMapper.updateById(task);

49
dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessService.java
Unescape View File

@ -23,7 +23,7 @@ import static org.apache.dolphinscheduler.common.Constants.CMD_PARAM_EMPTY_SUB_P
import static org.apache.dolphinscheduler.common.Constants.CMD_PARAM_FATHER_PARAMS;
import static org.apache.dolphinscheduler.common.Constants.CMD_PARAM_RECOVER_PROCESS_ID_STRING;
import static org.apache.dolphinscheduler.common.Constants.CMD_PARAM_SUB_PROCESS;
import static org.apache.dolphinscheduler.common.Constants.CMD_PARAM_SUB_PROCESS_DEFINE_ID;
import static org.apache.dolphinscheduler.common.Constants.CMD_PARAM_SUB_PROCESS_DEFINE_CODE;
import static org.apache.dolphinscheduler.common.Constants.CMD_PARAM_SUB_PROCESS_PARENT_INSTANCE_ID;
import static org.apache.dolphinscheduler.common.Constants.LOCAL_PARAMS;
@ -329,18 +329,6 @@ public class ProcessService {
return null;
}
/**
* check thread num
*
* @param command command
* @param validThreadNum validThreadNum
* @return if thread is enough
*/
private boolean checkThreadNum(Command command, int validThreadNum) {
int commandThreadCount = this.workProcessThreadNumCount(command.getProcessDefinitionCode());
return validThreadNum >= commandThreadCount;
}
/**
* insert one command
*
@ -412,8 +400,8 @@ public class ProcessService {
/**
* get task node list by definitionId
*/
public List<TaskDefinition> getTaskNodeListByDefinitionId(Integer defineId) {
ProcessDefinition processDefinition = processDefineMapper.selectById(defineId);
public List<TaskDefinition> getTaskNodeListByDefinition(long defineCode) {
ProcessDefinition processDefinition = processDefineMapper.queryByCode(defineCode);
if (processDefinition == null) {
logger.error("process define not exists");
return new ArrayList<>();
@ -535,38 +523,24 @@ public class ProcessService {
}
}
/**
* calculate sub process number in the process define.
*
* @param processDefinitionCode processDefinitionCode
* @return process thread num count
*/
private Integer workProcessThreadNumCount(long processDefinitionCode) {
ProcessDefinition processDefinition = processDefineMapper.queryByCode(processDefinitionCode);
List<Integer> ids = new ArrayList<>();
recurseFindSubProcessId(processDefinition.getId(), ids);
return ids.size() + 1;
}
/**
* recursive query sub process definition id by parent id.
*
* @param parentId parentId
* @param parentCode parentCode
* @param ids ids
*/
public void recurseFindSubProcessId(int parentId, List<Integer> ids) {
List<TaskDefinition> taskNodeList = this.getTaskNodeListByDefinitionId(parentId);
public void recurseFindSubProcess(long parentCode, List<Long> ids) {
List<TaskDefinition> taskNodeList = this.getTaskNodeListByDefinition(parentCode);
if (taskNodeList != null && !taskNodeList.isEmpty()) {
for (TaskDefinition taskNode : taskNodeList) {
String parameter = taskNode.getTaskParams();
ObjectNode parameterJson = JSONUtils.parseObject(parameter);
if (parameterJson.get(CMD_PARAM_SUB_PROCESS_DEFINE_ID) != null) {
if (parameterJson.get(CMD_PARAM_SUB_PROCESS_DEFINE_CODE) != null) {
SubProcessParameters subProcessParam = JSONUtils.parseObject(parameter, SubProcessParameters.class);
ids.add(subProcessParam.getProcessDefinitionId());
recurseFindSubProcessId(subProcessParam.getProcessDefinitionId(), ids);
ids.add(subProcessParam.getProcessDefinitionCode());
recurseFindSubProcess(subProcessParam.getProcessDefinitionCode(), ids);
}
}
}
@ -678,7 +652,6 @@ public class ProcessService {
processInstance.setStartTime(new Date());
processInstance.setRunTimes(1);
processInstance.setMaxTryTimes(0);
//processInstance.setProcessDefinitionId(command.getProcessDefinitionId());
processInstance.setCommandParam(command.getCommandParam());
processInstance.setCommandType(command.getCommandType());
processInstance.setIsSubProcess(Flag.NO);
@ -1307,8 +1280,8 @@ public class ProcessService {
TaskInstance task) {
CommandType commandType = getSubCommandType(parentProcessInstance, childInstance);
Map<String, String> subProcessParam = JSONUtils.toMap(task.getTaskParams());
int childDefineId = Integer.parseInt(subProcessParam.get(Constants.CMD_PARAM_SUB_PROCESS_DEFINE_ID));
ProcessDefinition subProcessDefinition = processDefineMapper.queryByDefineId(childDefineId);
int childDefineCode = Integer.parseInt(subProcessParam.get(Constants.CMD_PARAM_SUB_PROCESS_DEFINE_CODE));
ProcessDefinition subProcessDefinition = processDefineMapper.queryByCode(childDefineCode);
Object localParams = subProcessParam.get(Constants.LOCAL_PARAMS);
List<Property> allParam = JSONUtils.toList(JSONUtils.toJsonString(localParams), Property.class);

17
dolphinscheduler-service/src/test/java/org/apache/dolphinscheduler/service/process/ProcessServiceTest.java
Unescape View File

@ -19,7 +19,7 @@ package org.apache.dolphinscheduler.service.process;
import static org.apache.dolphinscheduler.common.Constants.CMD_PARAM_RECOVER_PROCESS_ID_STRING;
import static org.apache.dolphinscheduler.common.Constants.CMD_PARAM_START_PARAMS;
import static org.apache.dolphinscheduler.common.Constants.CMD_PARAM_SUB_PROCESS_DEFINE_ID;
import static org.apache.dolphinscheduler.common.Constants.CMD_PARAM_SUB_PROCESS_DEFINE_CODE;
import static org.mockito.ArgumentMatchers.any;
import org.apache.dolphinscheduler.common.Constants;
@ -128,7 +128,7 @@ public class ProcessServiceTest {
parentInstance.setWarningGroupId(0);
TaskInstance task = new TaskInstance();
task.setTaskParams("{\"processDefinitionId\":100}}");
task.setTaskParams("{\"processDefinitionCode\":10}}");
task.setId(10);
task.setTaskCode(1L);
task.setTaskDefinitionVersion(1);
@ -143,8 +143,9 @@ public class ProcessServiceTest {
parentInstance.setHistoryCmd("START_PROCESS");
parentInstance.setCommandType(CommandType.START_PROCESS);
ProcessDefinition processDefinition = new ProcessDefinition();
processDefinition.setCode(1L);
processDefinition.setCode(10L);
Mockito.when(processDefineMapper.queryByDefineId(100)).thenReturn(processDefinition);
Mockito.when(processDefineMapper.queryByCode(10L)).thenReturn(processDefinition);
command = processService.createSubProcessCommand(parentInstance, childInstance, instanceMap, task);
Assert.assertEquals(CommandType.START_PROCESS, command.getCommandType());
@ -246,7 +247,7 @@ public class ProcessServiceTest {
command.setProcessDefinitionCode(222);
command.setCommandType(CommandType.REPEAT_RUNNING);
command.setCommandParam("{\"" + CMD_PARAM_RECOVER_PROCESS_ID_STRING + "\":\"111\",\""
+ CMD_PARAM_SUB_PROCESS_DEFINE_ID + "\":\"222\"}");
+ CMD_PARAM_SUB_PROCESS_DEFINE_CODE + "\":\"222\"}");
Assert.assertNull(processService.handleCommand(logger, host, command, processDefinitionCacheMaps));
int definitionVersion = 1;
@ -442,14 +443,14 @@ public class ProcessServiceTest {
List<TaskDefinitionLog> taskDefinitionLogs = new ArrayList<>();
TaskDefinitionLog taskDefinitionLog1 = new TaskDefinitionLog();
taskDefinitionLog1.setTaskParams("{\"processDefinitionId\": 123}");
taskDefinitionLog1.setTaskParams("{\"processDefinitionCode\": 123L}");
taskDefinitionLogs.add(taskDefinitionLog1);
Mockito.when(taskDefinitionLogMapper.queryByTaskDefinitions(Mockito.anySet())).thenReturn(taskDefinitionLogs);
List<Integer> ids = new ArrayList<>();
processService.recurseFindSubProcessId(parentProcessDefineId, ids);
List<Long> ids = new ArrayList<>();
processService.recurseFindSubProcess(parentProcessDefineCode, ids);
Assert.assertEquals(1, ids.size());
Assert.assertEquals(0, ids.size());
}
@Test

14
dolphinscheduler-ui/src/js/conf/home/pages/dag/_source/formModel/tasks/sub_process.vue
Unescape View File

@ -29,7 +29,7 @@
<el-option
v-for="city in processDefinitionList"
:key="city.code"
:value="city.id"
:value="city.code"
:label="city.name">
</el-option>
</el-select>
@ -72,7 +72,7 @@
return false
}
this.$emit('on-params', {
processDefinitionId: this.wdiCurr
processDefinitionCode: this.wdiCurr
})
return true
},
@ -85,8 +85,8 @@
/**
* Return the name according to the process definition id
*/
_handleName (id) {
return _.filter(this.processDefinitionList, v => id === v.id)[0].name
_handleName (code) {
return _.filter(this.processDefinitionList, v => code === v.code)[0].name
},
/**
* Get all processDefinition list
@ -101,7 +101,7 @@
watch: {
wdiCurr (val) {
this.$emit('on-cache-params', {
processDefinitionId: this.wdiCurr
processDefinitionCode: this.wdiCurr
})
}
},
@ -124,10 +124,10 @@
let o = this.backfillItem
// Non-null objects represent backfill
if (!_.isEmpty(o)) {
this.wdiCurr = o.params.processDefinitionId
this.wdiCurr = o.params.processDefinitionCode
} else {
if (this.processDefinitionList.length) {
this.wdiCurr = this.processDefinitionList[0].id
this.wdiCurr = this.processDefinitionList[0].code
this.$emit('on-set-process-name', this._handleName(this.wdiCurr))
}
}

Write Preview
Loading…
Cancel
Save