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