PIP-45: Implement load managers locks using coordination service #10391

merlimat · 2021-04-27T01:15:00Z

Motivation

Implemented load manager lock and load report using ResourceLock from CoordinationService instead of direct ZK access.

eolivelli

Overall looks good to me.
I left a few comments and suggestions, mostly for some follow up work

pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/NoopLoadManager.java

...r-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java

eolivelli · 2021-04-29T06:39:21Z

...r-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java

@@ -363,24 +343,27 @@ private void reapDeadBrokerPreallocations(Set<String> aliveBrokers) {
    @Override
    public Set<String> getAvailableBrokers() {
        try {
-            return availableActiveBrokers.get();
+            return new TreeSet<>(brokersData.listLocks(LoadManager.LOADBALANCE_BROKERS_ROOT).get());
        } catch (Exception e) {
            log.warn("Error when trying to get active brokers", e);


in case of ZK problem, do we see lots of stacktraces written in logs ?
does it make sense to strip out the stacktrace here ?

(we should also deal with InterruptedException), as this method can be called from anywhere

...r-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java

...ar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/SimpleLoadManagerImpl.java

rdhabalia

LGTM.. few minor comments.

rdhabalia · 2021-04-29T18:23:56Z

...r-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java

@@ -363,24 +343,27 @@ private void reapDeadBrokerPreallocations(Set<String> aliveBrokers) {
    @Override
    public Set<String> getAvailableBrokers() {
        try {
-            return availableActiveBrokers.get();
+            return new TreeSet<>(brokersData.listLocks(LoadManager.LOADBALANCE_BROKERS_ROOT).get());


is there any requirement for ordering? can't we use HashSet?

It was mostly that we used a TreeSet before for all the ZK returned children. In this case, we don't really care. I'll change it to HashSet

rdhabalia · 2021-04-29T19:08:41Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/NoopLoadManager.java

-
-        } catch (Exception e) {
-            throw new PulsarServerException(e);
+            lockManager.acquireLock(brokerReportPath, localData).join();


can we add log here as it's a blocking call and we can use the log to troubleshoot if server is taking time to come-up or not coming up..

rdhabalia · 2021-04-29T19:12:51Z

...r-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java

-                bundleData = readJson(zkClient.getData(bundleZPath, null, null), BundleData.class);
-            } else if (zkClient.exists(quotaZPath, null) != null) {
-                final ResourceQuota quota = readJson(zkClient.getData(quotaZPath, null, null), ResourceQuota.class);
+            Optional<BundleData> optBundleData = bundlesCache.get(getBundleDataPath(bundle)).join();


can we use time bounded get(timeout) instead join to break the deadlock ?

Rather than putting get(Timeout), I prefer having timeouts on the async operations in the metadata store implementation.

umm.. you mean it will be metadata store-impl's responsibility to complete the returned CompletableFuture in certain time? ZK-Impl depends on ZK-Client and returns future. So, if we put timeout responsibility to impl class then Impl has to add extra handling to return time based future. this doesn't seem feasible solution.

Yes, we can add the timeouts in the AbstractMetadataStoreImpl. Otherwise we only have timeouts for sync calls, but in most places we're doing (and we should do more) the async calls.

rdhabalia · 2021-04-29T19:16:55Z

...r-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java

-    // ZooKeeper cache of the local broker data, stored in LoadManager.LOADBALANCE_BROKER_ROOT.
-    private ZooKeeperDataCache<LocalBrokerData> brokerDataCache;
+    // Cache of the local broker data, stored in LoadManager.LOADBALANCE_BROKER_ROOT.
+    private LockManager<LocalBrokerData> brokersData;


Can we use MetadataCache instead LockManager? LocalBrokerData is a metadata stored in localzk right now and it doesn't need to acquire lock.

It is a lock though, in the sense that we're acquiring an ephemeral z-node with the broker name. If there's already a z-node there we need to handle the scenario and that's what the ResourceLock does

rdhabalia · 2021-04-29T19:18:41Z

...r-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java

-                    ZkUtils.createFullPathOptimistic(zkClient, brokerZnodePath, localData.getJsonBytes(),
-                            ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
-                }
+                brokerDataLock.updateValue(localData).join();


same here, we should use time bounded get instead join . this can cause deadlock in system if future never completes.

rdhabalia · 2021-04-29T19:21:00Z

pulsar-metadata/src/main/java/org/apache/pulsar/metadata/api/MetadataCache.java

+     *            the object to insert in metadata store
+     * @return a future to track the completion of the operation
+     */
+    CompletableFuture<Void> updateOrCreate(String path, T value);


we have used readModifyUpdateOrCreate at few places for this usecase.

Good point. the usage here is slightly different in that we're not caring for the existing value, but we can avoid adding a new method just for that.

lhotari · 2021-05-11T16:10:15Z

I wonder if this change has introduced flakiness to the LoadBalancerTest, reported as #10537 . @merlimat would you be able to check?

…native#488) This PR upgrades pulsar dependency to 2.8.0-rc-202105092228, which has two major API changes. apache/pulsar#10391 changed `LoadManager` API so that `MetadataCache` is used instead of `ZookeeperCache` in this PR. apache/pulsar#7406 changed the throttling strategy. However, currently KoP is different from Pulsar that the produce and its callback may be in different threads. KoP calls `PersistentTopic#publishMessages` in a callback of `KafkaTopicManager#getTopic` if the returned future is not completed immediately. Otherwise, it's called just in the I/O thread. Therefore, here we still use a **channel based** publish bytes stats for throttling, while apache/pulsar#7406 uses a **thread based** publish bytes stats. The other refactors are: 1. Change the throttling related fields from `InternalServerCnx` to `KafkaRequestHandler`. 2. Use `BrokerService#getPausedConnections` to check if the channel's auto read is disabled and modify the tests as well.

merlimat · 2021-05-11T16:50:26Z

@lhotari That's probably related, taking a look

This PR upgrades pulsar dependency to 2.8.0-rc-202105092228, which has two major API changes. apache/pulsar#10391 changed `LoadManager` API so that `MetadataCache` is used instead of `ZookeeperCache` in this PR. apache/pulsar#7406 changed the throttling strategy. However, currently KoP is different from Pulsar that the produce and its callback may be in different threads. KoP calls `PersistentTopic#publishMessages` in a callback of `KafkaTopicManager#getTopic` if the returned future is not completed immediately. Otherwise, it's called just in the I/O thread. Therefore, here we still use a **channel based** publish bytes stats for throttling, while apache/pulsar#7406 uses a **thread based** publish bytes stats. The other refactors are: 1. Change the throttling related fields from `InternalServerCnx` to `KafkaRequestHandler`. 2. Use `BrokerService#getPausedConnections` to check if the channel's auto read is disabled and modify the tests as well. * Fix LoadManager interface * Refactor publish throttling * Remove ZookeeperCache usage

merlimat added the type/enhancement The enhancements for the existing features or docs. e.g. reduce memory usage of the delayed messages label Apr 27, 2021

merlimat added this to the 2.8.0 milestone Apr 27, 2021

merlimat requested review from sijie, rdhabalia, jerrypeng and codelipenghui April 27, 2021 01:15

merlimat self-assigned this Apr 27, 2021

PIP-45: Implement load managers locks using coordination service

3b9b63b

merlimat force-pushed the load-manager branch from 87ce114 to 3b9b63b Compare April 27, 2021 01:30

merlimat added 6 commits April 27, 2021 14:49

Handle executor rejected exception when shutting down

6979bf0

Merge remote-tracking branch 'apache/master' into load-manager

f30530e

Fixed SimpleLoadManager still updating the load report on zk client

54c35c2

Merge remote-tracking branch 'apache/master' into load-manager

68448ea

Removed unused import

16745b5

Fixed imports order

137368c

eolivelli approved these changes Apr 29, 2021

View reviewed changes

Simplified listLocks calls

3ccc1e8

eolivelli approved these changes Apr 29, 2021

View reviewed changes

rdhabalia reviewed Apr 29, 2021

View reviewed changes

merlimat added 2 commits April 29, 2021 12:30

Merge remote-tracking branch 'apache/master' into load-manager

1eb1fa8

Addressed comments

f1fbc55

rdhabalia approved these changes Apr 29, 2021

View reviewed changes

merlimat added 5 commits April 29, 2021 13:36

Removed updateOrCreate

c580e35

Remove additional put method that's not needed anymore

cc261bf

Removed unused import

7a24f72

Fixed LoadBalanceTest

2601b61

Fixed ModularLoadManagerImplTest

2e39024

merlimat merged commit 1579b0f into apache:master Apr 30, 2021

merlimat deleted the load-manager branch April 30, 2021 22:13

lhotari mentioned this pull request May 10, 2021

lh use conscrypt for jetty lhotari/pulsar#39

Closed

BewareMyPower mentioned this pull request May 10, 2021

Fix publish throttling and LoadManager API for pulsar upgrade streamnative/kop#488

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

PIP-45: Implement load managers locks using coordination service #10391

PIP-45: Implement load managers locks using coordination service #10391

merlimat commented Apr 27, 2021

eolivelli left a comment

eolivelli Apr 29, 2021

rdhabalia left a comment

rdhabalia Apr 29, 2021

merlimat Apr 29, 2021

rdhabalia Apr 29, 2021 •

edited

Loading

merlimat Apr 29, 2021

rdhabalia Apr 29, 2021

merlimat Apr 29, 2021

rdhabalia Apr 29, 2021

merlimat Apr 29, 2021

rdhabalia Apr 29, 2021

merlimat Apr 29, 2021

rdhabalia Apr 29, 2021

rdhabalia Apr 29, 2021

merlimat Apr 29, 2021

lhotari commented May 11, 2021

merlimat commented May 11, 2021

PIP-45: Implement load managers locks using coordination service #10391

PIP-45: Implement load managers locks using coordination service #10391

Conversation

merlimat commented Apr 27, 2021

Motivation

eolivelli left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rdhabalia left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rdhabalia Apr 29, 2021 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lhotari commented May 11, 2021

merlimat commented May 11, 2021

rdhabalia Apr 29, 2021 •

edited

Loading