You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@dolphinscheduler.apache.org by we...@apache.org on 2022/11/14 04:01:36 UTC
[dolphinscheduler] branch 3.0.2-prepare updated: Fix memory leak in worker due to message retry map (#12878) (#12887)
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 <we...@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) {