[fix][meta] Check if metadata store is closed in RocksdbMetadataStore (…

…apache#22852) (cherry picked from commit 7419287)
datastax · Jun 6, 2024 · 54ad6cd · 54ad6cd
1 parent bbc4b4c
commit 54ad6cd
Show file tree

Hide file tree

Showing 2 changed files with 27 additions and 13 deletions.
diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/AbstractMetadataStore.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/AbstractMetadataStore.java
@@ -247,8 +247,7 @@ public <T> MetadataCache<T> getMetadataCache(MetadataSerde<T> serde, MetadataCac
     @Override
     public CompletableFuture<Optional<GetResult>> get(String path) {
         if (isClosed()) {
-            return FutureUtil.failedFuture(
-                    new MetadataStoreException.AlreadyClosedException());
+            return alreadyClosedFailedFuture();
         }
         long start = System.currentTimeMillis();
         if (!isValidPath(path)) {
@@ -276,8 +275,7 @@ public CompletableFuture<Stat> put(String path, byte[] value, Optional<Long> exp
     @Override
     public final CompletableFuture<List<String>> getChildren(String path) {
         if (isClosed()) {
-            return FutureUtil.failedFuture(
-                    new MetadataStoreException.AlreadyClosedException());
+            return alreadyClosedFailedFuture();
         }
         if (!isValidPath(path)) {
             return FutureUtil.failedFuture(new MetadataStoreException.InvalidPathException(path));
@@ -288,8 +286,7 @@ public final CompletableFuture<List<String>> getChildren(String path) {
     @Override
     public final CompletableFuture<Boolean> exists(String path) {
         if (isClosed()) {
-            return FutureUtil.failedFuture(
-                    new MetadataStoreException.AlreadyClosedException());
+            return alreadyClosedFailedFuture();
         }
         if (!isValidPath(path)) {
             return FutureUtil.failedFuture(new MetadataStoreException.InvalidPathException(path));
@@ -351,8 +348,7 @@ public void accept(Notification n) {
     @Override
     public final CompletableFuture<Void> delete(String path, Optional<Long> expectedVersion) {
         if (isClosed()) {
-            return FutureUtil.failedFuture(
-                    new MetadataStoreException.AlreadyClosedException());
+            return alreadyClosedFailedFuture();
         }
         long start = System.currentTimeMillis();
         if (!isValidPath(path)) {
@@ -401,8 +397,7 @@ private CompletableFuture<Void> deleteInternal(String path, Optional<Long> expec
     @Override
     public CompletableFuture<Void> deleteRecursive(String path) {
         if (isClosed()) {
-            return FutureUtil.failedFuture(
-                    new MetadataStoreException.AlreadyClosedException());
+            return alreadyClosedFailedFuture();
         }
         return getChildren(path)
                 .thenCompose(children -> FutureUtil.waitForAll(
@@ -426,8 +421,7 @@ protected abstract CompletableFuture<Stat> storePut(String path, byte[] data, Op
     public final CompletableFuture<Stat> put(String path, byte[] data, Optional<Long> optExpectedVersion,
             EnumSet<CreateOption> options) {
         if (isClosed()) {
-            return FutureUtil.failedFuture(
-                    new MetadataStoreException.AlreadyClosedException());
+            return alreadyClosedFailedFuture();
         }
         long start = System.currentTimeMillis();
         if (!isValidPath(path)) {
@@ -507,10 +501,15 @@ protected void receivedSessionEvent(SessionEvent event) {
         }
     }
 
-    private boolean isClosed() {
+    protected boolean isClosed() {
         return isClosed.get();
     }
 
+    protected static <T> CompletableFuture<T> alreadyClosedFailedFuture() {
+        return FutureUtil.failedFuture(
+                new MetadataStoreException.AlreadyClosedException());
+    }
+
     @Override
     public void close() throws Exception {
         executor.shutdownNow();

diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/RocksdbMetadataStore.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/RocksdbMetadataStore.java
@@ -375,6 +375,9 @@ public CompletableFuture<Optional<GetResult>> storeGet(String path) {
         }
         try {
             dbStateLock.readLock().lock();
+            if (isClosed()) {
+                return alreadyClosedFailedFuture();
+            }
             byte[] value = db.get(optionCache, toBytes(path));
             if (value == null) {
                 return CompletableFuture.completedFuture(Optional.empty());
@@ -407,6 +410,9 @@ protected CompletableFuture<List<String>> getChildrenFromStore(String path) {
         }
         try {
             dbStateLock.readLock().lock();
+            if (isClosed()) {
+                return alreadyClosedFailedFuture();
+            }
             try (RocksIterator iterator = db.newIterator(optionDontCache)) {
                 Set<String> result = new HashSet<>();
                 String firstKey = path.equals("/") ? path : path + "/";
@@ -449,6 +455,9 @@ protected CompletableFuture<Boolean> existsFromStore(String path) {
         }
         try {
             dbStateLock.readLock().lock();
+            if (isClosed()) {
+                return alreadyClosedFailedFuture();
+            }
             byte[] value = db.get(optionDontCache, toBytes(path));
             if (log.isDebugEnabled()) {
                 if (value != null) {
@@ -471,6 +480,9 @@ protected CompletableFuture<Void> storeDelete(String path, Optional<Long> expect
         }
         try {
             dbStateLock.readLock().lock();
+            if (isClosed()) {
+                return alreadyClosedFailedFuture();
+            }
             try (Transaction transaction = db.beginTransaction(writeOptions)) {
                 byte[] pathBytes = toBytes(path);
                 byte[] oldValueData = transaction.getForUpdate(optionDontCache, pathBytes, true);
@@ -507,6 +519,9 @@ protected CompletableFuture<Stat> storePut(String path, byte[] data, Optional<Lo
         }
         try {
             dbStateLock.readLock().lock();
+            if (isClosed()) {
+                return alreadyClosedFailedFuture();
+            }
             try (Transaction transaction = db.beginTransaction(writeOptions)) {
                 byte[] pathBytes = toBytes(path);
                 byte[] oldValueData = transaction.getForUpdate(optionDontCache, pathBytes, true);