test/DolphinScheduler
1
0
Fork 0
mirror of https://gitee.com/dolphinscheduler/DolphinScheduler.git synced 2024-12-02 04:08:31 +08:00
Code Issues Actions 10 Packages Projects Releases Wiki Activity

Fix memory leak in worker due to message retry map (#12878)

Browse Source
This commit is contained in:
Wenjun Ruan 2022-11-13 20:08:38 +08:00 committed by GitHub
parent feb077035e
commit 403c6a6bb6
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 11 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/message/MessageRetryRunner.java
View File

@ -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) {