[dolphinscheduler] branch dev updated: Use parallelStream to improve the performance of upgrade (#13442)

[wenjun]

This is an automated email from the ASF dual-hosted git repository.

wenjun pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/dolphinscheduler.git


The following commit(s) were added to refs/heads/dev by this push:
     new ef47e7efeb Use parallelStream to improve the performance of upgrade 
(#13442)
ef47e7efeb is described below

commit ef47e7efeb8f0c68d3fc1a1c92621333b6479ae9
Author: Wenjun Ruan <wen...@apache.org>
AuthorDate: Mon Jan 23 13:59:20 2023 +0800

    Use parallelStream to improve the performance of upgrade (#13442)
---
 .../v320/V320DolphinSchedulerUpgrader.java         | 63 ++++++++++------------
 1 file changed, 28 insertions(+), 35 deletions(-)

diff --git 
a/dolphinscheduler-tools/src/main/java/org/apache/dolphinscheduler/tools/datasource/upgrader/v320/V320DolphinSchedulerUpgrader.java
 
b/dolphinscheduler-tools/src/main/java/org/apache/dolphinscheduler/tools/datasource/upgrader/v320/V320DolphinSchedulerUpgrader.java
index 32e6b011d6..d33c9017d1 100644
--- 
a/dolphinscheduler-tools/src/main/java/org/apache/dolphinscheduler/tools/datasource/upgrader/v320/V320DolphinSchedulerUpgrader.java
+++ 
b/dolphinscheduler-tools/src/main/java/org/apache/dolphinscheduler/tools/datasource/upgrader/v320/V320DolphinSchedulerUpgrader.java
@@ -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);
-            }
         }
     }