[bookkeeper] branch master updated: Create ledgersRootPath recursively for 'bin/bookkeeper shell metaformat'

[si...@apache.org]

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

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


The following commit(s) were added to refs/heads/master by this push:
     new 004cfe3  Create ledgersRootPath recursively for 'bin/bookkeeper shell metaformat'
004cfe3 is described below

commit 004cfe36af8365c6a9bb198ccf74eab311f0249d
Author: Dapeng <su...@cmss.chinamobile.com>
AuthorDate: Wed Jan 22 18:09:41 2020 +0800

    Create ledgersRootPath recursively for 'bin/bookkeeper shell metaformat'
    
    Create ledgersRootPath recursively for `bin/bookkeeper shell metaformat`. The existence of any parent znodes is not an error condition
    
    
    Reviewers: Enrico Olivelli <eo...@gmail.com>, Sijie Guo <None>
    
    This closes #2237 from SunDapeng1/branch-2236
---
 .../bookkeeper/discover/ZKRegistrationManager.java |  3 +-
 .../discover/TestZkRegistrationManager.java        | 43 ++++++++++++++++++++++
 2 files changed, 45 insertions(+), 1 deletion(-)

diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/discover/ZKRegistrationManager.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/discover/ZKRegistrationManager.java
index a2a3e7c..e9bd784 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/discover/ZKRegistrationManager.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/discover/ZKRegistrationManager.java
@@ -400,7 +400,8 @@ public class ZKRegistrationManager implements RegistrationManager {
         boolean availableNodeExists = null != zk.exists(bookieRegistrationPath, false);
         // Create ledgers root node if not exists
         if (!ledgerRootExists) {
-            zk.create(ledgersRootPath, "".getBytes(Charsets.UTF_8), zkAcls, CreateMode.PERSISTENT);
+            ZkUtils.createFullPathOptimistic(zk, ledgersRootPath, "".getBytes(Charsets.UTF_8), zkAcls,
+                    CreateMode.PERSISTENT);
         }
         // create available bookies node if not exists
         if (!availableNodeExists) {
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/discover/TestZkRegistrationManager.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/discover/TestZkRegistrationManager.java
index ff1b7ea..8949e12 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/discover/TestZkRegistrationManager.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/discover/TestZkRegistrationManager.java
@@ -18,8 +18,51 @@
  */
 package org.apache.bookkeeper.discover;
 
+import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.bookkeeper.conf.TestBKConfiguration;
+import org.apache.bookkeeper.test.ZooKeeperCluster;
+import org.apache.bookkeeper.test.ZooKeeperUtil;
+import org.apache.zookeeper.ZooKeeper;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import static org.junit.Assert.assertTrue;
+
+
 /**
  * Unit test of {@link RegistrationManager}.
  */
 public class TestZkRegistrationManager {
+
+    private ZooKeeperCluster localZkServer;
+    private ZooKeeper zkc;
+
+    @Before
+    public void setup() throws Exception {
+        localZkServer = new ZooKeeperUtil();
+        localZkServer.startCluster();
+    }
+
+    @After
+    public void teardown() throws Exception {
+        localZkServer.stopCluster();
+    }
+
+    @Test
+    public void testPrepareFormat () throws Exception{
+        try {
+            ServerConfiguration conf = TestBKConfiguration.newServerConfiguration();
+            conf.setMetadataServiceUri("zk+hierarchical://localhost:2181/test/ledgers");
+            zkc = localZkServer.getZooKeeperClient();
+            ZKRegistrationManager zkRegistrationManager = new ZKRegistrationManager(conf, zkc,() -> {} );
+            zkRegistrationManager.prepareFormat();
+            assertTrue(zkc.exists("/test/ledgers",false) != null);
+        } finally {
+            if (zkc != null) {
+                zkc.close();
+            }
+        }
+    }
+
 }