[improve][meta] Support tuning Oxia MetadataStoreConfig through metadata-store URIs

conf/bookkeeper.conf

@@ -665,6 +665,10 @@ diskCheckInterval=10000
 #  - metadataServiceUri=zk+hierarchical://my-zk-1:2181/ledgers
 #  - metadataServiceUri=metadata-store:zk:my-zk-1:2281/ledgers
 #  - metadataServiceUri=metadata-store:oxia://oxia-server:6648/bookkeeper
+#  - metadataServiceUri=metadata-store:oxia://oxia-server:6648/bookkeeper?batchingMaxDelayMillis=10&batchingMaxSizeKb=256&numSerDesThreads=4
+# Supported MetadataStoreConfig query parameters for the "metadata-store:" driver are allowReadOnlyOperations,
+# batchingEnabled, batchingMaxDelayMillis, batchingMaxOperations, batchingMaxSizeKb, configFilePath, fsyncEnable,
+# numSerDesThreads, and sessionTimeoutMillis. Other query parameters are passed through to the metadata store provider.
 # If you use metadata-store configuration, you need to configure following items in JVM option:
 # -Dbookkeeper.metadata.client.drivers=org.apache.pulsar.metadata.bookkeeper.PulsarMetadataClientDriver
 # -Dbookkeeper.metadata.bookie.drivers=org.apache.pulsar.metadata.bookkeeper.PulsarMetadataBookieDriver

conf/broker.conf

@@ -1148,6 +1148,11 @@ maxConcurrentHttpRequests=1024
 # Examples:
 # - metadata-store:zk:zk1:2181,zk2:2181,zk3:2181/ledgers
 # - metadata-store:oxia://oxia-server:6648/bookkeeper
+# - metadata-store:oxia://oxia-server:6648/bookkeeper?batchingMaxDelayMillis=10&batchingMaxSizeKb=256&numSerDesThreads=4
+# When using the "metadata-store:" driver with a separated BookKeeper metadata service, supported
+# MetadataStoreConfig query parameters are allowReadOnlyOperations, batchingEnabled, batchingMaxDelayMillis,
+# batchingMaxOperations, batchingMaxSizeKb, configFilePath, fsyncEnable, numSerDesThreads, and
+# sessionTimeoutMillis. Other query parameters are passed through to the metadata store provider.
 # When the value is empty, the broker will default to using broker's metadataStoreUrl config appended with "/ledgers"
 bookkeeperMetadataServiceUri=
 

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

@@ -278,6 +278,13 @@ public void testSetMetadataServiceUriBookkeeperMetadataServiceUri() {
                         .getMetadataServiceUri(), expectedUri);
 
             }
