[Issue 15896] Fix LockTimeout when storePut on the same key concurren…

…tly in RocksdbMetadataStore (#16005) * Fix LockTimeout when storePut or storeDelete on the same key concurrently * Add cleanup
apache · Jun 15, 2022 · 8a14fe0 · 8a14fe0
1 parent 0d4c0b1
commit 8a14fe0
Show file tree

Hide file tree

Showing 2 changed files with 40 additions and 2 deletions.
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
@@ -487,7 +487,7 @@ protected CompletableFuture<Void> storeDelete(String path, Optional<Long> expect
             }
             try (Transaction transaction = db.beginTransaction(optionSync)) {
                 byte[] pathBytes = toBytes(path);
-                byte[] oldValueData = transaction.getForUpdate(optionDontCache, pathBytes, false);
+                byte[] oldValueData = transaction.getForUpdate(optionDontCache, pathBytes, true);
                 MetaValue metaValue = MetaValue.parse(oldValueData);
                 if (metaValue == null) {
                     throw new MetadataStoreException.NotFoundException(String.format("path %s not found.", path));
@@ -504,6 +504,9 @@ protected CompletableFuture<Void> storeDelete(String path, Optional<Long> expect
                 return CompletableFuture.completedFuture(null);
             }
         } catch (Throwable e) {
+            if (log.isDebugEnabled()) {
+                log.debug("error in storeDelete,path={}", path, e);
+            }
             return FutureUtil.failedFuture(MetadataStoreException.wrap(e));
         } finally {
             dbStateLock.readLock().unlock();
@@ -523,7 +526,7 @@ protected CompletableFuture<Stat> storePut(String path, byte[] data, Optional<Lo
             }
             try (Transaction transaction = db.beginTransaction(optionSync)) {
                 byte[] pathBytes = toBytes(path);
-                byte[] oldValueData = transaction.getForUpdate(optionDontCache, pathBytes, false);
+                byte[] oldValueData = transaction.getForUpdate(optionDontCache, pathBytes, true);
                 MetaValue metaValue = MetaValue.parse(oldValueData);
                 if (expectedVersion.isPresent()) {
                     if (metaValue == null && expectedVersion.get() != -1
@@ -572,6 +575,9 @@ protected CompletableFuture<Stat> storePut(String path, byte[] data, Optional<Lo
                                 metaValue.ephemeral, true));
             }
         } catch (Throwable e) {
+            if (log.isDebugEnabled()) {
+                log.debug("error in storePut,path={}", path, e);
+            }
             return FutureUtil.failedFuture(MetadataStoreException.wrap(e));
         } finally {
             dbStateLock.readLock().unlock();

diff --git a/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/MetadataStoreTest.java b/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/MetadataStoreTest.java
@@ -49,6 +49,7 @@
 import org.apache.pulsar.metadata.api.NotificationType;
 import org.apache.pulsar.metadata.api.Stat;
 import org.assertj.core.util.Lists;
+import org.awaitility.Awaitility;
 import org.testng.annotations.Test;
 
 @Slf4j
@@ -447,4 +448,35 @@ public void run() {
         assertEquals(successWrites.get(), maxValue);
         assertEquals(store.get(path).get().get().getValue()[0], maxValue);
     }
+
+    @Test(dataProvider = "impl")
+    public void testConcurrentPut(String provider, Supplier<String> urlSupplier) throws Exception {
+        @Cleanup
+        MetadataStore store = MetadataStoreFactory.create(urlSupplier.get(), MetadataStoreConfig.builder().build());
+
+        String k = newKey();
+        CompletableFuture<Void> f1 =
+                CompletableFuture.runAsync(() -> store.put(k, new byte[0], Optional.of(-1L)).join());
+        CompletableFuture<Void> f2 =
+                CompletableFuture.runAsync(() -> store.put(k, new byte[0], Optional.of(-1L)).join());
+        Awaitility.await().until(() -> f1.isDone() && f2.isDone());
+        assertTrue(f1.isCompletedExceptionally() && !f2.isCompletedExceptionally() ||
+                ! f1.isCompletedExceptionally() && f2.isCompletedExceptionally());
+    }
+
+    @Test(dataProvider = "impl")
+    public void testConcurrentDelete(String provider, Supplier<String> urlSupplier) throws Exception {
+        @Cleanup
+        MetadataStore store = MetadataStoreFactory.create(urlSupplier.get(), MetadataStoreConfig.builder().build());
+
+        String k = newKey();
+        store.put(k, new byte[0], Optional.of(-1L)).join();
+        CompletableFuture<Void> f1 =
+                CompletableFuture.runAsync(() -> store.delete(k, Optional.empty()).join());
+        CompletableFuture<Void> f2 =
+                CompletableFuture.runAsync(() -> store.delete(k, Optional.empty()).join());
+        Awaitility.await().until(() -> f1.isDone() && f2.isDone());
+        assertTrue(f1.isCompletedExceptionally() && !f2.isCompletedExceptionally() ||
+                ! f1.isCompletedExceptionally() && f2.isCompletedExceptionally());
+    }
 }