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

[caishunfeng]

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

caishunfeng 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 403c6a6bb6 Fix memory leak in worker due to message retry map (#12878)
403c6a6bb6 is described below

commit 403c6a6bb6ec9fc380ffd763301891ad57029e2b
Author: Wenjun Ruan <wen...@apache.org>
AuthorDate: Sun Nov 13 20:08:38 2022 +0800

    Fix memory leak in worker due to message retry map (#12878)
---
 .../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 fec345ecf0..6f11ab4d0d 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
@@ -28,6 +28,7 @@ import org.apache.commons.collections.MapUtils;
 
 import java.time.Duration;
 import java.util.HashMap;
+import java.util.Iterator;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 
@@ -108,11 +109,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) {