Remove redundant ledger dir creation when necessary (apache#7196)

### Motivation Fixes apache#7033 ### Modifications Add the logic to handle the creation of ledger dir when using an existing bookkeeper cluster and a test for this.
yew1eb · Jun 10, 2020 · 5421a4c · 5421a4c
1 parent 3da6452
commit 5421a4c
Show file tree

Hide file tree

Showing 2 changed files with 55 additions and 8 deletions.
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/PulsarClusterMetadataSetup.java b/pulsar-broker/src/main/java/org/apache/pulsar/PulsarClusterMetadataSetup.java
@@ -110,6 +110,10 @@ private static class Arguments {
         }, description = "Num transaction coordinators will assigned in cluster")
         private int numTransactionCoordinators = 16;
 
+        @Parameter(names = {
+            "--bookkeeper-metadata-service-uri" }, description = "Metadata service uri of BookKeeper")
+        private String bkMetadataServiceUri;
+
         @Parameter(names = { "-h", "--help" }, description = "Show this help message")
         private boolean help = false;
     }
@@ -150,7 +154,7 @@ public static void main(String[] args) throws Exception {
         }
 
         if (arguments.configurationStore != null && arguments.globalZookeeper != null) {
-            System.err.println("Configuration store argument (--configuration-store) supercedes the deprecated (--global-zookeeper) argument");
+            System.err.println("Configuration store argument (--configuration-store) supersedes the deprecated (--global-zookeeper) argument");
             jcommander.usage();
             System.exit(1);
         }
@@ -172,16 +176,19 @@ public static void main(String[] args) throws Exception {
 
         // Format BookKeeper ledger storage metadata
         ServerConfiguration bkConf = new ServerConfiguration();
-        bkConf.setZkServers(arguments.zookeeper);
-        bkConf.setZkTimeout(arguments.zkSessionTimeoutMillis);
-        if (localZk.exists("/ledgers", false) == null // only format if /ledgers doesn't exist
-                && !BookKeeperAdmin.format(bkConf, false /* interactive */, false /* force */)) {
-            throw new IOException("Failed to initialize BookKeeper metadata");
+        if (arguments.bkMetadataServiceUri == null) {
+            bkConf.setZkServers(arguments.zookeeper);
+            bkConf.setZkTimeout(arguments.zkSessionTimeoutMillis);
+            if (localZk.exists("/ledgers", false) == null // only format if /ledgers doesn't exist
+                    && !BookKeeperAdmin.format(bkConf, false /* interactive */, false /* force */)) {
+                throw new IOException("Failed to initialize BookKeeper metadata");
+            }
         }
 
         // Format BookKeeper stream storage metadata
         if (arguments.numStreamStorageContainers > 0) {
-            ServiceURI bkMetadataServiceUri = ServiceURI.create(bkConf.getMetadataServiceUri());
+            String uriStr = arguments.bkMetadataServiceUri == null ? bkConf.getMetadataServiceUri() : arguments.bkMetadataServiceUri;
+            ServiceURI bkMetadataServiceUri = ServiceURI.create(uriStr);
             ClusterInitializer initializer = new ZkClusterInitializer(arguments.zookeeper);
             initializer.initializeCluster(bkMetadataServiceUri.getUri(), arguments.numStreamStorageContainers);
         }

diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/zookeeper/ClusterMetadataSetupTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/zookeeper/ClusterMetadataSetupTest.java
@@ -20,14 +20,18 @@
 
 import org.apache.pulsar.PulsarClusterMetadataSetup;
 import org.apache.pulsar.broker.zookeeper.ZooKeeperClientAspectJTest.ZookeeperServerTest;
+import org.apache.zookeeper.ZooKeeper;
 import org.testng.annotations.AfterMethod;
 import org.testng.annotations.BeforeMethod;
 import org.testng.annotations.Test;
 
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertNull;
+
 public class ClusterMetadataSetupTest {
     private ZookeeperServerTest localZkS;
 
-    // test SetupClusterMetadata several times, all should be suc
+    // test SetupClusterMetadata several times, all should be successful
     @Test
     public void testReSetupClusterMetadata() throws Exception {
         String[] args = {
@@ -44,6 +48,42 @@ public void testReSetupClusterMetadata() throws Exception {
         PulsarClusterMetadataSetup.main(args);
     }
 
+    @Test
+    public void testSetupWithBkMetadataServiceUri() throws Exception {
+        String zkConnection = "127.0.0.1:" + localZkS.getZookeeperPort();
+        String[] args = {
+                "--cluster", "testReSetupClusterMetadata-cluster",
+                "--zookeeper", zkConnection,
+                "--configuration-store", zkConnection,
+                "--bookkeeper-metadata-service-uri", "zk+null://" + zkConnection + "/chroot/ledgers",
+                "--web-service-url", "http://127.0.0.1:8080",
+                "--web-service-url-tls", "https://127.0.0.1:8443",
+                "--broker-service-url", "pulsar://127.0.0.1:6650",
+                "--broker-service-url-tls","pulsar+ssl://127.0.0.1:6651"
+        };
+
+        PulsarClusterMetadataSetup.main(args);
+
+        ZooKeeper localZk = PulsarClusterMetadataSetup.initZk(zkConnection, 30000);
+        // expected not exist
+        assertNull(localZk.exists("/ledgers", false));
+
+        String[] args1 = {
+                "--cluster", "testReSetupClusterMetadata-cluster",
+                "--zookeeper", zkConnection,
+                "--configuration-store", zkConnection,
+                "--web-service-url", "http://127.0.0.1:8080",
+                "--web-service-url-tls", "https://127.0.0.1:8443",
+                "--broker-service-url", "pulsar://127.0.0.1:6650",
+                "--broker-service-url-tls","pulsar+ssl://127.0.0.1:6651"
+        };
+
+        PulsarClusterMetadataSetup.main(args1);
+
+        // expected exist
+        assertNotNull(localZk.exists("/ledgers", false));
+    }
+
     @BeforeMethod
     void setup() throws Exception {
         localZkS = new ZookeeperServerTest(0);