From e2c1cc057918c8c70fd468d21e9ed727547ddc56 Mon Sep 17 00:00:00 2001 From: guoshupei <15764973965@163.com> Date: Tue, 29 Mar 2022 22:46:02 +0800 Subject: [PATCH] [Fix-9222] [master] Support for deploying multiple masters on one node (#9240) This closes #9222 Co-authored-by: guoshupei --- .../server/master/registry/ServerNodeManager.java | 15 +++++++++++---- .../service/queue/MasterPriorityQueue.java | 10 ++++++---- 2 files changed, 17 insertions(+), 8 deletions(-) diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/registry/ServerNodeManager.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/registry/ServerNodeManager.java index 7f11576d13..0666ca82a0 100644 --- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/registry/ServerNodeManager.java +++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/registry/ServerNodeManager.java @@ -31,6 +31,7 @@ import org.apache.dolphinscheduler.registry.api.Event; import org.apache.dolphinscheduler.registry.api.Event.Type; import org.apache.dolphinscheduler.registry.api.SubscribeListener; import org.apache.dolphinscheduler.remote.utils.NamedThreadFactory; +import org.apache.dolphinscheduler.server.master.config.MasterConfig; import org.apache.dolphinscheduler.service.queue.MasterPriorityQueue; import org.apache.dolphinscheduler.service.registry.RegistryClient; @@ -124,6 +125,12 @@ public class ServerNodeManager implements InitializingBean { @Autowired private AlertDao alertDao; + /** + * master config + */ + @Autowired + private MasterConfig masterConfig; + private static volatile int MASTER_SLOT = 0; private static volatile int MASTER_SIZE = 0; @@ -338,18 +345,18 @@ public class ServerNodeManager implements InitializingBean { private void syncMasterNodes(Collection nodes, List masterNodes) { masterLock.lock(); try { - String host = NetUtils.getHost(); + String addr = NetUtils.getAddr(NetUtils.getHost(), masterConfig.getListenPort()); this.masterNodes.addAll(nodes); this.masterPriorityQueue.clear(); this.masterPriorityQueue.putList(masterNodes); - int index = masterPriorityQueue.getIndex(host); + int index = masterPriorityQueue.getIndex(addr); if (index >= 0) { MASTER_SIZE = nodes.size(); MASTER_SLOT = index; } else { - logger.warn("current host:{} is not in active master list", host); + logger.warn("current addr:{} is not in active master list", addr); } - logger.info("update master nodes, master size: {}, slot: {}", MASTER_SIZE, MASTER_SLOT); + logger.info("update master nodes, master size: {}, slot: {}, addr: {}", MASTER_SIZE, MASTER_SLOT, addr); } finally { masterLock.unlock(); } diff --git a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/queue/MasterPriorityQueue.java b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/queue/MasterPriorityQueue.java index 66385db048..73d420ccf0 100644 --- a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/queue/MasterPriorityQueue.java +++ b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/queue/MasterPriorityQueue.java @@ -18,6 +18,7 @@ package org.apache.dolphinscheduler.service.queue; import org.apache.dolphinscheduler.common.model.Server; +import org.apache.dolphinscheduler.common.utils.NetUtils; import java.util.Comparator; import java.util.HashMap; @@ -83,17 +84,18 @@ public class MasterPriorityQueue implements TaskPriorityQueue { int index = 0; while (iterator.hasNext()) { Server server = iterator.next(); - hostIndexMap.put(server.getHost(), index); + String addr = NetUtils.getAddr(server.getHost(), server.getPort()); + hostIndexMap.put(addr, index); index += 1; } } - public int getIndex(String host) { - if (!hostIndexMap.containsKey(host)) { + public int getIndex(String addr) { + if (!hostIndexMap.containsKey(addr)) { return -1; } - return hostIndexMap.get(host); + return hostIndexMap.get(addr); } /**