Fixed counter base path getting deleted (apache#12193)

bobosmile1170556238 · Sep 26, 2021 · 55e304b · 55e304b
1 parent a69611c
commit 55e304b
Show file tree

Hide file tree

Showing 2 changed files with 49 additions and 0 deletions.
diff --git a/...a/src/main/java/org/apache/pulsar/metadata/coordination/impl/CoordinationServiceImpl.java b/...a/src/main/java/org/apache/pulsar/metadata/coordination/impl/CoordinationServiceImpl.java
@@ -82,6 +82,19 @@ public <T> LockManager<T> getLockManager(Class<T> clazz) {
 
     @Override
     public CompletableFuture<Long> getNextCounterValue(String path) {
+        return store.exists(path)
+                .thenCompose(exists -> {
+                    if (exists) {
+                        // The base path already exists
+                        return incrementCounter(path);
+                    } else {
+                        return store.put(path, new byte[0], Optional.empty())
+                                .thenCompose(__ -> incrementCounter(path));
+                    }
+                });
+    }
+
+    private CompletableFuture<Long> incrementCounter(String path) {
         String counterBasePath = path + "/-";
         return store
                 .put(counterBasePath, new byte[0], Optional.of(-1L),

diff --git a/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/CounterTest.java b/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/CounterTest.java
@@ -51,4 +51,40 @@ public void basicTest(String provider, Supplier<String> urlSupplier) throws Exce
         long l4 = cs1.getNextCounterValue("/my/path").join();
         assertNotEquals(l3, l4);
     }
+
+    @Test(dataProvider = "impl")
+    public void testCounterDoesNotAutoReset(String provider, Supplier<String> urlSupplier) throws Exception {
+        if (provider.equals("Memory")) {
+            // Test doesn't make sense for local memory since we're testing across different instances
+            return;
+        }
+
+        MetadataStoreExtended store1 = MetadataStoreExtended.create(urlSupplier.get(),
+                MetadataStoreConfig.builder().build());
+
+        CoordinationService cs1 = new CoordinationServiceImpl(store1);
+
+        long l1 = cs1.getNextCounterValue("/my/path").join();
+        long l2 = cs1.getNextCounterValue("/my/path").join();
+        long l3 = cs1.getNextCounterValue("/my/path").join();
+
+        assertNotEquals(l1, l2);
+        assertNotEquals(l2, l3);
+
+        cs1.close();
+        store1.close();;
+
+        // Delete all the empty container nodes
+        zks.checkContainers();
+
+        MetadataStoreExtended store2 = MetadataStoreExtended.create(urlSupplier.get(),
+                MetadataStoreConfig.builder().build());
+        @Cleanup
+        CoordinationService cs2 = new CoordinationServiceImpl(store2);
+
+        long l4 = cs2.getNextCounterValue("/my/path").join();
+        assertNotEquals(l1, l4);
+        assertNotEquals(l2, l4);
+        assertNotEquals(l3, l4);
+    }
 }