[Issue 5935] Support multi pulsar clusters to use the same bk cluster

conf/broker.conf

@@ -478,6 +478,11 @@ httpMaxRequestSize=-1
 
 ### --- BookKeeper Client --- ###
 
+# Metadata service uri that bookkeeper is used for loading corresponding metadata driver
+# and resolving its metadata service location.
+# For example: zk+hierarchical://localhost:2181/ledgers
+bookkeeperMetadataServiceUri=
+
 # Authentication plugin to use when connecting to bookies
 bookkeeperClientAuthenticationPlugin=
 

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java

@@ -816,6 +816,12 @@ public class ServiceConfiguration implements PulsarConfiguration {
     private String kinitCommand = "/usr/bin/kinit";
 
     /**** --- BookKeeper Client --- ****/
+    @FieldContext(
+        category = CATEGORY_STORAGE_BK,
+        doc = "Metadata service uri that bookkeeper is used for loading corresponding metadata driver"
+            + " and resolving its metadata service location"
+    )
+    private String bookkeeperMetadataServiceUri;
     @FieldContext(
         category = CATEGORY_STORAGE_BK,
         doc = "Authentication plugin to use when connecting to bookies"

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

@@ -37,6 +37,7 @@
 import org.apache.bookkeeper.client.RackawareEnsemblePlacementPolicy;
 import org.apache.bookkeeper.client.RegionAwareEnsemblePlacementPolicy;
 import org.apache.bookkeeper.conf.ClientConfiguration;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.pulsar.common.protocol.Commands;
 import org.apache.pulsar.zookeeper.ZkBookieRackAffinityMapping;
 import org.apache.pulsar.zookeeper.ZkIsolatedBookieEnsemblePlacementPolicy;
@@ -66,7 +67,6 @@ public BookKeeper create(ServiceConfiguration conf, ZooKeeper zkClient,
         try {
             return BookKeeper.forConfig(bkConf)
                     .allocator(PulsarByteBufAllocator.DEFAULT)
-                    .zk(zkClient)
                     .build();
         } catch (InterruptedException | BKException e) {
             throw new IOException(e);
@@ -106,6 +106,13 @@ ClientConfiguration createBkClientConfiguration(ServiceConfiguration conf) {
         bkConf.setNettyMaxFrameSizeBytes(conf.getMaxMessageSize() + Commands.MESSAGE_SIZE_FRAME_PADDING);
         bkConf.setDiskWeightBasedPlacementEnabled(conf.isBookkeeperDiskWeightBasedPlacementEnabled());
 
+        if (StringUtils.isNotBlank(conf.getBookkeeperMetadataServiceUri())) {
+            bkConf.setMetadataServiceUri(conf.getBookkeeperMetadataServiceUri());
+        } else {
+            String metadataServiceUri = PulsarService.bookieMetadataServiceUri(conf);
+            bkConf.setMetadataServiceUri(metadataServiceUri);
+        }
+
         if (conf.isBookkeeperClientHealthCheckEnabled()) {
             bkConf.enableBookieHealthCheck();
             bkConf.setBookieHealthCheckInterval(conf.getBookkeeperHealthCheckIntervalSec(), TimeUnit.SECONDS);

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

@@ -59,7 +59,7 @@ public ManagedLedgerClientFactory(ServiceConfiguration conf, ZooKeeper zkClient,
         managedLedgerFactoryConfig.setCopyEntriesInCache(conf.isManagedLedgerCacheCopyEntries());
 
         this.defaultBkClient = bookkeeperProvider.create(conf, zkClient, Optional.empty(), null);
-        
+
         BookkeeperFactoryForCustomEnsemblePlacementPolicy bkFactory = (
                 EnsemblePlacementPolicyConfig ensemblePlacementPolicyConfig) -> {
             BookKeeper bkClient = null;

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

@@ -66,6 +66,7 @@
 import org.apache.bookkeeper.mledger.offload.OffloaderUtils;
 import org.apache.bookkeeper.mledger.offload.Offloaders;
 import org.apache.bookkeeper.util.ZkUtils;
+import org.apache.commons.configuration.ConfigurationException;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.commons.lang3.builder.ReflectionToStringBuilder;
 import org.apache.pulsar.PulsarVersion;
@@ -721,6 +722,29 @@ public ZooKeeper getZkClient() {
         return this.localZooKeeperConnectionProvider.getLocalZooKeeper();
     }
 
+    /**
+     * Get default bookkeeper metadata service uri.
+     */
+    public String getMetadataServiceUri() {
+        return bookieMetadataServiceUri(this.getConfiguration());
+    }
+
+    public InternalConfigurationData getInternalConfigurationData() {
+
+        String metadataServiceUri = getMetadataServiceUri();
+
+        if (StringUtils.isNotBlank(config.getBookkeeperMetadataServiceUri())) {
+            metadataServiceUri = this.getConfiguration().getBookkeeperMetadataServiceUri();
+        }
+
+        return new InternalConfigurationData(
+            this.getConfiguration().getZookeeperServers(),
+            this.getConfiguration().getConfigurationStoreServers(),
+            new ClientConfiguration().getZkLedgersRootPath(),
+            metadataServiceUri,
+            this.getWorkerConfig().map(wc -> wc.getStateStorageServiceUrl()).orElse(null));
+    }
+
     public ConfigurationCacheService getConfigurationCache() {
         return configurationCacheService;
     }
@@ -1064,6 +1088,26 @@ public String getSafeBrokerServiceUrl() {
         return brokerServiceUrl != null ? brokerServiceUrl : brokerServiceUrlTls;
     }
 
+    /**
+     * Get bookkeeper metadata service uri.
+     *
+     * @param config broker configuration
+     * @return the metadata service uri that bookkeeper is used
+     */
+    public static String bookieMetadataServiceUri(ServiceConfiguration config) {
+        ClientConfiguration bkConf = new ClientConfiguration();
+        // init bookkeeper metadata service uri
+        String metadataServiceUri = null;
+        try {
+            String zkServers = config.getZookeeperServers();
+            bkConf.setZkServers(zkServers);
+            metadataServiceUri = bkConf.getMetadataServiceUri();
+        } catch (ConfigurationException e) {
+            LOG.error("Failed to get bookkeeper metadata service uri", e);
+        }
+        return metadataServiceUri;
+    }
+
     private void startWorkerService(AuthenticationService authenticationService,
                                     AuthorizationService authorizationService)
             throws InterruptedException, IOException, KeeperException {
@@ -1153,21 +1197,15 @@ private void startWorkerService(AuthenticationService authenticationService,
                 throw e;
             }
 
-            InternalConfigurationData internalConf = new InternalConfigurationData(
-                    this.getConfiguration().getZookeeperServers(),
-                    this.getConfiguration().getConfigurationStoreServers(),
-                    new ClientConfiguration().getZkLedgersRootPath(),
-                    this.getWorkerConfig().map(wc -> wc.getStateStorageServiceUrl()).orElse(null));
+            InternalConfigurationData internalConf = this.getInternalConfigurationData();
 
             URI dlogURI;
             try {
                 // initializing dlog namespace for function worker
-                dlogURI = WorkerUtils.initializeDlogNamespace(
-                        internalConf.getZookeeperServers(),
-                        internalConf.getLedgersRootPath());
+                dlogURI = WorkerUtils.initializeDlogNamespace(internalConf);
             } catch (IOException ioe) {
-                LOG.error("Failed to initialize dlog namespace at zookeeper {} for storing function packages",
-                        internalConf.getZookeeperServers(), ioe);
+                LOG.error("Failed to initialize dlog namespace with zookeeper {} at at metadata service uri {} for storing function packages",
+                    internalConf.getZookeeperServers(), internalConf.getBookkeeperMetadataServiceUri(), ioe);
                 throw ioe;
             }
             LOG.info("Function worker service setup completed");

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/BrokersBase.java

@@ -239,12 +239,7 @@ private synchronized void updateDynamicConfigurationOnZk(String configName, Stri
     @Path("/internal-configuration")
     @ApiOperation(value = "Get the internal configuration data", response = InternalConfigurationData.class)
     public InternalConfigurationData getInternalConfigurationData() {
-        ClientConfiguration conf = new ClientConfiguration();
-        return new InternalConfigurationData(
-            pulsar().getConfiguration().getZookeeperServers(),
-            pulsar().getConfiguration().getConfigurationStoreServers(),
-            conf.getZkLedgersRootPath(),
-            pulsar().getWorkerConfig().map(wc -> wc.getStateStorageServiceUrl()).orElse(null));
+        return pulsar().getInternalConfigurationData();
     }
 
     @GET

pulsar-broker/src/test/java/org/apache/pulsar/broker/BookKeeperClientFactoryImplTest.java

@@ -28,10 +28,12 @@
 import static org.testng.Assert.assertFalse;
 import static org.testng.Assert.assertNull;
 import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.fail;
 
 import java.util.concurrent.atomic.AtomicReference;
 import org.apache.bookkeeper.conf.ClientConfiguration;
 import org.apache.bookkeeper.net.CachedDNSToSwitchMapping;
+import org.apache.commons.configuration.ConfigurationException;
 import org.apache.pulsar.zookeeper.ZkBookieRackAffinityMapping;
 import org.apache.pulsar.zookeeper.ZooKeeperCache;
 import org.apache.zookeeper.ZooKeeper;
@@ -154,6 +156,7 @@ public void testSetDefaultEnsemblePlacementPolicyRackAwareEnabledChangedValues()
     public void testSetDiskWeightBasedPlacementEnabled() {
         BookKeeperClientFactoryImpl factory = new BookKeeperClientFactoryImpl();
         ServiceConfiguration conf = new ServiceConfiguration();
+        conf.setZookeeperServers("localhost:2181");
         assertFalse(factory.createBkClientConfiguration(conf).getDiskWeightBasedPlacementEnabled());
         conf.setBookkeeperDiskWeightBasedPlacementEnabled(true);
         assertTrue(factory.createBkClientConfiguration(conf).getDiskWeightBasedPlacementEnabled());
@@ -163,9 +166,27 @@ public void testSetDiskWeightBasedPlacementEnabled() {
     public void testSetExplicitLacInterval() {
         BookKeeperClientFactoryImpl factory = new BookKeeperClientFactoryImpl();
         ServiceConfiguration conf = new ServiceConfiguration();
+        conf.setZookeeperServers("localhost:2181");
         assertEquals(factory.createBkClientConfiguration(conf).getExplictLacInterval(), 0);
         conf.setBookkeeperExplicitLacIntervalInMills(5);
         assertEquals(factory.createBkClientConfiguration(conf).getExplictLacInterval(), 5);
     }
 
+    @Test
+    public void testSetMetadataServiceUri() {
+        BookKeeperClientFactoryImpl factory = new BookKeeperClientFactoryImpl();
+        ServiceConfiguration conf = new ServiceConfiguration();
+        conf.setZookeeperServers("localhost:2181");
+        try {
+            String defaultUri = "zk+null://localhost:2181/ledgers";
+            assertEquals(factory.createBkClientConfiguration(conf).getMetadataServiceUri(), defaultUri);
+            String expectedUri = "zk+hierarchical://localhost:2181/chroot/ledgers";
+            conf.setBookkeeperMetadataServiceUri(expectedUri);
+            assertEquals(factory.createBkClientConfiguration(conf).getMetadataServiceUri(), expectedUri);
+        } catch (ConfigurationException e) {
+            e.printStackTrace();
+            fail("Get metadata service uri should be successful", e);
+        }
+    }
+
 }

pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminTest.java

@@ -221,6 +221,7 @@ void internalConfiguration() throws Exception {
             pulsar.getConfiguration().getZookeeperServers(),
             pulsar.getConfiguration().getConfigurationStoreServers(),
             new ClientConfiguration().getZkLedgersRootPath(),
+            pulsar.getMetadataServiceUri(),
             pulsar.getWorkerConfig().map(wc -> wc.getStateStorageServiceUrl()).orElse(null));
 
         assertEquals(brokers.getInternalConfigurationData(), expectedData);

pulsar-common/src/main/java/org/apache/pulsar/common/conf/InternalConfigurationData.java

@@ -28,7 +28,9 @@ public class InternalConfigurationData {
 
     private String zookeeperServers;
     private String configurationStoreServers;
+    @Deprecated
     private String ledgersRootPath;
+    private String bookkeeperMetadataServiceUri;
     private String stateStorageServiceUrl;
 
     public InternalConfigurationData() {
@@ -37,10 +39,12 @@ public InternalConfigurationData() {
     public InternalConfigurationData(String zookeeperServers,
                                      String configurationStoreServers,
                                      String ledgersRootPath,
+                                     String bookkeeperMetadataServiceUri,
                                      String stateStorageServiceUrl) {
         this.zookeeperServers = zookeeperServers;
         this.configurationStoreServers = configurationStoreServers;
         this.ledgersRootPath = ledgersRootPath;
+        this.bookkeeperMetadataServiceUri = bookkeeperMetadataServiceUri;
         this.stateStorageServiceUrl = stateStorageServiceUrl;
     }
 
@@ -52,10 +56,16 @@ public String getConfigurationStoreServers() {
         return configurationStoreServers;
     }
 
+    /** @deprecated */
+    @Deprecated
     public String getLedgersRootPath() {
         return ledgersRootPath;
     }
 
+    public String getBookkeeperMetadataServiceUri() {
+        return bookkeeperMetadataServiceUri;
+    }
+
     public String getStateStorageServiceUrl() {
         return stateStorageServiceUrl;
     }
@@ -69,12 +79,17 @@ public boolean equals(Object obj) {
         return Objects.equals(zookeeperServers, other.zookeeperServers)
             && Objects.equals(configurationStoreServers, other.configurationStoreServers)
             && Objects.equals(ledgersRootPath, other.ledgersRootPath)
+            && Objects.equals(bookkeeperMetadataServiceUri, other.bookkeeperMetadataServiceUri)
             && Objects.equals(stateStorageServiceUrl, other.stateStorageServiceUrl);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(zookeeperServers, configurationStoreServers, ledgersRootPath, stateStorageServiceUrl);
+        return Objects.hash(zookeeperServers,
+                configurationStoreServers,
+                ledgersRootPath,
+                bookkeeperMetadataServiceUri,
+                stateStorageServiceUrl);
     }
 
     @Override
@@ -83,6 +98,7 @@ public String toString() {
             .add("zookeeperServers", zookeeperServers)
             .add("configurationStoreServers", configurationStoreServers)
             .add("ledgersRootPath", ledgersRootPath)
+            .add("bookkeeperMetadataServiceUri", bookkeeperMetadataServiceUri)
             .add("stateStorageServiceUrl", stateStorageServiceUrl)
             .toString();
     }

pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/Worker.java

@@ -143,12 +143,10 @@ private static URI initialize(WorkerConfig workerConfig)
         // initialize the dlog namespace
         // TODO: move this as part of pulsar cluster initialization later
         try {
-            return WorkerUtils.initializeDlogNamespace(
-                    internalConf.getZookeeperServers(),
-                    internalConf.getLedgersRootPath());
+            return WorkerUtils.initializeDlogNamespace(internalConf);
         } catch (IOException ioe) {
-            log.error("Failed to initialize dlog namespace at zookeeper {} for storing function packages",
-                    internalConf.getZookeeperServers(), ioe);
+            log.error("Failed to initialize dlog namespace with zookeeper {} at metadata service uri {} for storing function packages",
+                internalConf.getZookeeperServers(), internalConf.getBookkeeperMetadataServiceUri(), ioe);
             throw ioe;
         }
     }

pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerUtils.java

@@ -32,6 +32,7 @@
 import org.apache.pulsar.client.api.ClientBuilder;
 import org.apache.pulsar.client.api.PulsarClient;
 import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.common.conf.InternalConfigurationData;
 import org.apache.pulsar.common.policies.data.FunctionStats;
 import org.apache.pulsar.functions.proto.Function;
 import org.apache.pulsar.functions.proto.InstanceCommunication;
@@ -146,10 +147,14 @@ public static DistributedLogConfiguration getDlogConf(WorkerConfig workerConfig)
         return conf;
     }
 
-    public static URI initializeDlogNamespace(String zkServers, String ledgersRootPath) throws IOException {
-        BKDLConfig dlConfig = new BKDLConfig(zkServers, ledgersRootPath);
+    public static URI initializeDlogNamespace(InternalConfigurationData internalConf) throws IOException {
+        String zookeeperServers = internalConf.getZookeeperServers();
+        URI metadataServiceUri = URI.create(internalConf.getBookkeeperMetadataServiceUri());
+        String ledgersStoreServers = metadataServiceUri.getAuthority().replace(";", ",");
+        String ledgersRootPath = metadataServiceUri.getPath();
+        BKDLConfig dlConfig = new BKDLConfig(ledgersStoreServers, ledgersRootPath);
         DLMetadata dlMetadata = DLMetadata.create(dlConfig);
-        URI dlogUri = URI.create(String.format("distributedlog://%s/pulsar/functions", zkServers));
+        URI dlogUri = URI.create(String.format("distributedlog://%s/pulsar/functions", zookeeperServers));
 
         try {
             dlMetadata.create(dlogUri);

site2/docs/reference-configuration.md

@@ -168,6 +168,7 @@ Pulsar brokers are responsible for handling incoming messages from producers, di
 |brokerClientAuthenticationParameters|||
 |athenzDomainNames| Supported Athenz provider domain names(comma separated) for authentication  ||
 |exposePreciseBacklogInPrometheus| Enable expose the precise backlog stats, set false to use published counter and consumed counter to calculate, this would be more efficient but may be inaccurate. |false|
+|bookkeeperMetadataServiceUri| Metadata service uri that bookkeeper is used for loading corresponding metadata driver and resolving its metadata service location. For example: zk+hierarchical://localhost:2181/ledgers ||
 |bookkeeperClientAuthenticationPlugin|  Authentication plugin to use when connecting to bookies ||
 |bookkeeperClientAuthenticationParametersName|  BookKeeper auth plugin implementatation specifics parameters name and values  ||
 |bookkeeperClientAuthenticationParameters|||