From 9458e09343f1b13d93b992d346fb64e0a28867db Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Mon, 13 Dec 2021 15:23:52 -0800 Subject: [PATCH 1/5] PIP-45: Allow to start bookie with Pulsar metadata store backend --- bin/bookkeeper | 4 ++ bin/pulsar | 4 ++ conf/bookkeeper.conf | 33 +++++++++++ .../pulsar/PulsarClusterMetadataSetup.java | 56 +++++++++++++------ 4 files changed, 79 insertions(+), 18 deletions(-) diff --git a/bin/bookkeeper b/bin/bookkeeper index 0d01537586357..428e975959105 100755 --- a/bin/bookkeeper +++ b/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 diff --git a/bin/pulsar b/bin/pulsar index d28fbe88b40f7..43e137af5317a 100755 --- a/bin/pulsar +++ b/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 diff --git a/conf/bookkeeper.conf b/conf/bookkeeper.conf index f26abe086e4f0..01adae7b2ca38 100644 --- a/conf/bookkeeper.conf +++ b/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 diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/PulsarClusterMetadataSetup.java b/pulsar-broker/src/main/java/org/apache/pulsar/PulsarClusterMetadataSetup.java index 7829600ef129b..8e0da91dd816d 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/PulsarClusterMetadataSetup.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/PulsarClusterMetadataSetup.java @@ -48,6 +48,7 @@ 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.impl.ZKMetadataStore; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -79,9 +80,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") @@ -175,6 +183,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,22 +206,26 @@ 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 */)) { @@ -216,21 +234,23 @@ public static void main(String[] args) throws Exception { } - String uriStr = bkConf.getMetadataServiceUri(); - if (arguments.existingBkMetadataServiceUri != null) { - uriStr = arguments.existingBkMetadataServiceUri; - } else if (arguments.bookieMetadataServiceUri != null) { - uriStr = arguments.bookieMetadataServiceUri; - } - ServiceURI bkMetadataServiceUri = ServiceURI.create(uriStr); + if (localStore instanceof ZKMetadataStore && configStore instanceof ZKMetadataStore) { + 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()) { From 5c71f6de4bed76e3d5732f3dc9c347d992ae4ae8 Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Tue, 14 Dec 2021 08:29:38 -0800 Subject: [PATCH 2/5] Fixed checkstyle --- .../java/org/apache/pulsar/PulsarClusterMetadataSetup.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/PulsarClusterMetadataSetup.java b/pulsar-broker/src/main/java/org/apache/pulsar/PulsarClusterMetadataSetup.java index 8e0da91dd816d..52ab2a4a64724 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/PulsarClusterMetadataSetup.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/PulsarClusterMetadataSetup.java @@ -218,7 +218,8 @@ public static void main(String[] args) throws Exception { log.info("Setting up cluster {} with metadata-store={} configuration-store={}", arguments.cluster, arguments.metadataStoreUrl, arguments.configurationStore); - MetadataStoreExtended localStore = initMetadataStore(arguments.metadataStoreUrl, arguments.zkSessionTimeoutMillis); + MetadataStoreExtended localStore = + initMetadataStore(arguments.metadataStoreUrl, arguments.zkSessionTimeoutMillis); MetadataStoreExtended configStore = initMetadataStore(arguments.configurationStore, arguments.zkSessionTimeoutMillis); From 9d57be13a26dc11ee3ae8bcc1e720cf878c3c0d8 Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Tue, 14 Dec 2021 11:04:09 -0800 Subject: [PATCH 3/5] Fixed DLog initialization when using zk --- .../main/java/org/apache/pulsar/PulsarClusterMetadataSetup.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/PulsarClusterMetadataSetup.java b/pulsar-broker/src/main/java/org/apache/pulsar/PulsarClusterMetadataSetup.java index 52ab2a4a64724..af049105d01e6 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/PulsarClusterMetadataSetup.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/PulsarClusterMetadataSetup.java @@ -236,7 +236,7 @@ public static void main(String[] args) throws Exception { if (localStore instanceof ZKMetadataStore && configStore instanceof ZKMetadataStore) { - String uriStr = bkConf.getMetadataServiceUri(); + String uriStr = bkConf.getMetadataServiceUri().replace("metadata-store:", "zk://"); if (arguments.existingBkMetadataServiceUri != null) { uriStr = arguments.existingBkMetadataServiceUri; } else if (arguments.bookieMetadataServiceUri != null) { From 07065c1471370cd318bb8e57f1d4f3c9a9cf0912 Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Tue, 11 Jan 2022 21:13:48 -0800 Subject: [PATCH 4/5] Fixed setup of PulsarClusterMetadataSetup --- .../java/org/apache/pulsar/PulsarClusterMetadataSetup.java | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/PulsarClusterMetadataSetup.java b/pulsar-broker/src/main/java/org/apache/pulsar/PulsarClusterMetadataSetup.java index af049105d01e6..efddc3c61e7a3 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/PulsarClusterMetadataSetup.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/PulsarClusterMetadataSetup.java @@ -48,6 +48,8 @@ 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; @@ -163,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 { From b15f3308aaf279aa13742cfad933847b0a57b6b1 Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Sat, 15 Jan 2022 10:00:10 -0800 Subject: [PATCH 5/5] Fixed order to avoid NPE --- .../java/org/apache/pulsar/PulsarClusterMetadataSetup.java | 5 +++-- 1 file changed, 3 insertions(+), 2 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 index efddc3c61e7a3..52d7e0d8e3bb5 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/PulsarClusterMetadataSetup.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/PulsarClusterMetadataSetup.java @@ -239,13 +239,14 @@ public static void main(String[] args) throws Exception { } } - if (localStore instanceof ZKMetadataStore && configStore instanceof ZKMetadataStore) { - String uriStr = bkConf.getMetadataServiceUri().replace("metadata-store:", "zk://"); + 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);