From 5264243e5b5b37c79687523444afbce870f23885 Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Thu, 10 Aug 2023 17:05:42 +0800 Subject: [PATCH 1/4] follower do not need to retrieve load data in zk. --- .../impl/ModularLoadManagerImpl.java | 47 +++++++++++++++---- .../pulsar/metadata/api/MetadataCache.java | 5 ++ .../api/coordination/LockManager.java | 5 ++ .../coordination/impl/LockManagerImpl.java | 5 ++ 4 files changed, 52 insertions(+), 10 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java index 8882c2cc6c56b..ea52a1d6f0f0a 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java @@ -34,9 +34,9 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletionException; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.locks.Lock; @@ -173,7 +173,7 @@ public class ModularLoadManagerImpl implements ModularLoadManager { private PulsarService pulsar; // Executor service used to update broker data. - private final ExecutorService executors; + private final ScheduledExecutorService scheduler; // check if given broker can load persistent/non-persistent topic private final BrokerTopicLoadingPredicate brokerTopicLoadingPredicate; @@ -198,6 +198,8 @@ public class ModularLoadManagerImpl implements ModularLoadManager { private final Lock lock = new ReentrantLock(); private final Set knownBrokers = new HashSet<>(); private Map bundleBrokerAffinityMap; + // flag to indicate whether the load data is old. + private volatile boolean oldFlag = false; /** * Initializes fields which do not depend on PulsarService. initialize(PulsarService) should subsequently be called. @@ -213,7 +215,7 @@ public ModularLoadManagerImpl() { loadData = new LoadData(); loadSheddingPipeline = new ArrayList<>(); preallocatedBundleToBroker = new ConcurrentHashMap<>(); - executors = Executors.newSingleThreadExecutor( + scheduler = Executors.newSingleThreadScheduledExecutor( new ExecutorProvider.ExtendedThreadFactory("pulsar-modular-load-manager")); this.brokerToFailureDomainMap = new HashMap<>(); this.bundleBrokerAffinityMap = new ConcurrentHashMap<>(); @@ -275,24 +277,32 @@ public void initialize(final PulsarService pulsar) { // register listeners for domain changes pulsar.getPulsarResources().getClusterResources().getFailureDomainResources() .registerListener(__ -> { - executors.execute( + scheduler.execute( () -> LoadManagerShared.refreshBrokerToFailureDomainMap(pulsar, brokerToFailureDomainMap)); }); loadSheddingPipeline.add(createLoadSheddingStrategy()); } + private boolean isLeader() { + return pulsar.getLeaderElectionService() != null + && pulsar.getLeaderElectionService().isLeader(); + } + public void handleDataNotification(Notification t) { if (t.getPath().startsWith(LoadManager.LOADBALANCE_BROKERS_ROOT)) { brokersData.listLocks(LoadManager.LOADBALANCE_BROKERS_ROOT) .thenAccept(brokers -> { reapDeadBrokerPreallocations(brokers); }); + oldFlag = true; - try { - executors.execute(ModularLoadManagerImpl.this::updateAll); - } catch (RejectedExecutionException e) { - // Executor is shutting down + if (isLeader()) { + try { + scheduler.submit(ModularLoadManagerImpl.this::updateAll); + } catch (RejectedExecutionException e) { + // Executor is shutting down + } } } } @@ -479,6 +489,7 @@ public void updateAll() { updateBundleData(); // broker has latest load-report: check if any bundle requires split checkNamespaceBundleSplit(); + oldFlag = false; } private synchronized void cleanupDeadBrokersData() { @@ -834,6 +845,7 @@ private void updateBundleSplitMetrics(int bundlesSplit) { /** * As the leader broker, find a suitable broker for the assignment of the given bundle. + * If leader broker is inactive, decision will be made by the current broker. * * @param serviceUnit * ServiceUnitId for the bundle. @@ -841,6 +853,22 @@ private void updateBundleSplitMetrics(int bundlesSplit) { */ @Override public Optional selectBrokerForAssignment(final ServiceUnitId serviceUnit) { + if (oldFlag) { + log.info("LoadData is old, need to update loadData before make decision"); + updateAll(); + if (!isLeader()) { + log.info("Not leader broker, need to evict the cache after some times" + + " in case of updating loadData continuously."); + // invalidate all cache after some times in case of non-leader broker requesting + // metadata store continuously. + scheduler.schedule(() -> { + bundlesCache.invalidateAll(); + brokersData.releaseAllResourcesInCache(); + }, Math.min(conf.getLoadBalancerResourceQuotaUpdateIntervalMinutes(), + conf.getLoadBalancerReportUpdateMaxIntervalMinutes()), TimeUnit.MINUTES); + } + } + // Use brokerCandidateCache as a lock to reduce synchronization. long startTime = System.nanoTime(); @@ -1004,7 +1032,6 @@ public void start() throws PulsarServerException { timeAverageBrokerDataCache.readModifyUpdateOrCreate(timeAverageZPath, __ -> new TimeAverageBrokerData()).join(); - updateAll(); } catch (Exception e) { log.error("Unable to acquire lock for broker: [{}]", brokerZnodePath, e); throw new PulsarServerException(e); @@ -1019,7 +1046,7 @@ public void start() throws PulsarServerException { */ @Override public void stop() throws PulsarServerException { - executors.shutdownNow(); + scheduler.shutdownNow(); try { brokersData.close(); diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/api/MetadataCache.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/api/MetadataCache.java index 94da382b74dcf..c97e6cab4a8a2 100644 --- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/api/MetadataCache.java +++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/api/MetadataCache.java @@ -154,4 +154,9 @@ public interface MetadataCache { * @param path the path of the object in the metadata store */ void refresh(String path); + + /** + * Force the invalidation of all objects in the metadata cache. + */ + public void invalidateAll(); } diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/api/coordination/LockManager.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/api/coordination/LockManager.java index cb6601b170135..35711db9f4eb0 100644 --- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/api/coordination/LockManager.java +++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/api/coordination/LockManager.java @@ -92,4 +92,9 @@ public interface LockManager extends AutoCloseable { */ CompletableFuture asyncClose(); + /** + * release all resources in cache. + * @return + */ + void releaseAllResourcesInCache(); } diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/LockManagerImpl.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/LockManagerImpl.java index 4da6b7998a0c4..724de92509b9e 100644 --- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/LockManagerImpl.java +++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/LockManagerImpl.java @@ -179,4 +179,9 @@ public CompletableFuture asyncClose() { .map(ResourceLock::release) .collect(Collectors.toList())); } + + @Override + public void releaseAllResourcesInCache() { + cache.invalidateAll(); + } } From e806165cbf733c0ec60551553e1a377fcda403a5 Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Thu, 10 Aug 2023 17:52:00 +0800 Subject: [PATCH 2/4] fix checkstyle. --- .../main/java/org/apache/pulsar/metadata/api/MetadataCache.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/api/MetadataCache.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/api/MetadataCache.java index c97e6cab4a8a2..bf8f5dddc1f3a 100644 --- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/api/MetadataCache.java +++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/api/MetadataCache.java @@ -158,5 +158,5 @@ public interface MetadataCache { /** * Force the invalidation of all objects in the metadata cache. */ - public void invalidateAll(); + void invalidateAll(); } From 36b94a3f140398810ed94ff07b0b792efbd33d9f Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Thu, 10 Aug 2023 18:25:36 +0800 Subject: [PATCH 3/4] fix checkstyle. --- .../broker/loadbalance/impl/ModularLoadManagerImpl.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java index ea52a1d6f0f0a..5eab83be4cc69 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java @@ -857,8 +857,8 @@ public Optional selectBrokerForAssignment(final ServiceUnitId serviceUni log.info("LoadData is old, need to update loadData before make decision"); updateAll(); if (!isLeader()) { - log.info("Not leader broker, need to evict the cache after some times" + - " in case of updating loadData continuously."); + log.info("Not leader broker, need to evict the cache after some times" + + " in case of updating loadData continuously."); // invalidate all cache after some times in case of non-leader broker requesting // metadata store continuously. scheduler.schedule(() -> { From 7ee505c18e8dafb21fa8044359a04b92eacc972b Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Fri, 11 Aug 2023 14:22:52 +0800 Subject: [PATCH 4/4] change. --- .../impl/ModularLoadManagerImpl.java | 24 +++++++++---------- 1 file changed, 11 insertions(+), 13 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java index 5eab83be4cc69..1d27af6e0c4ff 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java @@ -853,20 +853,18 @@ private void updateBundleSplitMetrics(int bundlesSplit) { */ @Override public Optional selectBrokerForAssignment(final ServiceUnitId serviceUnit) { - if (oldFlag) { - log.info("LoadData is old, need to update loadData before make decision"); + if (!isLeader() && oldFlag) { + log.info("Make decision on follower broker, but the LoadData is old," + + " need to update LoadData before make decision"); updateAll(); - if (!isLeader()) { - log.info("Not leader broker, need to evict the cache after some times" - + " in case of updating loadData continuously."); - // invalidate all cache after some times in case of non-leader broker requesting - // metadata store continuously. - scheduler.schedule(() -> { - bundlesCache.invalidateAll(); - brokersData.releaseAllResourcesInCache(); - }, Math.min(conf.getLoadBalancerResourceQuotaUpdateIntervalMinutes(), - conf.getLoadBalancerReportUpdateMaxIntervalMinutes()), TimeUnit.MINUTES); - } + // invalidate all cache after some times in case of non-leader broker requesting + // metadata store continuously. + scheduler.schedule(() -> { + bundlesCache.invalidateAll(); + brokersData.releaseAllResourcesInCache(); + oldFlag = true; + }, Math.min(conf.getLoadBalancerResourceQuotaUpdateIntervalMinutes(), + conf.getLoadBalancerReportUpdateMaxIntervalMinutes()), TimeUnit.MINUTES); } // Use brokerCandidateCache as a lock to reduce synchronization.