From 71af0e0c9e3a4122b6ac8c66740bacb8a398f38d Mon Sep 17 00:00:00 2001 From: Enrico Olivelli Date: Mon, 18 Jan 2021 08:44:28 +0100 Subject: [PATCH 1/6] Allow to configure BK Opportunistc Striping --- .../pulsar/broker/ServiceConfiguration.java | 8 ++ .../broker/BookKeeperClientFactoryImpl.java | 1 + .../BookKeeperClientFactoryImplTest.java | 14 +++ .../broker/service/BkEnsemblesTestBase.java | 15 +-- .../service/OpportunisticStripingTest.java | 96 +++++++++++++++++++ 5 files changed, 127 insertions(+), 7 deletions(-) create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OpportunisticStripingTest.java 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..fb3e4b2ee3e82 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,14 @@ public class ServiceConfiguration implements PulsarConfiguration { + " and resolving its metadata service location" ) private String bookkeeperMetadataServiceUri; + @FieldContext( + category = CATEGORY_STORAGE_BK, + doc = "With Opportustic Striping the bookeeper client will fallback to use less bookies in case " + + "of lack of a sufficient number of writable bookies for striping." + + "But you still have WriteQuorum bookies, in order to preserve durability configuration." + ) + private boolean bookkeeperOpportunisticStriping; + @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..645df8b01637e 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 @@ -111,6 +111,7 @@ ClientConfiguration createBkClientConfiguration(ServiceConfiguration conf) { bkConf.setNumChannelsPerBookie(conf.getBookkeeperNumberOfChannelsPerBookie()); bkConf.setUseV2WireProtocol(conf.isBookkeeperUseV2WireProtocol()); bkConf.setEnableDigestTypeAutodetection(true); + bkConf.setOpportunisticStriping(conf.isBookkeeperOpportunisticStriping()); bkConf.setStickyReadsEnabled(conf.isBookkeeperEnableStickyReads()); bkConf.setNettyMaxFrameSizeBytes(conf.getMaxMessageSize() + Commands.MESSAGE_SIZE_FRAME_PADDING); bkConf.setDiskWeightBasedPlacementEnabled(conf.isBookkeeperDiskWeightBasedPlacementEnabled()); 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..08c9523cfb2b4 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,18 @@ public void testSetMetadataServiceUri() { } } + @Test + public void testOpportunisticStripingConfiguration() { + BookKeeperClientFactoryImpl factory = new BookKeeperClientFactoryImpl(); + ServiceConfiguration conf = new ServiceConfiguration(); + // default value + assertEquals(factory.createBkClientConfiguration(conf).getOpportunisticStriping(), false); + conf.setBookkeeperOpportunisticStriping(true); + assertEquals(factory.createBkClientConfiguration(conf).getOpportunisticStriping(), true); + + conf.setBookkeeperOpportunisticStriping(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..9cb19e3a11f23 --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OpportunisticStripingTest.java @@ -0,0 +1,96 @@ +/** + * 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.Consumer; +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.setBookkeeperOpportunisticStriping(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); + } + } + } + } + +} From 28960266f473fe4773fa5fbbd4e76b3e34a6893f Mon Sep 17 00:00:00 2001 From: Enrico Olivelli Date: Mon, 18 Jan 2021 17:24:29 +0100 Subject: [PATCH 2/6] example configuration --- conf/broker.conf | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/conf/broker.conf b/conf/broker.conf index 42cad7bcf9ca6..ee9f6c011e8f7 100644 --- a/conf/broker.conf +++ b/conf/broker.conf @@ -800,6 +800,10 @@ 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 +bookkeeperOpportunisticStriping=false + # How frequently to flush the cursor positions that were accumulated due to rate limiting. (seconds). # Default is 60 seconds managedLedgerCursorPositionFlushSeconds = 60 From 0c9febb6d9428ddf9c5ebbbd38f27d93e0c07c41 Mon Sep 17 00:00:00 2001 From: Enrico Olivelli Date: Tue, 19 Jan 2021 14:33:29 +0100 Subject: [PATCH 3/6] Fix typo --- .../java/org/apache/pulsar/broker/ServiceConfiguration.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 fb3e4b2ee3e82..52ca2ae326fce 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 @@ -1072,7 +1072,7 @@ public class ServiceConfiguration implements PulsarConfiguration { private String bookkeeperMetadataServiceUri; @FieldContext( category = CATEGORY_STORAGE_BK, - doc = "With Opportustic Striping the bookeeper client will fallback to use less bookies in case " + doc = "With Opportustic Striping the bookeeper client will fall back to use fewer bookies in case " + "of lack of a sufficient number of writable bookies for striping." + "But you still have WriteQuorum bookies, in order to preserve durability configuration." ) From 47575e535d9551abc0cb7a8bcd590a5394a6cb74 Mon Sep 17 00:00:00 2001 From: Enrico Olivelli Date: Wed, 20 Jan 2021 13:54:50 +0100 Subject: [PATCH 4/6] Introduce a more general way to configure BK client --- conf/broker.conf | 5 ++++- .../apache/pulsar/broker/ServiceConfiguration.java | 7 ------- .../pulsar/broker/BookKeeperClientFactoryImpl.java | 14 ++++++++++++-- .../broker/BookKeeperClientFactoryImplTest.java | 5 ++--- .../broker/service/OpportunisticStripingTest.java | 2 +- 5 files changed, 19 insertions(+), 14 deletions(-) diff --git a/conf/broker.conf b/conf/broker.conf index ee9f6c011e8f7..9faf152e0f7a6 100644 --- a/conf/broker.conf +++ b/conf/broker.conf @@ -802,7 +802,10 @@ managedLedgerDefaultAckQuorum=2 # with OpportunisticStriping=true the ensembleSize is adapted automatically to writeQuorum # in case of lack of enough bookies -bookkeeperOpportunisticStriping=false +#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 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 52ca2ae326fce..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,13 +1070,6 @@ public class ServiceConfiguration implements PulsarConfiguration { + " and resolving its metadata service location" ) private String bookkeeperMetadataServiceUri; - @FieldContext( - category = CATEGORY_STORAGE_BK, - doc = "With Opportustic Striping the bookeeper client will fall back to use fewer bookies in case " - + "of lack of a sufficient number of writable bookies for striping." - + "But you still have WriteQuorum bookies, in order to preserve durability configuration." - ) - private boolean bookkeeperOpportunisticStriping; @FieldContext( category = CATEGORY_STORAGE_BK, 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 645df8b01637e..b21268a9acfaa 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<>(); @@ -111,7 +114,6 @@ ClientConfiguration createBkClientConfiguration(ServiceConfiguration conf) { bkConf.setNumChannelsPerBookie(conf.getBookkeeperNumberOfChannelsPerBookie()); bkConf.setUseV2WireProtocol(conf.isBookkeeperUseV2WireProtocol()); bkConf.setEnableDigestTypeAutodetection(true); - bkConf.setOpportunisticStriping(conf.isBookkeeperOpportunisticStriping()); bkConf.setStickyReadsEnabled(conf.isBookkeeperEnableStickyReads()); bkConf.setNettyMaxFrameSizeBytes(conf.getMaxMessageSize() + Commands.MESSAGE_SIZE_FRAME_PADDING); bkConf.setDiskWeightBasedPlacementEnabled(conf.isBookkeeperDiskWeightBasedPlacementEnabled()); @@ -138,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 08c9523cfb2b4..2fce7f1c907cb 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 @@ -211,10 +211,9 @@ public void testOpportunisticStripingConfiguration() { ServiceConfiguration conf = new ServiceConfiguration(); // default value assertEquals(factory.createBkClientConfiguration(conf).getOpportunisticStriping(), false); - conf.setBookkeeperOpportunisticStriping(true); + conf.getProperties().setProperty("bookkeeper.opportunisticStriping", "true"); assertEquals(factory.createBkClientConfiguration(conf).getOpportunisticStriping(), true); - - conf.setBookkeeperOpportunisticStriping(false); + 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/OpportunisticStripingTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OpportunisticStripingTest.java index 9cb19e3a11f23..34d7d12273a5a 100644 --- 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 @@ -51,7 +51,7 @@ protected void configurePulsar(ServiceConfiguration config) throws Exception { config.setManagedLedgerDefaultAckQuorum(2); config.setBrokerDeleteInactiveTopicsEnabled(false); - config.setBookkeeperOpportunisticStriping(true); + config.getProperties().setProperty("bookkeeper.opportunisticStriping", "true"); } @Test From 451f8da051571ad277b1e80b7e1bfc1874ab3902 Mon Sep 17 00:00:00 2001 From: Enrico Olivelli Date: Thu, 21 Jan 2021 14:47:38 +0100 Subject: [PATCH 5/6] Use underscore --- .../org/apache/pulsar/broker/BookKeeperClientFactoryImpl.java | 2 +- .../apache/pulsar/broker/BookKeeperClientFactoryImplTest.java | 4 ++-- .../pulsar/broker/service/OpportunisticStripingTest.java | 3 +-- 3 files changed, 4 insertions(+), 5 deletions(-) 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 b21268a9acfaa..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 @@ -143,7 +143,7 @@ ClientConfiguration createBkClientConfiguration(ServiceConfiguration conf) { Properties allProps = conf.getProperties(); allProps.forEach((key, value) -> { String sKey = key.toString(); - if (sKey.startsWith("bookkeeper.") && value != null) { + 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); 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 2fce7f1c907cb..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 @@ -211,9 +211,9 @@ public void testOpportunisticStripingConfiguration() { ServiceConfiguration conf = new ServiceConfiguration(); // default value assertEquals(factory.createBkClientConfiguration(conf).getOpportunisticStriping(), false); - conf.getProperties().setProperty("bookkeeper.opportunisticStriping", "true"); + conf.getProperties().setProperty("bookkeeper_opportunisticStriping", "true"); assertEquals(factory.createBkClientConfiguration(conf).getOpportunisticStriping(), true); - conf.getProperties().setProperty("bookkeeper.opportunisticStriping", "false"); + 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/OpportunisticStripingTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OpportunisticStripingTest.java index 34d7d12273a5a..221367cb1bc45 100644 --- 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 @@ -24,7 +24,6 @@ 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.Consumer; import org.apache.pulsar.client.api.Producer; import org.apache.pulsar.client.api.PulsarClient; import static org.testng.AssertJUnit.assertEquals; @@ -51,7 +50,7 @@ protected void configurePulsar(ServiceConfiguration config) throws Exception { config.setManagedLedgerDefaultAckQuorum(2); config.setBrokerDeleteInactiveTopicsEnabled(false); - config.getProperties().setProperty("bookkeeper.opportunisticStriping", "true"); + config.getProperties().setProperty("bookkeeper_opportunisticStriping", "true"); } @Test From 75f649a18611e54c7ef37ec4eb2ab98ff1455207 Mon Sep 17 00:00:00 2001 From: Enrico Olivelli Date: Thu, 21 Jan 2021 14:48:50 +0100 Subject: [PATCH 6/6] Fix default config file --- conf/broker.conf | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/conf/broker.conf b/conf/broker.conf index 9faf152e0f7a6..52b63d99b309c 100644 --- a/conf/broker.conf +++ b/conf/broker.conf @@ -802,10 +802,10 @@ managedLedgerDefaultAckQuorum=2 # with OpportunisticStriping=true the ensembleSize is adapted automatically to writeQuorum # in case of lack of enough bookies -#bookkeeper.opportunisticStriping=false +#bookkeeper_opportunisticStriping=false # you can add other configuration options for the BookKeeper client -# by prefixing them with bookkeeper. +# by prefixing them with bookkeeper_ # How frequently to flush the cursor positions that were accumulated due to rate limiting. (seconds). # Default is 60 seconds