[improve][broker] Improve knownBrokers update in ModularLoadManagerIm…

…pl (apache#20196)
eugene-cheverda · Apr 28, 2023 · a362eaa · a362eaa
1 parent 7ccd1ba
commit a362eaa
Showing 1 changed file with 5 additions and 7 deletions.
diff --git a/...roker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java b/...roker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java
@@ -195,7 +195,7 @@ public class ModularLoadManagerImpl implements ModularLoadManager {
     private long unloadBundleCount = 0;
 
     private final Lock lock = new ReentrantLock();
-    private Set<String> knownBrokers = ConcurrentHashMap.newKeySet();
+    private final Set<String> knownBrokers = new HashSet<>();
     private Map<String, String> bundleBrokerAffinityMap;
 
     /**
@@ -480,13 +480,11 @@ public void updateAll() {
         checkNamespaceBundleSplit();
     }
 
-    private void cleanupDeadBrokersData() {
+    private synchronized void cleanupDeadBrokersData() {
         final Set<String> activeBrokers = getAvailableBrokers();
-        final Set<String> knownBrokersCopy = new HashSet<>(this.knownBrokers);
-        Collection<String> newBrokers = CollectionUtils.subtract(activeBrokers, knownBrokersCopy);
-        this.knownBrokers.addAll(newBrokers);
-        Collection<String> deadBrokers = CollectionUtils.subtract(knownBrokersCopy, activeBrokers);
-        this.knownBrokers.removeAll(deadBrokers);
+        Collection<String> deadBrokers = CollectionUtils.subtract(knownBrokers, activeBrokers);
+        this.knownBrokers.clear();
+        this.knownBrokers.addAll(activeBrokers);
         if (pulsar.getLeaderElectionService() != null
                 && pulsar.getLeaderElectionService().isLeader()) {
             deadBrokers.forEach(this::deleteTimeAverageDataFromMetadataStoreAsync);