PIP-45: Allow to start bookie with Pulsar metadata store backend

bin/bookkeeper

@@ -194,6 +194,10 @@ OPTS="$OPTS -Dbk.log.level=error"
 OPTS="$OPTS -Dpulsar.log.dir=$BOOKIE_LOG_DIR"
 OPTS="$OPTS -Dpulsar.log.file=$BOOKIE_LOG_FILE"
 
+# Adding pulsar metadata as a recognized provider
+BK_METADATA_OPTIONS="-Dbookkeeper.metadata.bookie.drivers=org.apache.pulsar.metadata.bookkeeper.PulsarMetadataBookieDriver -Dbookkeeper.metadata.client.drivers=org.apache.pulsar.metadata.bookkeeper.PulsarMetadataClientDriver"
+OPTS="$OPTS $BK_METADATA_OPTIONS"
+
 #Change to BK_HOME to support relative paths
 cd "$BK_HOME"
 if [ $COMMAND == "bookie" ]; then

bin/pulsar

@@ -315,6 +315,10 @@ ZK_OPTS=" -Dzookeeper.4lw.commands.whitelist=* -Dzookeeper.snapshot.trust.empty=
 
 LOG4J2_SHUTDOWN_HOOK_DISABLED="-Dlog4j.shutdownHookEnabled=false"
 
+# Adding pulsar metadata as a recognized provider
+BK_METADATA_OPTIONS="-Dbookkeeper.metadata.bookie.drivers=org.apache.pulsar.metadata.bookkeeper.PulsarMetadataBookieDriver -Dbookkeeper.metadata.client.drivers=org.apache.pulsar.metadata.bookkeeper.PulsarMetadataClientDriver"
+OPTS="$OPTS $BK_METADATA_OPTIONS"
+
 #Change to PULSAR_HOME to support relative paths
 cd "$PULSAR_HOME"
 if [ $COMMAND == "broker" ]; then

conf/bookkeeper.conf

@@ -623,10 +623,43 @@ diskUsageThreshold=0.95
 # Default is 10000
 diskCheckInterval=10000
 
+
+############################################## Metadata Services ##############################################
+
+
+#############################################################################
+## Metadata Service settings
+#############################################################################
+
+# metadata service uri that bookkeeper is used for loading corresponding metadata driver and resolving its metadata service location
+# Examples: 
+#  - metadataServiceUri=metadata-store:zk:my-zk-1:2181
+#  - metadataServiceUri=metadata-store:etcd:http://my-etcd:2379
+metadataServiceUri=
+
+# @Deprecated - `ledgerManagerFactoryClass` is deprecated in favor of using `metadataServiceUri`
+# Ledger Manager Class
+# What kind of ledger manager is used to manage how ledgers are stored, managed
+# and garbage collected. Try to read 'BookKeeper Internals' for detail info.
+# ledgerManagerFactoryClass=org.apache.bookkeeper.meta.HierarchicalLedgerManagerFactory
+
+# @Deprecated - `ledgerManagerType` is deprecated in favor of using `ledgerManagerFactoryClass`.
+# ledgerManagerType=hierarchical
+
+# sometimes the bookkeeper server classes are shaded. the ledger manager factory classes might be relocated to be under other packages.
+# this would fail the clients using shaded factory classes since the factory classes are not matched. Users can enable this flag to
+# allow using shaded ledger manager class to connect to a bookkeeper cluster.
+# allowShadedLedgerManagerFactoryClass=false
+
+# the shaded ledger manager factory prefix. this is used when `allowShadedLedgerManagerFactoryClass` is set to true.
+# shadedLedgerManagerFactoryClassPrefix=dlshade.
+
+
 #############################################################################
 ## ZooKeeper parameters
 #############################################################################
 
+# @Deprecated - `zkLedgers` is deprecated in favor of using `metadataServiceUri`
 # A list of one of more servers on which Zookeeper is running.
 # The server list can be comma separated values, for example:
 # zkServers=zk1:2181,zk2:2181,zk3:2181

pulsar-broker/src/main/java/org/apache/pulsar/PulsarClusterMetadataSetup.java

@@ -48,6 +48,9 @@
 import org.apache.pulsar.metadata.api.MetadataStoreException;
 import org.apache.pulsar.metadata.api.MetadataStoreLifecycle;
 import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended;
+import org.apache.pulsar.metadata.bookkeeper.PulsarMetadataBookieDriver;
+import org.apache.pulsar.metadata.bookkeeper.PulsarMetadataClientDriver;
+import org.apache.pulsar.metadata.impl.ZKMetadataStore;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -79,9 +82,16 @@ private static class Arguments {
         private String clusterBrokerServiceUrlTls;
 
         @Parameter(names = { "-zk",
-                "--zookeeper" }, description = "Local ZooKeeper quorum connection string", required = true)
+                "--zookeeper" }, description = "Local ZooKeeper quorum connection string",
+                required = false,
+                hidden = true
+                )
         private String zookeeper;
 
+        @Parameter(names = { "-md",
+                "--metadata-store" }, description = "Metadata Store service url. eg: zk:my-zk:2181", required = false)
+        private String metadataStoreUrl;
+
         @Parameter(names = {
             "--zookeeper-session-timeout-ms"
         }, description = "Local zookeeper session timeout ms")
