Change callbacks in CleanupLedgerManager as a Set (#4123)

apache · Jan 11, 2024 · 113d40a · 113d40a
1 parent 54168b5
commit 113d40a
Showing 1 changed file with 8 additions and 11 deletions.
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/CleanupLedgerManager.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/CleanupLedgerManager.java
@@ -23,7 +23,6 @@
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.bookkeeper.client.BKException;
@@ -55,7 +54,7 @@ private class CleanupGenericCallback<T> implements GenericCallback<T> {
         public void operationComplete(int rc, T result) {
             closeLock.readLock().lock();
             try {
-                if (!closed && null != removeCallback(cb)) {
+                if (!closed && removeCallback(cb)) {
                     cb.operationComplete(rc, result);
                 }
             } finally {
@@ -78,8 +77,7 @@ public LedgerRange next() throws IOException {
     }
 
     private final LedgerManager underlying;
-    private final ConcurrentMap<GenericCallback, GenericCallback> callbacks =
-        new ConcurrentHashMap<GenericCallback, GenericCallback>();
+    private final Set<GenericCallback> callbacks = ConcurrentHashMap.newKeySet();
     private boolean closed = false;
     private final ReentrantReadWriteLock closeLock = new ReentrantReadWriteLock();
     private final Set<CompletableFuture<?>> futures = ConcurrentHashMap.newKeySet();
@@ -94,7 +92,7 @@ public LedgerManager getUnderlying() {
     }
 
     private void addCallback(GenericCallback callback) {
-        callbacks.put(callback, callback);
+        callbacks.add(callback);
     }
 
     @Override
@@ -107,7 +105,7 @@ public void unregisterLedgerMetadataListener(long ledgerId, LedgerMetadataListen
         underlying.unregisterLedgerMetadataListener(ledgerId, listener);
     }
 
-    private GenericCallback removeCallback(GenericCallback callback) {
+    private boolean removeCallback(GenericCallback callback) {
         return callbacks.remove(callback);
     }
 
@@ -206,7 +204,7 @@ public void operationComplete(int rc, Void result) {
             underlying.asyncProcessLedgers(processor, new AsyncCallback.VoidCallback() {
                 @Override
                 public void processResult(int rc, String path, Object ctx) {
-                    if (null != removeCallback(stub)) {
+                    if (removeCallback(stub)) {
                         finalCb.processResult(rc, path, ctx);
                     }
                 }
@@ -239,14 +237,13 @@ public void close() throws IOException {
                 return;
             }
             closed = true;
-            keys = new HashSet<GenericCallback>(callbacks.keySet());
+            keys = new HashSet<>(callbacks);
         } finally {
             closeLock.writeLock().unlock();
         }
         for (GenericCallback key : keys) {
-            GenericCallback callback = callbacks.remove(key);
-            if (null != callback) {
-                callback.operationComplete(BKException.Code.ClientClosedException, null);
+            if (callbacks.remove(key)) {
+                key.operationComplete(BKException.Code.ClientClosedException, null);
             }
         }
         BKException exception = new BKException.BKClientClosedException();