[pulsar] branch branch-2.11 updated: [improve][broker] Remove locallyAcquiredLock when removeOwnership (#18197)

[te...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

technoboy pushed a commit to branch branch-2.11
in repository https://gitbox.apache.org/repos/asf/pulsar.git


The following commit(s) were added to refs/heads/branch-2.11 by this push:
     new d6e30b87cd3 [improve][broker] Remove locallyAcquiredLock when removeOwnership (#18197)
d6e30b87cd3 is described below

commit d6e30b87cd37f329136a6b1613427ca3715839d3
Author: Lei Zhiyuan <le...@gmail.com>
AuthorDate: Thu Oct 27 14:43:32 2022 +0800

    [improve][broker] Remove locallyAcquiredLock when removeOwnership (#18197)
---
 .../org/apache/pulsar/broker/namespace/OwnershipCache.java  | 13 ++++++++++++-
 .../apache/pulsar/broker/namespace/OwnershipCacheTest.java  |  1 +
 2 files changed, 13 insertions(+), 1 deletion(-)

diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/OwnershipCache.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/OwnershipCache.java
index 67e986b804c..7fd573187f3 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/OwnershipCache.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/OwnershipCache.java
@@ -22,6 +22,7 @@ import com.github.benmanes.caffeine.cache.AsyncCacheLoader;
 import com.github.benmanes.caffeine.cache.AsyncLoadingCache;
 import com.github.benmanes.caffeine.cache.Caffeine;
 import com.google.common.collect.Lists;
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.util.concurrent.MoreExecutors;
 import java.util.List;
 import java.util.Map;
@@ -213,7 +214,7 @@ public class OwnershipCache {
      *
      */
     public CompletableFuture<Void> removeOwnership(NamespaceBundle bundle) {
-        ResourceLock<NamespaceEphemeralData> lock = locallyAcquiredLocks.get(bundle);
+        ResourceLock<NamespaceEphemeralData> lock = locallyAcquiredLocks.remove(bundle);
         if (lock == null) {
             // We don't own the specified bundle anymore
             return CompletableFuture.completedFuture(null);
@@ -330,6 +331,16 @@ public class OwnershipCache {
         this.ownedBundlesCache.synchronous().invalidateAll();
     }
 
+    public void invalidateLocalOwnerCache(NamespaceBundle namespaceBundle) {
+        this.ownedBundlesCache.synchronous().invalidate(namespaceBundle);
+    }
+
+    @VisibleForTesting
+    public Map<NamespaceBundle, ResourceLock<NamespaceEphemeralData>> getLocallyAcquiredLocks() {
+        return locallyAcquiredLocks;
+    }
+
+
     public synchronized boolean refreshSelfOwnerInfo() {
         this.selfOwnerInfo = new NamespaceEphemeralData(pulsar.getBrokerServiceUrl(),
                 pulsar.getBrokerServiceUrlTls(), pulsar.getSafeWebServiceAddress(),
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/OwnershipCacheTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/OwnershipCacheTest.java
index 6168c61bb20..afe4d4c8b5b 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/OwnershipCacheTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/OwnershipCacheTest.java
@@ -367,6 +367,7 @@ public class OwnershipCacheTest {
         Awaitility.await().untilAsserted(() -> {
             assertTrue(cache.getOwnedBundles().isEmpty());
             assertFalse(store.exists(ServiceUnitUtils.path(bundle)).join());
+            assertNull(cache.getLocallyAcquiredLocks().get(bundle));
         });
     }