[fix][broker] Avoid ConcurrentModificationException for ModularLoadManagerImpl.cleanupDeadBrokersData()
@@ -485,10 +485,11 @@ public class ModularLoadManagerImpl implements ModularLoadManager {
private void cleanupDeadBrokersData() {
final Set<String> activeBrokers = getAvailableBrokers();
- Collection<String> newBrokers = CollectionUtils.subtract(activeBrokers, knownBrokers);
- knownBrokers.addAll(newBrokers);
- Collection<String> deadBrokers = CollectionUtils.subtract(knownBrokers, activeBrokers);
- knownBrokers.removeAll(deadBrokers);
+ 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);
if (pulsar.getLeaderElectionService() != null
&& pulsar.getLeaderElectionService().isLeader()) {
deadBrokers.forEach(this::deleteTimeAverageDataFromMetadataStoreAsync);
