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 <[email protected]> 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)); + } + } +}
