You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by he...@apache.org on 2024/08/16 16:41:57 UTC
(pulsar) branch branch-3.1 updated: [improve][broker] Should notify bundle ownership listener onLoad event when ServiceUnitState start (ExtensibleLoadManagerImpl only) (#23152)
This is an automated email from the ASF dual-hosted git repository.
heesung pushed a commit to branch branch-3.1
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-3.1 by this push:
new 0d3358cfc09 [improve][broker] Should notify bundle ownership listener onLoad event when ServiceUnitState start (ExtensibleLoadManagerImpl only) (#23152)
0d3358cfc09 is described below
commit 0d3358cfc09ad854fb2e3885100712e9a9fa9cd3
Author: Kai Wang <kw...@apache.org>
AuthorDate: Sat Aug 17 00:17:55 2024 +0800
[improve][broker] Should notify bundle ownership listener onLoad event when ServiceUnitState start (ExtensibleLoadManagerImpl only) (#23152)
(cherry picked from commit 3053b647e0ca646b2df9f03815947104cd2e705f)
---
.../channel/ServiceUnitStateChannelImpl.java | 16 ++++++-
.../extensions/ExtensibleLoadManagerImplTest.java | 50 ++++++++++++++++++++++
2 files changed, 64 insertions(+), 2 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java
index c18938c0fdb..dfc36160d5d 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java
@@ -334,7 +334,8 @@ public class ServiceUnitStateChannelImpl implements ServiceUnitStateChannel {
"topicCompactionStrategyClassName",
ServiceUnitStateCompactionStrategy.class.getName()))
.create();
- tableview.listen((key, value) -> handle(key, value));
+ tableview.listen(this::handleEvent);
+ tableview.forEach(this::handleExisting);
var strategy = (ServiceUnitStateCompactionStrategy) TopicCompactionStrategy.getInstance(TABLE_VIEW_TAG);
if (strategy == null) {
String err = TABLE_VIEW_TAG + "tag TopicCompactionStrategy is null.";
@@ -662,7 +663,7 @@ public class ServiceUnitStateChannelImpl implements ServiceUnitStateChannel {
}).thenApply(__ -> null);
}
- private void handle(String serviceUnit, ServiceUnitStateData data) {
+ private void handleEvent(String serviceUnit, ServiceUnitStateData data) {
long totalHandledRequests = getHandlerTotalCounter(data).incrementAndGet();
if (debug()) {
log.info("{} received a handle request for serviceUnit:{}, data:{}. totalHandledRequests:{}",
@@ -688,6 +689,17 @@ public class ServiceUnitStateChannelImpl implements ServiceUnitStateChannel {
}
}
+ private void handleExisting(String serviceUnit, ServiceUnitStateData data) {
+ if (debug()) {
+ log.info("Loaded the service unit state data. serviceUnit: {}, data: {}", serviceUnit, data);
+ }
+ ServiceUnitState state = state(data);
+ if (state.equals(Owned) && isTargetBroker(data.dstBroker())) {
+ pulsar.getNamespaceService()
+ .onNamespaceBundleOwned(LoadManagerShared.getNamespaceBundle(pulsar, serviceUnit));
+ }
+ }
+
private static boolean isTransferCommand(ServiceUnitStateData data) {
if (data == null) {
return false;
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java
index 2feaabe9f99..59c21d01954 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java
@@ -345,6 +345,56 @@ public class ExtensibleLoadManagerImplTest extends ExtensibleLoadManagerImplBase
}
}
+ @Test(timeOut = 30 * 1000)
+ public void testNamespaceOwnershipListener() throws Exception {
+ Pair<TopicName, NamespaceBundle> topicAndBundle =
+ getBundleIsNotOwnByChangeEventTopic("test-namespace-ownership-listener");
+ TopicName topicName = topicAndBundle.getLeft();
+ NamespaceBundle bundle = topicAndBundle.getRight();
+
+ String broker = admin.lookups().lookupTopic(topicName.toString());
+ log.info("Assign the bundle {} to {}", bundle, broker);
+
+ checkOwnershipState(broker, bundle);
+
+ AtomicInteger onloadCount = new AtomicInteger(0);
+ AtomicInteger unloadCount = new AtomicInteger(0);
+
+ NamespaceBundleOwnershipListener listener = new NamespaceBundleOwnershipListener() {
+ @Override
+ public void onLoad(NamespaceBundle bundle) {
+ onloadCount.incrementAndGet();
+ }
+
+ @Override
+ public void unLoad(NamespaceBundle bundle) {
+ unloadCount.incrementAndGet();
+ }
+
+ @Override
+ public boolean test(NamespaceBundle namespaceBundle) {
+ return namespaceBundle.equals(bundle);
+ }
+ };
+ pulsar1.getNamespaceService().addNamespaceBundleOwnershipListener(listener);
+ pulsar2.getNamespaceService().addNamespaceBundleOwnershipListener(listener);
+
+ // There are a service unit state channel already started, when add listener, it will trigger the onload event.
+ Awaitility.await().untilAsserted(() -> {
+ assertEquals(onloadCount.get(), 1);
+ assertEquals(unloadCount.get(), 0);
+ });
+
+ ServiceUnitStateChannelImpl channel = new ServiceUnitStateChannelImpl(pulsar1);
+ channel.start();
+ Awaitility.await().untilAsserted(() -> {
+ assertEquals(onloadCount.get(), 2);
+ assertEquals(unloadCount.get(), 0);
+ });
+
+ channel.close();
+ }
+
private void checkOwnershipState(String broker, NamespaceBundle bundle)
throws ExecutionException, InterruptedException {
var targetLoadManager = secondaryLoadManager;