@@ -155,6 +165,9 @@ private static void initialDlogNamespaceMetadata(String configurationStore, Stri
     }
 
     public static void main(String[] args) throws Exception {
+        System.setProperty("bookkeeper.metadata.bookie.drivers", PulsarMetadataBookieDriver.class.getName());
+        System.setProperty("bookkeeper.metadata.client.drivers", PulsarMetadataClientDriver.class.getName());
+
         Arguments arguments = new Arguments();
         JCommander jcommander = new JCommander();
         try {
@@ -175,6 +188,12 @@ public static void main(String[] args) throws Exception {
             throw e;
         }
 
+        if (arguments.metadataStoreUrl == null && arguments.zookeeper == null) {
+            System.err.println("Metadata store address argument is required (--metadata-store)");
+            jcommander.usage();
+            System.exit(1);
+        }
+
         if (arguments.configurationStore == null && arguments.globalZookeeper == null) {
             System.err.println("Configuration store address argument is required (--configuration-store)");
             jcommander.usage();
@@ -192,45 +211,53 @@ public static void main(String[] args) throws Exception {
             arguments.configurationStore = arguments.globalZookeeper;
         }
 
+        if (arguments.metadataStoreUrl == null) {
+            arguments.metadataStoreUrl = arguments.zookeeper;
+        }
+
         if (arguments.numTransactionCoordinators <= 0) {
             System.err.println("Number of transaction coordinators must greater than 0");
             System.exit(1);
         }
 
-        log.info("Setting up cluster {} with zk={} configuration-store={}", arguments.cluster, arguments.zookeeper,
-                arguments.configurationStore);
+        log.info("Setting up cluster {} with metadata-store={} configuration-store={}", arguments.cluster,
+                arguments.metadataStoreUrl, arguments.configurationStore);
 
-        MetadataStoreExtended localStore = initMetadataStore(arguments.zookeeper, arguments.zkSessionTimeoutMillis);
+        MetadataStoreExtended localStore =
+                initMetadataStore(arguments.metadataStoreUrl, arguments.zkSessionTimeoutMillis);
         MetadataStoreExtended configStore = initMetadataStore(arguments.configurationStore,
                 arguments.zkSessionTimeoutMillis);
 
         // Format BookKeeper ledger storage metadata
         ServerConfiguration bkConf = new ServerConfiguration();
         if (arguments.existingBkMetadataServiceUri == null && arguments.bookieMetadataServiceUri == null) {
-            bkConf.setZkServers(arguments.zookeeper);
+            bkConf.setMetadataServiceUri("metadata-store:" + arguments.zookeeper);
             bkConf.setZkTimeout(arguments.zkSessionTimeoutMillis);
             if (!localStore.exists("/ledgers").get() // only format if /ledgers doesn't exist
                 && !BookKeeperAdmin.format(bkConf, false /* interactive */, false /* force */)) {
                 throw new IOException("Failed to initialize BookKeeper metadata");
             }
         }
 
+        if (localStore instanceof ZKMetadataStore && configStore instanceof ZKMetadataStore) {
+            String uriStr;
+            if (arguments.existingBkMetadataServiceUri != null) {
+                uriStr = arguments.existingBkMetadataServiceUri;
+            } else if (arguments.bookieMetadataServiceUri != null) {
+                uriStr = arguments.bookieMetadataServiceUri;
+            } else {
+                 uriStr = bkConf.getMetadataServiceUri().replace("metadata-store:", "zk://");
+            }
+            ServiceURI bkMetadataServiceUri = ServiceURI.create(uriStr);
 
-        String uriStr = bkConf.getMetadataServiceUri();
-        if (arguments.existingBkMetadataServiceUri != null) {
-            uriStr = arguments.existingBkMetadataServiceUri;
-        } else if (arguments.bookieMetadataServiceUri != null) {
-            uriStr = arguments.bookieMetadataServiceUri;
-        }
-        ServiceURI bkMetadataServiceUri = ServiceURI.create(uriStr);
-
-        // initial distributed log metadata
-        initialDlogNamespaceMetadata(arguments.configurationStore, uriStr);
+            // initial distributed log metadata
+            initialDlogNamespaceMetadata(arguments.configurationStore, uriStr);
 
-        // Format BookKeeper stream storage metadata
-        if (arguments.numStreamStorageContainers > 0) {
-            ClusterInitializer initializer = new ZkClusterInitializer(arguments.zookeeper);
-            initializer.initializeCluster(bkMetadataServiceUri.getUri(), arguments.numStreamStorageContainers);
+            // Format BookKeeper stream storage metadata
+            if (arguments.numStreamStorageContainers > 0) {
+                ClusterInitializer initializer = new ZkClusterInitializer(arguments.metadataStoreUrl);
+                initializer.initializeCluster(bkMetadataServiceUri.getUri(), arguments.numStreamStorageContainers);
+            }
         }
 
         if (!localStore.exists(BookieRackAffinityMapping.BOOKIE_INFO_ROOT_PATH).get()) {