Allow to configure BookKeeper Opportunistic Striping Feature (and all BK client features using bookkeeper_ prefix)

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

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"

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<ZooKeeperCache> 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;
     }
 

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);
+
+    }
+
 }

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();
     }
 
 }

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