[dolphinscheduler] branch 3.0.2-prepare updated: Fix memory leak in worker due to message retry map (#12878) (#12887)

[wenjun]

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

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


The following commit(s) were added to refs/heads/3.0.2-prepare by this push:
     new 53449a06c4 Fix memory leak in worker due to message retry map (#12878) 
(#12887)
53449a06c4 is described below

commit 53449a06c4fa7be26a6ebce7f93a4aa88be34274
Author: Wenjun Ruan <wen...@apache.org>
AuthorDate: Mon Nov 14 12:01:30 2022 +0800

    Fix memory leak in worker due to message retry map (#12878) (#12887)
    
    (cherry picked from commit 403c6a6bb6ec9fc380ffd763301891ad57029e2b)
---
 .../server/worker/message/MessageRetryRunner.java           | 13 +++++++++++--
 1 file changed, 11 insertions(+), 2 deletions(-)

diff --git 
a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/message/MessageRetryRunner.java
 
b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/message/MessageRetryRunner.java
index 18dceb069b..c1cf73b56f 100644
--- 
a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/message/MessageRetryRunner.java
+++ 
b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/message/MessageRetryRunner.java
@@ -26,6 +26,7 @@ import org.apache.dolphinscheduler.remote.command.CommandType;
 
 import java.time.Duration;
 import java.util.HashMap;
+import java.util.Iterator;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 
@@ -106,11 +107,19 @@ public class MessageRetryRunner extends BaseDaemonThread {
                 }
 
                 long now = System.currentTimeMillis();
-                for (Map.Entry<Integer, Map<CommandType, BaseCommand>> 
taskEntry : needToRetryMessages.entrySet()) {
+                Iterator<Map.Entry<Integer, Map<CommandType, BaseCommand>>> 
iterator =
+                        needToRetryMessages.entrySet().iterator();
+                while (iterator.hasNext()) {
+                    Map.Entry<Integer, Map<CommandType, BaseCommand>> 
taskEntry = iterator.next();
                     Integer taskInstanceId = taskEntry.getKey();
+                    Map<CommandType, BaseCommand> retryMessageMap = 
taskEntry.getValue();
+                    if (retryMessageMap.isEmpty()) {
+                        iterator.remove();
+                        continue;
+                    }
                     LoggerUtils.setTaskInstanceIdMDC(taskInstanceId);
                     try {
-                        for (Map.Entry<CommandType, BaseCommand> messageEntry 
: taskEntry.getValue().entrySet()) {
+                        for (Map.Entry<CommandType, BaseCommand> messageEntry 
: retryMessageMap.entrySet()) {
                             CommandType messageType = messageEntry.getKey();
                             BaseCommand message = messageEntry.getValue();
                             if (now - message.getMessageSendTime() > 
MESSAGE_RETRY_WINDOW) {