Avoid potentially blocking method during topic ownership check (apach…

…e#4190)
yciabaud · May 2, 2019 · 10b8811 · 10b8811
1 parent 30674dd
commit 10b8811
Show file tree

Hide file tree

Showing 2 changed files with 15 additions and 1 deletion.
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
@@ -165,6 +165,11 @@ public CompletableFuture<NamespaceBundle> getBundleAsync(TopicName topic) {
                 .thenApply(bundles -> bundles.findBundle(topic));
     }
 
+    public Optional<NamespaceBundle> getBundleIfPresent(TopicName topicName) throws Exception {
+        Optional<NamespaceBundles> bundles = bundleFactory.getBundlesIfPresent(topicName.getNamespaceObject());
+        return bundles.map(b -> b.findBundle(topicName));
+    }
+
     public NamespaceBundle getBundle(TopicName topicName) throws Exception {
         return bundleFactory.getBundles(topicName.getNamespaceObject()).findBundle(topicName);
     }
@@ -824,7 +829,12 @@ private CompletableFuture<Boolean> isTopicOwnedAsync(TopicName topic) {
     }
 
     private boolean isTopicOwned(TopicName topicName) throws Exception {
-        return ownershipCache.getOwnedBundle(getBundle(topicName)) != null;
+        Optional<NamespaceBundle> bundle = getBundleIfPresent(topicName);
+        if (!bundle.isPresent()) {
+            return false;
+        } else {
+            return ownershipCache.getOwnedBundle(bundle.get()) != null;
+        }
     }
 
     public void removeOwnedServiceUnit(NamespaceName nsName) throws Exception {

diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/common/naming/NamespaceBundleFactory.java b/pulsar-broker/src/main/java/org/apache/pulsar/common/naming/NamespaceBundleFactory.java
@@ -155,6 +155,10 @@ public NamespaceBundles getBundles(NamespaceName nsname) throws Exception {
         return bundlesCache.synchronous().get(nsname);
     }
 
+    public Optional<NamespaceBundles> getBundlesIfPresent(NamespaceName nsname) throws Exception {
+        return Optional.ofNullable(bundlesCache.synchronous().getIfPresent(nsname));
+    }
+
     public NamespaceBundle getBundle(NamespaceName nsname, Range<Long> hashRange) {
         return new NamespaceBundle(nsname, hashRange, this);
     }