You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by ma...@apache.org on 2026/07/06 04:34:08 UTC
(pulsar) 01/01: [improve][meta] Support tuning Oxia MetadataStoreConfig through metadata-store URIs (#26067)
This is an automated email from the ASF dual-hosted git repository.
mattisonchao pushed a commit to branch fix/cherry-pick-26067-branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 14a70642cf9fac847dd12177c5d6861d329b9990
Author: void-ptr974 <vo...@protonmail.com>
AuthorDate: Mon Jun 22 00:02:00 2026 +0800
[improve][meta] Support tuning Oxia MetadataStoreConfig through metadata-store URIs (#26067)
(cherry picked from commit 9b5bdbec0e8b0744cc4a458b37d47d06d791dce1)
---
conf/bookkeeper.conf | 4 +
conf/broker.conf | 5 +
.../broker/BookKeeperClientFactoryImplTest.java | 7 +
.../common/naming/ServiceConfigurationTest.java | 6 +-
.../bookkeeper/AbstractMetadataDriver.java | 34 ++-
.../bookkeeper/MetadataStoreConfigQueryParams.java | 113 +++++++++
.../metadata/bookkeeper/MetadataStoreUrl.java | 80 ++++++
.../bookkeeper/AbstractMetadataDriverTest.java | 277 +++++++++++++++++++++
8 files changed, 511 insertions(+), 15 deletions(-)
diff --git a/conf/bookkeeper.conf b/conf/bookkeeper.conf
index a482a28cc75..ff672e56ad3 100644
--- a/conf/bookkeeper.conf
+++ b/conf/bookkeeper.conf
@@ -666,6 +666,10 @@ diskCheckInterval=10000
# - metadataServiceUri=etcd+hierarchical:http://my-etcd:2379
# - 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
diff --git a/conf/broker.conf b/conf/broker.conf
index 066138ef98d..84ccd1d852b 100644
--- a/conf/broker.conf
+++ b/conf/broker.conf
@@ -1097,6 +1097,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=
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/BookKeeperClientFactoryImplTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/BookKeeperClientFactoryImplTest.java
index fdf8dea0f04..f3c8acd365f 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/BookKeeperClientFactoryImplTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/BookKeeperClientFactoryImplTest.java
@@ -278,6 +278,13 @@ public class BookKeeperClientFactoryImplTest {
.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);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/common/naming/ServiceConfigurationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/common/naming/ServiceConfigurationTest.java
index ed551c8c1bb..e9662e85f8d 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/common/naming/ServiceConfigurationTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/common/naming/ServiceConfigurationTest.java
@@ -221,16 +221,18 @@ public class ServiceConfigurationTest {
@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());
}
diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/AbstractMetadataDriver.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/AbstractMetadataDriver.java
index 435f94b05dc..1ebeecea6e4 100644
--- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/AbstractMetadataDriver.java
+++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/AbstractMetadataDriver.java
@@ -32,7 +32,6 @@ import org.apache.bookkeeper.meta.LegacyHierarchicalLedgerManagerFactory;
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;
@@ -105,27 +104,36 @@ public abstract class AbstractMetadataDriver implements Closeable {
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;
}
diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/MetadataStoreConfigQueryParams.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/MetadataStoreConfigQueryParams.java
new file mode 100644
index 00000000000..61866232c64
--- /dev/null
+++ b/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);
+ }
+ }
+}
diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/MetadataStoreUrl.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/MetadataStoreUrl.java
new file mode 100644
index 00000000000..b7a52369b8d
--- /dev/null
+++ b/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);
+ }
+}
diff --git a/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/bookkeeper/AbstractMetadataDriverTest.java b/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/bookkeeper/AbstractMetadataDriverTest.java
new file mode 100644
index 00000000000..e4b64b4eb1e
--- /dev/null
+++ b/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/bookkeeper/AbstractMetadataDriverTest.java
@@ -0,0 +1,277 @@
+/*
+ * 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 static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.catchThrowable;
+import io.opentelemetry.api.OpenTelemetry;
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import org.apache.bookkeeper.conf.AbstractConfiguration;
+import org.apache.bookkeeper.conf.ClientConfiguration;
+import org.apache.bookkeeper.meta.exceptions.MetadataException;
+import org.apache.pulsar.metadata.api.GetResult;
+import org.apache.pulsar.metadata.api.MetadataStore;
+import org.apache.pulsar.metadata.api.MetadataStoreConfig;
+import org.apache.pulsar.metadata.api.MetadataStoreException;
+import org.apache.pulsar.metadata.api.MetadataStoreProvider;
+import org.apache.pulsar.metadata.api.Option;
+import org.apache.pulsar.metadata.api.Stat;
+import org.apache.pulsar.metadata.impl.AbstractMetadataStore;
+import org.apache.pulsar.metadata.impl.MetadataStoreFactoryImpl;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+public class AbstractMetadataDriverTest {
+ private static final String CAPTURING_METADATA_STORE_URL = "config-capture://host:4181/bookkeeper";
+
+ private String originalMetadataStoreProviders;
+
+ @BeforeClass
+ public void addCapturingMetadataStoreProvider() {
+ originalMetadataStoreProviders = System.getProperty(MetadataStoreFactoryImpl.METADATASTORE_PROVIDERS_PROPERTY);
+ String providers = CapturingMetadataStoreProvider.class.getName();
+ if (originalMetadataStoreProviders != null && !originalMetadataStoreProviders.isBlank()) {
+ providers = originalMetadataStoreProviders + "," + providers;
+ }
+ System.setProperty(MetadataStoreFactoryImpl.METADATASTORE_PROVIDERS_PROPERTY, providers);
+ }
+
+ @AfterClass(alwaysRun = true)
+ public void restoreMetadataStoreProviders() {
+ if (originalMetadataStoreProviders != null) {
+ System.setProperty(MetadataStoreFactoryImpl.METADATASTORE_PROVIDERS_PROPERTY,
+ originalMetadataStoreProviders);
+ } else {
+ System.clearProperty(MetadataStoreFactoryImpl.METADATASTORE_PROVIDERS_PROPERTY);
+ }
+ }
+
+ @Test
+ public void testMetadataStoreConfigCanBeConfiguredWithMetadataServiceUriQuery() throws Exception {
+ CapturingMetadataStoreProvider.reset();
+ ClientConfiguration conf = new ClientConfiguration();
+ conf.setZkTimeout(12_345);
+ conf.setMetadataServiceUri("metadata-store:" + CAPTURING_METADATA_STORE_URL
+ + "?batchingEnabled=false"
+ + "&batchingMaxDelayMillis=10"
+ + "&batchingMaxOperations=11"
+ + "&batchingMaxSizeKb=12"
+ + "&numSerDesThreads=3"
+ + "&sessionTimeoutMillis=40000"
+ + "&allowReadOnlyOperations=true"
+ + "&configFilePath=%2Ftmp%2Foxia-client.conf"
+ + "&fsyncEnable=false");
+
+ createMetadataStore(conf);
+
+ assertThat(CapturingMetadataStoreProvider.metadataURL).isEqualTo(CAPTURING_METADATA_STORE_URL);
+ MetadataStoreConfig metadataStoreConfig = CapturingMetadataStoreProvider.metadataStoreConfig;
+ assertThat(metadataStoreConfig.getMetadataStoreName()).isEqualTo(MetadataStoreConfig.METADATA_STORE);
+ assertThat(metadataStoreConfig.isBatchingEnabled()).isFalse();
+ assertThat(metadataStoreConfig.getBatchingMaxDelayMillis()).isEqualTo(10);
+ assertThat(metadataStoreConfig.getBatchingMaxOperations()).isEqualTo(11);
+ assertThat(metadataStoreConfig.getBatchingMaxSizeKb()).isEqualTo(12);
+ assertThat(metadataStoreConfig.getNumSerDesThreads()).isEqualTo(3);
+ assertThat(metadataStoreConfig.getSessionTimeoutMillis()).isEqualTo(40_000);
+ assertThat(metadataStoreConfig.isAllowReadOnlyOperations()).isTrue();
+ assertThat(metadataStoreConfig.getConfigFilePath()).isEqualTo("/tmp/oxia-client.conf");
+ assertThat(metadataStoreConfig.isFsyncEnable()).isFalse();
+ }
+
+ @Test
+ public void testUnknownMetadataServiceUriQueryParamsArePassedToProvider() throws Exception {
+ CapturingMetadataStoreProvider.reset();
+ ClientConfiguration conf = new ClientConfiguration();
+ conf.setMetadataServiceUri("metadata-store:" + CAPTURING_METADATA_STORE_URL
+ + "?providerParam=1"
+ + "&batchingMaxDelayMillis=10"
+ + "&providerFlag"
+ + "&batchingMaxDelayMillis=20"
+ + "&numSerDesThreads=3");
+
+ createMetadataStore(conf);
+
+ assertThat(CapturingMetadataStoreProvider.metadataURL)
+ .isEqualTo(CAPTURING_METADATA_STORE_URL + "?providerParam=1&providerFlag");
+ MetadataStoreConfig metadataStoreConfig = CapturingMetadataStoreProvider.metadataStoreConfig;
+ assertThat(metadataStoreConfig.getBatchingMaxDelayMillis()).isEqualTo(20);
+ assertThat(metadataStoreConfig.getNumSerDesThreads()).isEqualTo(3);
+ }
+
+ @Test
+ public void testMetadataServiceUriWithoutQueryUsesExistingDefaults() throws Exception {
+ CapturingMetadataStoreProvider.reset();
+ ClientConfiguration conf = new ClientConfiguration();
+ conf.setZkTimeout(12_345);
+ conf.setMetadataServiceUri("metadata-store:config-capture://host1:4181;host2:4181/bookkeeper");
+
+ createMetadataStore(conf);
+
+ assertThat(CapturingMetadataStoreProvider.metadataURL)
+ .isEqualTo("config-capture://host1:4181,host2:4181/bookkeeper");
+ MetadataStoreConfig metadataStoreConfig = CapturingMetadataStoreProvider.metadataStoreConfig;
+ assertThat(metadataStoreConfig.getMetadataStoreName()).isEqualTo(MetadataStoreConfig.METADATA_STORE);
+ assertThat(metadataStoreConfig.getSessionTimeoutMillis()).isEqualTo(12_345);
+ assertThat(metadataStoreConfig.isBatchingEnabled()).isTrue();
+ assertThat(metadataStoreConfig.getBatchingMaxDelayMillis()).isEqualTo(5);
+ assertThat(metadataStoreConfig.getBatchingMaxOperations()).isEqualTo(1_000);
+ assertThat(metadataStoreConfig.getBatchingMaxSizeKb()).isEqualTo(128);
+ assertThat(metadataStoreConfig.getNumSerDesThreads()).isEqualTo(1);
+ }
+
+ @Test(dataProvider = "invalidMetadataStoreConfigQueryParams")
+ public void testInvalidMetadataStoreConfigQueryParamFails(String query, String invalidParameter,
+ boolean expectBackendUrlInErrorMessage) {
+ assertInvalidConfigQuery(query, invalidParameter, expectBackendUrlInErrorMessage);
+ }
+
+ private static void assertInvalidConfigQuery(String query, String invalidParameter,
+ boolean expectBackendUrlInErrorMessage) {
+ ClientConfiguration conf = new ClientConfiguration();
+ String metadataServiceUri = "metadata-store:" + CAPTURING_METADATA_STORE_URL + query;
+ conf.setMetadataServiceUri(metadataServiceUri);
+
+ Throwable error = catchThrowable(() -> createMetadataStore(conf));
+
+ assertThat(error)
+ .isInstanceOf(MetadataException.class)
+ .hasCauseInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("BookKeeper metadata store")
+ .hasMessageContaining("metadataServiceUri")
+ .hasMessageContaining(metadataServiceUri);
+ if (expectBackendUrlInErrorMessage) {
+ assertThat(error).hasMessageContaining(CAPTURING_METADATA_STORE_URL);
+ }
+ assertThat(error.getCause()).hasMessageContaining(invalidParameter);
+ }
+
+ @DataProvider(name = "invalidMetadataStoreConfigQueryParams")
+ public Object[][] invalidMetadataStoreConfigQueryParams() {
+ return new Object[][] {
+ {"?batchingEnabled=maybe", "batchingEnabled", true},
+ {"?batchingMaxDelayMillis=-1", "batchingMaxDelayMillis", true},
+ {"?batchingMaxOperations=0", "batchingMaxOperations", true},
+ {"?numSerDesThreads=abc", "numSerDesThreads", true},
+ {"?batchingEnabled", "batchingEnabled", false}
+ };
+ }
+
+ @Test
+ public void testMetadataStoreInstancePathIgnoresMetadataServiceUriQuery() throws Exception {
+ CapturingMetadataStoreProvider.reset();
+ ClientConfiguration conf = new ClientConfiguration();
+ MetadataStore sharedStore = new CapturingMetadataStore();
+ conf.setProperty(AbstractMetadataDriver.METADATA_STORE_INSTANCE, sharedStore);
+ conf.setMetadataServiceUri("metadata-store:" + CAPTURING_METADATA_STORE_URL
+ + "?unknown=1&batchingMaxDelayMillis=bad");
+
+ TestMetadataDriver driver = new TestMetadataDriver();
+ try {
+ driver.createMetadataStore(conf);
+
+ assertThat(driver.store()).isSameAs(sharedStore);
+ assertThat(CapturingMetadataStoreProvider.metadataURL).isNull();
+ assertThat(CapturingMetadataStoreProvider.metadataStoreConfig).isNull();
+ } finally {
+ driver.close();
+ }
+ }
+
+ @SuppressWarnings("rawtypes")
+ private static void createMetadataStore(AbstractConfiguration conf) throws MetadataException {
+ TestMetadataDriver driver = new TestMetadataDriver();
+ try {
+ driver.createMetadataStore(conf);
+ } finally {
+ driver.close();
+ }
+ }
+
+ @SuppressWarnings("rawtypes")
+ private static class TestMetadataDriver extends AbstractMetadataDriver {
+ private void createMetadataStore(AbstractConfiguration conf) throws MetadataException {
+ this.conf = conf;
+ super.createMetadataStore();
+ }
+
+ private MetadataStore store() {
+ return store;
+ }
+ }
+
+ public static class CapturingMetadataStoreProvider implements MetadataStoreProvider {
+ private static volatile String metadataURL;
+ private static volatile MetadataStoreConfig metadataStoreConfig;
+
+ static void reset() {
+ metadataURL = null;
+ metadataStoreConfig = null;
+ }
+
+ @Override
+ public String urlScheme() {
+ return "config-capture";
+ }
+
+ @Override
+ public MetadataStore create(String metadataURL, MetadataStoreConfig metadataStoreConfig,
+ boolean enableSessionWatcher) throws MetadataStoreException {
+ CapturingMetadataStoreProvider.metadataURL = metadataURL;
+ CapturingMetadataStoreProvider.metadataStoreConfig = metadataStoreConfig;
+ return new CapturingMetadataStore();
+ }
+ }
+
+ private static class CapturingMetadataStore extends AbstractMetadataStore {
+ private CapturingMetadataStore() {
+ super("config-capture", OpenTelemetry.noop(), null, 1);
+ }
+
+ @Override
+ protected CompletableFuture<Boolean> existsFromStore(String path, Set<Option> opts) {
+ return CompletableFuture.completedFuture(false);
+ }
+
+ @Override
+ public CompletableFuture<List<String>> getChildrenFromStore(String path, Set<Option> opts) {
+ return CompletableFuture.completedFuture(List.of());
+ }
+
+ @Override
+ protected CompletableFuture<Optional<GetResult>> storeGet(String path, Set<Option> opts) {
+ return CompletableFuture.completedFuture(Optional.empty());
+ }
+
+ @Override
+ protected CompletableFuture<Void> storeDelete(String path, Optional<Long> expectedVersion, Set<Option> opts) {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ @Override
+ protected CompletableFuture<Stat> storePut(String path, byte[] data, Optional<Long> optExpectedVersion,
+ Set<Option> opts) {
+ return CompletableFuture.completedFuture(new Stat(path, 0, 0, 0, false, false, false));
+ }
+ }
+}