This is an automated email from the ASF dual-hosted git repository.
merlimat pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new 474474937d5 [improve][meta] Support tuning Oxia MetadataStoreConfig
through metadata-store URIs (#26150)
474474937d5 is described below
commit 474474937d5a2f5771e23743251115f25cfeec1a
Author: Qiang Zhao <[email protected]>
AuthorDate: Tue Jul 7 00:31:48 2026 +0800
[improve][meta] Support tuning Oxia MetadataStoreConfig through
metadata-store URIs (#26150)
Co-authored-by: void-ptr974 <[email protected]>
---
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..e7832568a14
--- /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.EnumSet;
+import java.util.List;
+import java.util.Optional;
+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.Stat;
+import org.apache.pulsar.metadata.api.extended.CreateOption;
+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) {
+ return CompletableFuture.completedFuture(false);
+ }
+
+ @Override
+ public CompletableFuture<List<String>> getChildrenFromStore(String
path) {
+ return CompletableFuture.completedFuture(List.of());
+ }
+
+ @Override
+ protected CompletableFuture<Optional<GetResult>> storeGet(String path)
{
+ return CompletableFuture.completedFuture(Optional.empty());
+ }
+
+ @Override
+ protected CompletableFuture<Void> storeDelete(String path,
Optional<Long> expectedVersion) {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ @Override
+ protected CompletableFuture<Stat> storePut(String path, byte[] data,
Optional<Long> optExpectedVersion,
+ EnumSet<CreateOption> opts)
{
+ return CompletableFuture.completedFuture(new Stat(path, 0, 0, 0,
false, false, false));
+ }
+ }
+}
