[pulsar] branch master updated: [metadata] fix MetadataStore#put may get unexcepted exception (#14903)

[mmerli]

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

mmerli pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git


The following commit(s) were added to refs/heads/master by this push:
     new 3b852bcb547 [metadata] fix MetadataStore#put may get unexcepted 
exception (#14903)
3b852bcb547 is described below

commit 3b852bcb54780ae26cd339e5e30bbe83de3a30cd
Author: WangJialing <65590138+wangjialing...@users.noreply.github.com>
AuthorDate: Wed Jul 13 08:02:46 2022 +0800

    [metadata] fix MetadataStore#put may get unexcepted exception (#14903)
    
    * fix MetadataStore#put may get unexcepted exception
    
    * fix checksytle
    
    * add test
    
    * update test code
---
 .../pulsar/metadata/impl/ZKMetadataStore.java       |  8 +++++++-
 .../apache/pulsar/metadata/MetadataStoreTest.java   | 21 +++++++++++++++++++++
 2 files changed, 28 insertions(+), 1 deletion(-)

diff --git 
a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/ZKMetadataStore.java
 
b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/ZKMetadataStore.java
index 6697934d56b..ad723f28f89 100644
--- 
a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/ZKMetadataStore.java
+++ 
b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/ZKMetadataStore.java
@@ -386,7 +386,13 @@ public class ZKMetadataStore extends 
AbstractBatchedMetadataStore
                                 put(opPut.getPath(), opPut.getData(), 
Optional.of(-1L)).thenAccept(
                                                 s -> future.complete(s))
                                         .exceptionally(ex -> {
-                                            
future.completeExceptionally(MetadataStoreException.wrap(ex.getCause()));
+                                            if (ex.getCause() instanceof 
BadVersionException) {
+                                                // The z-node exist now, let's 
overwrite it
+                                                internalStorePut(opPut);
+                                            } else {
+                                                future.completeExceptionally(
+                                                        
MetadataStoreException.wrap(ex.getCause()));
+                                            }
                                             return null;
                                         });
                             }
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
index f4e5bf779ba..b8a091fef25 100644
--- 
a/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/MetadataStoreTest.java
+++ 
b/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/MetadataStoreTest.java
@@ -38,6 +38,7 @@ import java.util.function.Supplier;
 import lombok.Cleanup;
 import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.common.util.FutureUtil;
 import org.apache.pulsar.metadata.api.GetResult;
 import org.apache.pulsar.metadata.api.MetadataStore;
 import org.apache.pulsar.metadata.api.MetadataStoreConfig;
@@ -84,6 +85,26 @@ public class MetadataStoreTest extends BaseMetadataStoreTest 
{
         }
     }
 
+    @Test(dataProvider = "impl")
+    public void concurrentPutTest(String provider, Supplier<String> 
urlSupplier) throws Exception {
+        @Cleanup
+        MetadataStore store = MetadataStoreFactory.create(urlSupplier.get(), 
MetadataStoreConfig.builder().build());
+
+        String data = "data";
+        String path = "/non-existing-key";
+        int concurrent = 50;
+        List<CompletableFuture<Stat>> futureList = new ArrayList<>();
+        for (int i = 0; i < concurrent; i++) {
+            futureList.add(store.put(path, data.getBytes(), 
Optional.empty()).exceptionally(ex -> {
+                fail("fail to execute concurrent put", ex);
+                return null;
+            }));
+        }
+        FutureUtil.waitForAll(futureList).join();
+
+        assertEquals(store.get(path).join().get().getValue(), data.getBytes());
+    }
+
     @Test(dataProvider = "impl")
     public void insertionTestWithExpectedVersion(String provider, 
Supplier<String> urlSupplier) throws Exception {
         @Cleanup