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..52d7e0d8e3bb5 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,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,22 +211,27 @@ 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 */)) { @@ -215,22 +239,25 @@ public static void main(String[] args) throws Exception { } } + 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()) {