diff --git a/conf/broker.conf b/conf/broker.conf index 42cad7bcf9ca6..52b63d99b309c 100644 --- a/conf/broker.conf +++ b/conf/broker.conf @@ -800,6 +800,13 @@ managedLedgerDefaultWriteQuorum=2 # Number of guaranteed copies (acks to wait before write is complete) managedLedgerDefaultAckQuorum=2 +# with OpportunisticStriping=true the ensembleSize is adapted automatically to writeQuorum +# in case of lack of enough bookies +#bookkeeper_opportunisticStriping=false + +# you can add other configuration options for the BookKeeper client +# by prefixing them with bookkeeper_ + # How frequently to flush the cursor positions that were accumulated due to rate limiting. (seconds). # Default is 60 seconds managedLedgerCursorPositionFlushSeconds = 60 diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java index ea457abf9bf92..0d382791cbb82 100644 --- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java +++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java @@ -1070,6 +1070,7 @@ public class ServiceConfiguration implements PulsarConfiguration { + " and resolving its metadata service location" ) private String bookkeeperMetadataServiceUri; + @FieldContext( category = CATEGORY_STORAGE_BK, doc = "Authentication plugin to use when connecting to bookies" diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/BookKeeperClientFactoryImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/BookKeeperClientFactoryImpl.java index be05270ebda4e..e6ac87fb30fdd 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/BookKeeperClientFactoryImpl.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/BookKeeperClientFactoryImpl.java @@ -27,8 +27,10 @@ import java.io.IOException; import java.util.Map; import java.util.Optional; +import java.util.Properties; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; +import lombok.extern.slf4j.Slf4j; import org.apache.bookkeeper.client.BKException; import org.apache.bookkeeper.client.BookKeeper; import org.apache.bookkeeper.client.EnsemblePlacementPolicy; @@ -46,6 +48,7 @@ import org.apache.zookeeper.ZooKeeper; @SuppressWarnings("deprecation") +@Slf4j public class BookKeeperClientFactoryImpl implements BookKeeperClientFactory { private final AtomicReference rackawarePolicyZkCache = new AtomicReference<>(); @@ -137,7 +140,15 @@ ClientConfiguration createBkClientConfiguration(ServiceConfiguration conf) { conf.getBookkeeperClientGetBookieInfoIntervalSeconds(), TimeUnit.SECONDS); bkConf.setGetBookieInfoRetryIntervalSeconds( conf.getBookkeeperClientGetBookieInfoRetryIntervalSeconds(), TimeUnit.SECONDS); - + Properties allProps = conf.getProperties(); + allProps.forEach((key, value) -> { + String sKey = key.toString(); + if (sKey.startsWith("bookkeeper_") && value != null) { + String bkExtraConfigKey = sKey.substring(11); + log.info("Extra BookKeeper client configuration {}, setting {}={}", sKey, bkExtraConfigKey, value); + bkConf.setProperty(bkExtraConfigKey, value); + } + }); return bkConf; } 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 0f1707175249c..13c352cd6a65d 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 @@ -205,4 +205,17 @@ public void testSetMetadataServiceUri() { } } + @Test + public void testOpportunisticStripingConfiguration() { + BookKeeperClientFactoryImpl factory = new BookKeeperClientFactoryImpl(); + ServiceConfiguration conf = new ServiceConfiguration(); + // default value + assertEquals(factory.createBkClientConfiguration(conf).getOpportunisticStriping(), false); + conf.getProperties().setProperty("bookkeeper_opportunisticStriping", "true"); + assertEquals(factory.createBkClientConfiguration(conf).getOpportunisticStriping(), true); + conf.getProperties().setProperty("bookkeeper_opportunisticStriping", "false"); + assertEquals(factory.createBkClientConfiguration(conf).getOpportunisticStriping(), false); + + } + } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BkEnsemblesTestBase.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BkEnsemblesTestBase.java index ec4e24d7a24d5..a44e6d6468f86 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BkEnsemblesTestBase.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BkEnsemblesTestBase.java @@ -57,6 +57,10 @@ public BkEnsemblesTestBase(int numberOfBookies) { this.numberOfBookies = numberOfBookies; } + protected void configurePulsar(ServiceConfiguration config) throws Exception { + //overridable by subclasses + } + @BeforeMethod protected void setup() throws Exception { try { @@ -78,6 +82,7 @@ protected void setup() throws Exception { config.setAdvertisedAddress("127.0.0.1"); config.setAllowAutoTopicCreationType("non-partitioned"); config.setZooKeeperOperationTimeoutSeconds(1); + configurePulsar(config); pulsar = new PulsarService(config); pulsar.start(); @@ -95,13 +100,9 @@ protected void setup() throws Exception { @AfterMethod(alwaysRun = true) protected void shutdown() throws Exception { - try { - admin.close(); - pulsar.close(); - bkEnsemble.stop(); - } catch (Throwable t) { - log.warn("Error cleaning up broker test setup state", t); - } + admin.close(); + pulsar.close(); + bkEnsemble.stop(); } } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OpportunisticStripingTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OpportunisticStripingTest.java new file mode 100644 index 0000000000000..221367cb1bc45 --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OpportunisticStripingTest.java @@ -0,0 +1,95 @@ +/** + * 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.broker.service; + +import java.util.concurrent.TimeUnit; +import org.apache.bookkeeper.client.api.BookKeeper; +import org.apache.bookkeeper.client.api.LedgerMetadata; +import org.apache.bookkeeper.client.api.ListLedgersResult; +import org.apache.bookkeeper.conf.ClientConfiguration; +import org.apache.pulsar.broker.ServiceConfiguration; +import org.apache.pulsar.client.api.Producer; +import org.apache.pulsar.client.api.PulsarClient; +import static org.testng.AssertJUnit.assertEquals; +import static org.testng.AssertJUnit.assertTrue; +import org.testng.annotations.Test; + +/** + * With BookKeeper Opportunistic Striping feature we can allow Pulsar to work + * with only WQ bookie during temporary outages of some bookie. + */ +public class OpportunisticStripingTest extends BkEnsemblesTestBase { + + public OpportunisticStripingTest() { + // starting only two bookies + super(2); + } + + @Override + protected void configurePulsar(ServiceConfiguration config) throws Exception { + // we would like to stripe over 5 bookies + config.setManagedLedgerDefaultEnsembleSize(5); + // we want 2 copies for each entry + config.setManagedLedgerDefaultWriteQuorum(2); + config.setManagedLedgerDefaultAckQuorum(2); + + config.setBrokerDeleteInactiveTopicsEnabled(false); + config.getProperties().setProperty("bookkeeper_opportunisticStriping", "true"); + } + + @Test + public void testOpportunisticStriping() throws Exception { + + try (PulsarClient client = PulsarClient.builder() + .serviceUrl(pulsar.getWebServiceAddress()) + .statsInterval(0, TimeUnit.SECONDS) + .build();) { + + final String ns1 = "prop/usc/opportunistic1"; + admin.namespaces().createNamespace(ns1); + + final String topic1 = "persistent://" + ns1 + "/my-topic"; + Producer producer = client.newProducer().topic(topic1).create(); + for (int i = 0; i < 10; i++) { + String message = "my-message-" + i; + producer.send(message.getBytes()); + } + + // verify that all ledgers has the proper writequorumsize, + // equals to the number of available bookies (in this case 2) + ClientConfiguration clientConfiguration = new ClientConfiguration(); + clientConfiguration.setZkServers("localhost:" + this.bkEnsemble.getZookeeperPort()); + + try (BookKeeper bkAdmin = BookKeeper.newBuilder(clientConfiguration).build()) { + try (ListLedgersResult list = bkAdmin.newListLedgersOp().execute().get();) { + int count = 0; + for (long ledgerId : list.toIterable()) { + LedgerMetadata ledgerMetadata = bkAdmin.getLedgerMetadata(ledgerId).get(); + assertEquals(2, ledgerMetadata.getEnsembleSize()); + assertEquals(2, ledgerMetadata.getWriteQuorumSize()); + assertEquals(2, ledgerMetadata.getAckQuorumSize()); + count++; + } + assertTrue(count > 0); + } + } + } + } + +}