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

Use parallelStream to improve the performance of upgrade (#13442)

Browse Source
This commit is contained in:
Wenjun Ruan 2023-01-23 13:59:20 +08:00 committed by GitHub
parent 713fdbc444
commit ef47e7efeb
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 28 additions and 35 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

63
dolphinscheduler-tools/src/main/java/org/apache/dolphinscheduler/tools/datasource/upgrader/v320/V320DolphinSchedulerUpgrader.java
View File

@ -86,25 +86,22 @@ public class V320DolphinSchedulerUpgrader implements DolphinSchedulerUpgrader {
if (CollectionUtils.isEmpty(needUpdateWorkflowInstance)) {
return;
}
for (ProcessInstance processInstance : needUpdateWorkflowInstance) {
ProcessDefinitionLog processDefinitionLog = processDefinitionLogMapper.queryByDefinitionCodeAndVersion(
processInstance.getProcessDefinitionCode(), processInstance.getProcessDefinitionVersion());
if (processDefinitionLog != null) {
processInstance.setProjectCode(processDefinitionLog.getProjectCode());
processInstance.setTenantCode(tenantMap.get(processDefinitionLog.getTenantId()));
processInstance.setExecutorName(userMap.get(processInstance.getExecutorId()));
} else {
processInstance.setProjectCode(-1L);
}
processInstanceMapper.updateById(processInstance);
}
needUpdateWorkflowInstance.parallelStream()
.forEach(processInstance -> {
ProcessDefinitionLog processDefinitionLog =
processDefinitionLogMapper.queryByDefinitionCodeAndVersion(
processInstance.getProcessDefinitionCode(),
processInstance.getProcessDefinitionVersion());
if (processDefinitionLog != null) {
processInstance.setProjectCode(processDefinitionLog.getProjectCode());
processInstance.setTenantCode(tenantMap.get(processDefinitionLog.getTenantId()));
processInstance.setExecutorName(userMap.get(processInstance.getExecutorId()));
} else {
processInstance.setProjectCode(-1L);
}
processInstanceMapper.updateById(processInstance);
});
log.info("Success upgrade workflow instance, current batch size: {}", needUpdateWorkflowInstance.size());
try {
Thread.sleep(1000L);
} catch (InterruptedException e) {
log.error("Upgrade workflow instance error", e);
}
}
}
@ -118,24 +115,20 @@ public class V320DolphinSchedulerUpgrader implements DolphinSchedulerUpgrader {
if (CollectionUtils.isEmpty(taskInstances)) {
return;
}
for (TaskInstance taskInstance : taskInstances) {
ProcessInstance processInstance = processInstanceMapper.selectById(taskInstance.getProcessInstanceId());
if (processInstance == null) {
taskInstance.setProjectCode(-1L);
} else {
taskInstance.setProjectCode(processInstance.getProjectCode());
taskInstance.setProcessInstanceName(processInstance.getName());
taskInstance.setExecutorName(processInstance.getExecutorName());
}
taskInstanceMapper.updateById(taskInstance);
}
taskInstances.parallelStream()
.forEach(taskInstance -> {
ProcessInstance processInstance =
processInstanceMapper.selectById(taskInstance.getProcessInstanceId());
if (processInstance == null) {
taskInstance.setProjectCode(-1L);
} else {
taskInstance.setProjectCode(processInstance.getProjectCode());
taskInstance.setProcessInstanceName(processInstance.getName());
taskInstance.setExecutorName(processInstance.getExecutorName());
}
taskInstanceMapper.updateById(taskInstance);
});
log.info("Success upgrade task instance, current batch size: {}", taskInstances.size());
try {
Thread.sleep(1000L);
} catch (InterruptedException e) {
log.error("Upgrade task instance error", e);
}
}
}