[improve][broker] PIP-192: Implement broker registry for new load manager #18810

Demogorgon314 · 2022-12-08T01:24:09Z

Master Issue: #16691

Motivation

We will start raising PRs to implement PIP-192, #16691

Modifications

This PR should not impact the existing broker load balance behavior.

For the PIP-192 project, this PR

Implemented broker registry.
Added units test.

Documentation

doc
doc-required
doc-not-needed
doc-complete

Matching PR in forked repository

PR in forked repository: Demogorgon314#7

...broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/BrokerRegistryImpl.java

...broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/BrokerRegistryTest.java

...broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/BrokerRegistryImpl.java

heesung-sn · 2022-12-30T00:48:30Z

...broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/BrokerRegistryImpl.java

+            Map<String, BrokerLookupData> map = new ConcurrentHashMap<>();
+            List<CompletableFuture<Void>> futures = new ArrayList<>();
+            for (String brokerId : availableBrokers) {
+                futures.add(this.lookupAsync(brokerId).thenAccept(lookupDataOpt ->


Let's check if lookupAsync is complete or not.

If not, we probably want to print a log with the i/o waiting time.

Can you explain it in detail?

In most cases, the returned future of this.lookupAsync(brokerId) should be Completed, as we expect the lookup is mainly from the cache. However, if the returned future is Incomplete, we probably need to measure the latency of the lookup i/o to the metadata store.

In general, I think every i/o to the dependent services needs to be tracked(preferably as metrics). I am not sure about this practice in the Pulsar repo, though.

I see. Thanks for the explanation.

...broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/BrokerRegistryImpl.java

codelipenghui · 2022-12-30T09:31:18Z

...broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/BrokerRegistryImpl.java

+        if (started.get()) {
+            return;
+        }
+        pulsar.getLocalMetadataStore().registerListener(this::handleMetadataStoreNotification);
+        try {
+            this.register();
+            this.started.set(true);
+        } catch (MetadataStoreException e) {
+            throw new PulsarServerException(e);
+        }


Suggested change

if (started.get()) {

return;

}

pulsar.getLocalMetadataStore().registerListener(this::handleMetadataStoreNotification);

try {

this.register();

this.started.set(true);

} catch (MetadataStoreException e) {

throw new PulsarServerException(e);

}

if (started.compareAndSet(false, true)) {

pulsar.getLocalMetadataStore().registerListener(this::handleMetadataStoreNotification);

try {

this.register();

} catch (MetadataStoreException e) {

throw new PulsarServerException(e);

}

}

codelipenghui · 2022-12-30T09:34:10Z

...broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/BrokerRegistryImpl.java

+    public synchronized void register() throws MetadataStoreException {
+        if (!registered) {
+            try {
+                this.brokerLookupDataLock = brokerLookupDataLockManager.acquireLock(keyPath(brokerId), brokerLookupData)
+                        .get(conf.getMetadataStoreOperationTimeoutSeconds(), TimeUnit.SECONDS);
+                registered = true;
+            } catch (InterruptedException | ExecutionException | TimeoutException e) {
+                throw MetadataStoreException.unwrap(e);
+            }
+        }
+    }
+
+    @Override
+    public synchronized void unregister() throws MetadataStoreException {
+        if (registered) {
+            try {
+                brokerLookupDataLock.release().get(conf.getMetadataStoreOperationTimeoutSeconds(), TimeUnit.SECONDS);
+                registered = false;
+            } catch (CompletionException | InterruptedException | ExecutionException | TimeoutException e) {
+                throw MetadataStoreException.unwrap(e);
+            }
+        }
+    }


Is it possible to change registered to AtomicBoolean? So that we don't need the synchronized

I used AtomicBoolean in the first version. #18810 (comment)

codelipenghui · 2022-12-30T09:36:45Z

...broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/BrokerRegistryImpl.java

+    public CompletableFuture<List<String>> getAvailableBrokersAsync() {
+        return brokerLookupDataLockManager.listLocks(LOOKUP_DATA_PATH).thenApply(Lists::newArrayList);
+    }
+
+    @Override
+    public CompletableFuture<Optional<BrokerLookupData>> lookupAsync(String broker) {
+        return brokerLookupDataLockManager.readLock(keyPath(broker));
+    }
+
+    public CompletableFuture<Map<String, BrokerLookupData>> getAvailableBrokerLookupDataAsync() {
+        return this.getAvailableBrokersAsync().thenCompose(availableBrokers -> {
+            Map<String, BrokerLookupData> map = new ConcurrentHashMap<>();
+            List<CompletableFuture<Void>> futures = new ArrayList<>();
+            for (String brokerId : availableBrokers) {
+                futures.add(this.lookupAsync(brokerId).thenAccept(lookupDataOpt -> {
+                    if (lookupDataOpt.isPresent()) {
+                        map.put(brokerId, lookupDataOpt.get());
+                    } else {
+                        log.warn("Got an empty lookup data, brokerId: {}", brokerId);
+                    }
+                }));
+            }
+            return FutureUtil.waitForAll(futures).thenApply(__ -> map);
+        });
+    }
+
+    public void listen(BiConsumer<String, NotificationType> listener) {
+        this.listeners.add(listener);
+    }


It's better to check if the registry is started or not.

codelipenghui · 2022-12-30T09:37:45Z

...broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/BrokerRegistryImpl.java

+        if (!started.get()) {
+            return;
+        }
+        try {
+            this.unregister();
+            brokerLookupDataLockManager.close();
+            scheduler.shutdown();
+            this.listeners.clear();
+            started.set(false);
+        } catch (Exception ex) {
+            if (ex.getCause() instanceof MetadataStoreException.NotFoundException) {
+                throw new PulsarServerException.NotFoundException(MetadataStoreException.unwrap(ex));
+            } else {
+                throw new PulsarServerException(MetadataStoreException.unwrap(ex));
+            }
+        }


Suggested change

if (!started.get()) {

return;

}

try {

this.unregister();

brokerLookupDataLockManager.close();

scheduler.shutdown();

this.listeners.clear();

started.set(false);

} catch (Exception ex) {

if (ex.getCause() instanceof MetadataStoreException.NotFoundException) {

throw new PulsarServerException.NotFoundException(MetadataStoreException.unwrap(ex));

} else {

throw new PulsarServerException(MetadataStoreException.unwrap(ex));

}

}

if (started.compareAndSet(true, false)) {

try {

this.unregister();

brokerLookupDataLockManager.close();

scheduler.shutdown();

this.listeners.clear();

} catch (Exception ex) {

if (ex.getCause() instanceof MetadataStoreException.NotFoundException) {

throw new PulsarServerException.NotFoundException(MetadataStoreException.unwrap(ex));

} else {

throw new PulsarServerException(MetadataStoreException.unwrap(ex));

}

}

}

Don't we need to reset started.set(false) when an exception is thrown?

We don't expect this register() / unregister() (or start() / close() in similar classes) to be called competitively many times from multiple threads. What's the practice in Pulsar for concurrency control in such cases? (I think the simple synchronized is enough in these cases because threads waiting happens less likely )

I noticed no synchronized for this method.

What's the practice in Pulsar for concurrency control in such cases?

Most places change to the CLOSING state from READY, and then to the CLOSED from CLOSING after all the tasks are done. And we can try multiple times to close the resource from CLOSING to CLOSED.

codelipenghui · 2022-12-30T09:40:28Z

...broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/BrokerRegistryImpl.java

+        if (!isVerifiedNotification(t)) {
+            return;
+        }
+        try {
+            if (log.isDebugEnabled()) {
+                log.debug("Handle notification: [{}]", t);
+            }
+            if (listeners.isEmpty()) {
+                return;
+            }
+            this.scheduler.submit(() -> {
+                String brokerId = t.getPath().substring(LOOKUP_DATA_PATH.length() + 1);
+                for (BiConsumer<String, NotificationType> listener : listeners) {
+                    listener.accept(brokerId, t.getType());
+                }
+            });
+        } catch (RejectedExecutionException e) {
+            // Executor is shutting down
+        }


We don't need to perform the listeners if the registry is closed, right?

codelipenghui · 2022-12-30T09:43:14Z

...broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/BrokerRegistryTest.java

+ */
+@Slf4j
+@Test(groups = "broker")
+public class BrokerRegistryTest {


Please also add tests for the behavior of a closed registry. Make sure the resources are released, the containers are cleaned up.

I have added a state field to store the registry state, PTAL.

Ping @codelipenghui

heesung-sn · 2023-01-11T08:13:55Z

nit:

Why not use LoadDataStore<BrokerLoadData> + TableView<BrokerLookupData> for this broker registry service?

LoadDataStore<BrokerLoadData>.keySet() can return available brokers.

TableView<BrokerLookupData>.get(broker) can return the pulsar version and other lookup info.

This will further reduce the metadata store dependency.

Sorry for this random idea, and I just wanted to share it here. We may need a separate discussion(a PIP) to think deeper if we agree with this direction.

Demogorgon314 · 2023-01-14T01:04:10Z

nit:

Why not use LoadDataStore<BrokerLoadData> + TableView<BrokerLookupData> for this broker registry service?

LoadDataStore<BrokerLoadData>.keySet() can return available brokers.

TableView<BrokerLookupData>.get(broker) can return the pulsar version and other lookup info.

This will further reduce the metadata store dependency.

Sorry for this random idea, and I just wanted to share it here. We may need a separate discussion(a PIP) to think deeper if we agree with this direction.

@heesung-sn The lookup data will not be too heavy for Zookeeper, right? Maybe we can do some tests to find out is necessary to use LoadDataStore to store the lookup data.

codelipenghui · 2023-01-17T04:41:34Z

The lookup data will not be too heavy for Zookeeper, right? Maybe we can do some tests to find out is necessary to use LoadDataStore to store the lookup data.

IMO, it's ok. One broker will have one znode to store the lookup data. And we also use the znode as a mutex lock to avoid the different brokers registering with the same lookup data.

BewareMyPower · 2023-02-01T03:19:09Z

org.apache.pulsar.broker.loadbalance.extensions.BrokerRegistryTest.MockLoadManager is not abstract and does not override abstract method setNamespaceBundleAffinity(java.lang.String,java.lang.String) in org.apache.pulsar.broker.loadbalance.LoadManager

Please fix the style check

Demogorgon314 · 2023-02-01T03:37:15Z

@BewareMyPower Fixed.

...broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/BrokerRegistryImpl.java

Technoboy-

LGTM

lhotari · 2023-02-02T12:43:08Z

This PR introduced a new flaky test #19401 .

Demogorgon314 added type/enhancement The enhancements for the existing features or docs. e.g. reduce memory usage of the delayed messages area/broker labels Dec 8, 2022

Demogorgon314 self-assigned this Dec 8, 2022

github-actions bot added the doc-not-needed Your PR changes do not impact docs label Dec 8, 2022

Demogorgon314 marked this pull request as ready for review December 8, 2022 07:31

Demogorgon314 mentioned this pull request Dec 8, 2022

PIP-192: New Pulsar Broker Load Balancer #16691

Closed

heesung-sn reviewed Dec 12, 2022

View reviewed changes

...broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/BrokerRegistryImpl.java Show resolved Hide resolved

Demogorgon314 requested a review from heesung-sn December 13, 2022 05:15

Demogorgon314 commented Dec 13, 2022

View reviewed changes

...broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/BrokerRegistryImpl.java Outdated Show resolved Hide resolved

heesung-sn approved these changes Dec 16, 2022

View reviewed changes

...broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/BrokerRegistryImpl.java Outdated Show resolved Hide resolved

...broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/BrokerRegistryImpl.java Outdated Show resolved Hide resolved

This was referenced Dec 22, 2022

[improve][broker] PIP-192: Implement broker version filter for new load manager #19023

Merged

[improve][broker] PIP-192: Implement broker version filter for new load manager Demogorgon314/pulsar#9

Open

Demogorgon314 requested review from BewareMyPower, codelipenghui, gaoran10, mattisonchao, merlimat and eolivelli December 28, 2022 08:05

heesung-sn reviewed Dec 30, 2022

View reviewed changes

heesung-sn mentioned this pull request Dec 30, 2022

[improve][broker] PIP-192: Implement extensible load manager #19102

Merged

4 tasks

codelipenghui added this to the 2.12.0 milestone Dec 30, 2022

codelipenghui reviewed Dec 30, 2022

View reviewed changes

Demogorgon314 requested review from codelipenghui and heesung-sn January 3, 2023 08:15

codelipenghui approved these changes Jan 17, 2023

View reviewed changes

Demogorgon314 added the ready-to-test label Jan 17, 2023

Demogorgon314 requested review from shibd and removed request for heesung-sn January 19, 2023 01:58

Demogorgon314 closed this Jan 31, 2023

Demogorgon314 reopened this Jan 31, 2023

Demogorgon314 added 14 commits February 1, 2023 11:20

Implement broker registry for new load manager

66b0d7c

Refactor

723f154

Cleanup the listeners first when close

3fe33e1

Add more unit tests

c6f8f13

Add more unit tests

ec723fb

Add test group

01335ab

Change the lookup address to broker id

582925b

Remove cache and add more tests

b66cd7f

Address reviewer's comments

48c1cbc

Add new available broker lookup data method

1d20a5f

Add log to warn empty lookup data

99ba81b

Use state field to store states.

0fa3392

Use try finally when close

d4c2326

Merge master into current branch

fdb9ebd

Demogorgon314 force-pushed the Demogorgon314/Implement_broker_registry branch from 38c0beb to fdb9ebd Compare February 1, 2023 03:27

Technoboy- reviewed Feb 1, 2023

View reviewed changes

...broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/BrokerRegistryImpl.java Outdated Show resolved Hide resolved

Technoboy- reviewed Feb 1, 2023

View reviewed changes

...broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/BrokerRegistryImpl.java Outdated Show resolved Hide resolved

Technoboy- approved these changes Feb 1, 2023

View reviewed changes

Address review comments

9c6d715

BewareMyPower approved these changes Feb 1, 2023

View reviewed changes

Technoboy- merged commit 8a4a8a6 into apache:master Feb 1, 2023

lhotari mentioned this pull request Feb 2, 2023

Flaky-test: BrokerRegistryTest.testRegisterAndLookup #19401

Closed

2 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[improve][broker] PIP-192: Implement broker registry for new load manager #18810

[improve][broker] PIP-192: Implement broker registry for new load manager #18810

Demogorgon314 commented Dec 8, 2022

heesung-sn Dec 30, 2022

Demogorgon314 Dec 30, 2022

heesung-sn Jan 11, 2023

Demogorgon314 Jan 13, 2023

codelipenghui Dec 30, 2022

codelipenghui Dec 30, 2022

Demogorgon314 Dec 30, 2022

codelipenghui Dec 30, 2022

Demogorgon314 Jan 3, 2023

codelipenghui Dec 30, 2022

heesung-sn Dec 30, 2022 •

edited

Loading

codelipenghui Jan 3, 2023

codelipenghui Dec 30, 2022

Demogorgon314 Jan 3, 2023

codelipenghui Dec 30, 2022

Demogorgon314 Jan 3, 2023

Demogorgon314 Jan 4, 2023

heesung-sn commented Jan 11, 2023 •

edited

Loading

Demogorgon314 commented Jan 14, 2023

codelipenghui commented Jan 17, 2023

BewareMyPower commented Feb 1, 2023

Demogorgon314 commented Feb 1, 2023

Technoboy- left a comment

lhotari commented Feb 2, 2023

[improve][broker] PIP-192: Implement broker registry for new load manager #18810

[improve][broker] PIP-192: Implement broker registry for new load manager #18810

Conversation

Demogorgon314 commented Dec 8, 2022

Motivation

Modifications

Documentation

Matching PR in forked repository

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

heesung-sn Dec 30, 2022 • 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

heesung-sn commented Jan 11, 2023 • edited Loading

Demogorgon314 commented Jan 14, 2023

codelipenghui commented Jan 17, 2023

BewareMyPower commented Feb 1, 2023

Demogorgon314 commented Feb 1, 2023

Technoboy- left a comment

Choose a reason for hiding this comment

lhotari commented Feb 2, 2023

heesung-sn Dec 30, 2022 •

edited

Loading

heesung-sn commented Jan 11, 2023 •

edited

Loading