+            {
+                String uri = "metadata-store:oxia://oxia-server:6648/bookkeeper"
+                        + "?batchingMaxDelayMillis=10&batchingMaxSizeKb=256&numSerDesThreads=4";
+                conf.setBookkeeperMetadataServiceUri(uri);
+                assertEquals(factory.createBkClientConfiguration(mock(MetadataStoreExtended.class), conf)
+                        .getMetadataServiceUri(), uri);
+            }
         } catch (ConfigurationException e) {
             e.printStackTrace();
             fail("Get metadata service uri should be successful", e);

pulsar-broker/src/test/java/org/apache/pulsar/common/naming/ServiceConfigurationTest.java

@@ -222,16 +222,18 @@ public void testConfigurationMetadataStoreUrl() throws Exception {
 
     @Test
     public void testBookkeeperMetadataStore() throws Exception {
+        String bookkeeperMetadataServiceUri = "metadata-store:oxia://oxia-server:6648/bookkeeper"
+                + "?batchingMaxDelayMillis=10&batchingMaxSizeKb=256&numSerDesThreads=4";
         String confFile = "metadataStoreUrl=zk1:2181\n"
                 + "configurationMetadataStoreUrl=zk2:2182\n"
-                + "bookkeeperMetadataServiceUri=xx:other-system\n";
+                + "bookkeeperMetadataServiceUri=" + bookkeeperMetadataServiceUri + "\n";
         @Cleanup
         InputStream stream = new ByteArrayInputStream(confFile.getBytes());
         final ServiceConfiguration conf = PulsarConfigurationLoader.create(stream, ServiceConfiguration.class);
 
         assertEquals(conf.getMetadataStoreUrl(), "zk1:2181");
         assertEquals(conf.getConfigurationMetadataStoreUrl(), "zk2:2182");
-        assertEquals(conf.getBookkeeperMetadataStoreUrl(), "xx:other-system");
+        assertEquals(conf.getBookkeeperMetadataStoreUrl(), bookkeeperMetadataServiceUri);
         assertTrue(conf.isConfigurationStoreSeparated());
         assertTrue(conf.isBookkeeperMetadataStoreSeparated());
     }

pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/AbstractMetadataDriver.java

@@ -32,7 +32,6 @@
 import org.apache.bookkeeper.meta.exceptions.Code;
 import org.apache.bookkeeper.meta.exceptions.MetadataException;
 import org.apache.bookkeeper.util.BookKeeperConstants;
-import org.apache.pulsar.metadata.api.MetadataStoreConfig;
 import org.apache.pulsar.metadata.api.MetadataStoreException;
 import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended;
 
@@ -107,27 +106,36 @@ void createMetadataStore() throws MetadataException {
             this.storeInstanceIsOwned = false;
         } else {
 
-            String url;
+            String metadataServiceUri = null;
+            MetadataStoreUrl metadataStoreUrl;
             try {
-                url = conf.getMetadataServiceUri()
-                        .replaceFirst(METADATA_STORE_SCHEME + ":", "")
-                        .replace(";", ",");
+                metadataServiceUri = conf.getMetadataServiceUri();
+                metadataStoreUrl = MetadataStoreUrl.parse(metadataServiceUri);
             } catch (Exception e) {
-                throw new MetadataException(Code.METADATA_SERVICE_ERROR, e);
+                throw metadataStoreCreationException(metadataServiceUri, null, e);
             }
             try {
-                this.store = MetadataStoreExtended.create(url,
-                        MetadataStoreConfig.builder()
-                                .sessionTimeoutMillis(conf.getZkTimeout())
-                                .metadataStoreName(MetadataStoreConfig.METADATA_STORE)
-                                .build());
+                this.store = MetadataStoreExtended.create(metadataStoreUrl.url(),
+                        MetadataStoreConfigQueryParams.createConfig(conf, metadataStoreUrl.configParams()));
                 this.storeInstanceIsOwned = true;
-            } catch (MetadataStoreException e) {
-                throw new MetadataException(Code.METADATA_SERVICE_ERROR, e);
+            } catch (IllegalArgumentException | MetadataStoreException e) {
+                throw metadataStoreCreationException(metadataServiceUri, metadataStoreUrl.url(), e);
             }
         }
     }
 
+    private static MetadataException metadataStoreCreationException(
+            String metadataServiceUri, String metadataStoreUrl, Exception cause) {
+        String message = "Failed to create BookKeeper metadata store";
+        if (metadataServiceUri != null) {
+            message += " from metadataServiceUri '" + metadataServiceUri + "'";
+        }
+        if (metadataStoreUrl != null) {
+            message += " (metadata store URL '" + metadataStoreUrl + "')";
+        }
+        return new MetadataException(Code.METADATA_SERVICE_ERROR, message, cause);
+    }
+
     public String getScheme() {
         return METADATA_STORE_SCHEME;
     }

pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/MetadataStoreConfigQueryParams.java

@@ -0,0 +1,113 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.metadata.bookkeeper;
+
+import java.util.Map;
+import java.util.function.BiConsumer;
+import org.apache.bookkeeper.conf.AbstractConfiguration;
+import org.apache.pulsar.metadata.api.MetadataStoreConfig;
+import org.apache.pulsar.metadata.api.MetadataStoreConfig.MetadataStoreConfigBuilder;
+
+class MetadataStoreConfigQueryParams {
+    private static final Map<String, BiConsumer<MetadataStoreConfigBuilder, String>> PARAM_SETTERS = Map.ofEntries(
+            booleanParam("allowReadOnlyOperations", MetadataStoreConfigBuilder::allowReadOnlyOperations),
+            booleanParam("batchingEnabled", MetadataStoreConfigBuilder::batchingEnabled),
+            nonNegativeIntParam("batchingMaxDelayMillis", MetadataStoreConfigBuilder::batchingMaxDelayMillis),
+            positiveIntParam("batchingMaxOperations", MetadataStoreConfigBuilder::batchingMaxOperations),
+            positiveIntParam("batchingMaxSizeKb", MetadataStoreConfigBuilder::batchingMaxSizeKb),
+            Map.entry("configFilePath", MetadataStoreConfigBuilder::configFilePath),
+            booleanParam("fsyncEnable", MetadataStoreConfigBuilder::fsyncEnable),
+            positiveIntParam("numSerDesThreads", MetadataStoreConfigBuilder::numSerDesThreads),
+            positiveIntParam("sessionTimeoutMillis", MetadataStoreConfigBuilder::sessionTimeoutMillis));
+
+    private MetadataStoreConfigQueryParams() {
+    }
+
+    static boolean contains(String key) {
+        return PARAM_SETTERS.containsKey(key);
+    }
+
+    @SuppressWarnings("rawtypes")
+    static MetadataStoreConfig createConfig(AbstractConfiguration conf, Map<String, String> configParams) {
+        MetadataStoreConfigBuilder builder = MetadataStoreConfig.builder()
+                .sessionTimeoutMillis(conf.getZkTimeout())
+                .metadataStoreName(MetadataStoreConfig.METADATA_STORE);
+        configParams.forEach((key, value) -> {
+            BiConsumer<MetadataStoreConfigBuilder, String> setter = PARAM_SETTERS.get(key);
+            if (setter == null) {
+                throw new IllegalArgumentException("Unsupported MetadataStoreConfig query parameter '" + key
+                        + "'. Supported parameters are " + PARAM_SETTERS.keySet());
+            }
+            setter.accept(builder, value);
+        });
+        return builder.build();
+    }
+
+    private static Map.Entry<String, BiConsumer<MetadataStoreConfigBuilder, String>> booleanParam(
+            String key, BiConsumer<MetadataStoreConfigBuilder, Boolean> setter) {
+        return Map.entry(key, (builder, value) -> setter.accept(builder, parseBoolean(key, value)));
+    }
+
+    private static Map.Entry<String, BiConsumer<MetadataStoreConfigBuilder, String>> nonNegativeIntParam(
+            String key, BiConsumer<MetadataStoreConfigBuilder, Integer> setter) {
+        return Map.entry(key, (builder, value) -> setter.accept(builder, parseNonNegativeInt(key, value)));
+    }
+
+    private static Map.Entry<String, BiConsumer<MetadataStoreConfigBuilder, String>> positiveIntParam(
+            String key, BiConsumer<MetadataStoreConfigBuilder, Integer> setter) {
+        return Map.entry(key, (builder, value) -> setter.accept(builder, parsePositiveInt(key, value)));
+    }
+
+    private static boolean parseBoolean(String key, String value) {
+        if ("true".equalsIgnoreCase(value)) {
+            return true;
+        } else if ("false".equalsIgnoreCase(value)) {
+            return false;
+        }
+        throw new IllegalArgumentException("MetadataStoreConfig query parameter '" + key
+                + "' must be true or false");
+    }
+
+    private static int parseNonNegativeInt(String key, String value) {
+        int intValue = parseInt(key, value);
+        if (intValue < 0) {
+            throw new IllegalArgumentException("MetadataStoreConfig query parameter '" + key
+                    + "' must be greater than or equal to 0");
+        }
+        return intValue;
+    }
+
+    private static int parsePositiveInt(String key, String value) {
+        int intValue = parseInt(key, value);
+        if (intValue <= 0) {
+            throw new IllegalArgumentException("MetadataStoreConfig query parameter '" + key
+                    + "' must be greater than 0");
+        }
+        return intValue;
+    }
+
+    private static int parseInt(String key, String value) {
+        try {
+            return Integer.parseInt(value);
+        } catch (NumberFormatException e) {
+            throw new IllegalArgumentException("MetadataStoreConfig query parameter '" + key
+                    + "' must be an integer", e);
+        }
+    }
+}

pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/MetadataStoreUrl.java

@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.metadata.bookkeeper;
+
+import java.net.URLDecoder;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+record MetadataStoreUrl(String url, Map<String, String> configParams) {
+    static MetadataStoreUrl parse(String metadataServiceUri) {
+        String metadataStoreUrl = removeMetadataStoreScheme(metadataServiceUri);
+        int queryStart = metadataStoreUrl.indexOf('?');
+        if (queryStart < 0) {
+            return new MetadataStoreUrl(metadataStoreUrl.replace(";", ","), Map.of());
+        }
+
+        String url = metadataStoreUrl.substring(0, queryStart).replace(";", ",");
+        String query = metadataStoreUrl.substring(queryStart + 1);
+        if (query.isEmpty()) {
+            return new MetadataStoreUrl(url, Map.of());
+        }
+
+        Map<String, String> configParams = new LinkedHashMap<>();
+        List<String> providerParams = new ArrayList<>();
+        for (String param : query.split("&", -1)) {
+            if (param.isEmpty()) {
+                continue;
+            }
+            int separator = param.indexOf('=');
+            String key = decodeQueryParam(separator > 0 ? param.substring(0, separator) : param);
+            if (!MetadataStoreConfigQueryParams.contains(key)) {
+                providerParams.add(param);
+                continue;
+            }
+            if (separator <= 0) {
+                throw new IllegalArgumentException("Invalid MetadataStoreConfig query parameter '" + param
+                        + "'. Expected key=value");
+            }
+
+            String value = decodeQueryParam(param.substring(separator + 1));
+            configParams.put(key, value);
+        }
+
+        if (!providerParams.isEmpty()) {
+            url += "?" + String.join("&", providerParams);
+        }
+        return new MetadataStoreUrl(url, Map.copyOf(configParams));
+    }
+
+    private static String removeMetadataStoreScheme(String metadataServiceUri) {
+        String prefix = AbstractMetadataDriver.METADATA_STORE_SCHEME + ":";
+        if (metadataServiceUri.startsWith(prefix)) {
+            return metadataServiceUri.substring(prefix.length());
+        }
+        return metadataServiceUri;
+    }
+
+    private static String decodeQueryParam(String value) {
+        return URLDecoder.decode(value, StandardCharsets.UTF_8);
+    }
+}