Add bookkeeperClientMinNumRacksPerWriteQuorum,...

Add bookkeeperClientMinNumRacksPerWriteQuorum, bookkeeperClientEnforceMinNumRacksPerWriteQuorum conf options (#7977)

Added bookkeeperClientMinNumRacksPerWriteQuorum and bookkeeperClientEnforceMinNumRacksPerWriteQuorum to be able to configure rack and region aware bookie selection policy more precisely.

Motivation
We have Apache Pulsar setup running on Azure in 3 different Availability Zones (2 bookkeepers per zone). We wanted to store copy of each pulsar message in all that AZs. After configuring rack aware bookie selection policy, we found that messages are usually stored only in 2 AZs. After digging, we've found that it's caused by default bookeeper client minNumRacksPerWriteQuorum=2 setting. We've added bookkeeperClientMinNumRacksPerWriteQuorum and also bookkeeperClientEnforceMinNumRacksPerWriteQuorum to be able to configure this behaviour.

Modifications
Added bookkeeperClientMinNumRacksPerWriteQuorum and bookkeeperClientEnforceMinNumRacksPerWriteQuorum configuration option, modified tests, config files and documentation.

Verifying this change
We've updated existing unit tests, deployed modified pulsar docker image on AKS kubernetes cluster using helm charts and checked that stored messages were written always to bookkeepers in all AZs.


* Added `bookkeeperClientMinNumRacksPerWriteQuorum` and `bookkeeperClientEnforceMinNumRacksPerWriteQuorum` broker configuration options to be able set `minNumRacksPerWriteQuorum` and `enforceMinNumRacksPerWriteQuorum` bookkeeper client options.

* Reverted accidental change in ServiceConfigurationTest.
Co-authored-by: NMichal Koziorowski <Michal.Koziorowski@tomtom.com>

conf/broker.conf

@@ -649,6 +649,15 @@ bookkeeperClientRackawarePolicyEnabled=true
 # If enabled, the value of bookkeeperClientRackawarePolicyEnabled is ignored
 bookkeeperClientRegionawarePolicyEnabled=false
 
+# Minimum number of racks per write quorum. BK rack-aware bookie selection policy will try to
+# get bookies from at least 'bookkeeperClientMinNumRacksPerWriteQuorum' racks for a write quorum.
+bookkeeperClientMinNumRacksPerWriteQuorum=2
+
+# Enforces rack-aware bookie selection policy to pick bookies from 'bookkeeperClientMinNumRacksPerWriteQuorum'
+# racks for a writeQuorum.
+# If BK can't find bookie then it would throw BKNotEnoughBookiesException instead of picking random one.
+bookkeeperClientEnforceMinNumRacksPerWriteQuorum=false
+
 # Enable/disable reordering read sequence on reading entries.
 bookkeeperClientReorderReadSequenceEnabled=false
 

conf/standalone.conf

@@ -438,6 +438,15 @@ bookkeeperClientRackawarePolicyEnabled=true
 # If enabled, the value of bookkeeperClientRackawarePolicyEnabled is ignored
 bookkeeperClientRegionawarePolicyEnabled=false
 
+# Minimum number of racks per write quorum. BK rack-aware bookie selection policy will try to
+# get bookies from at least 'bookkeeperClientMinNumRacksPerWriteQuorum' racks for a write quorum.
+bookkeeperClientMinNumRacksPerWriteQuorum=1
+
+# Enforces rack-aware bookie selection policy to pick bookies from 'bookkeeperClientMinNumRacksPerWriteQuorum'
+# racks for a writeQuorum.
+# If BK can't find bookie then it would throw BKNotEnoughBookiesException instead of picking random one.
+bookkeeperClientEnforceMinNumRacksPerWriteQuorum=false
+
 # Enable/disable reordering read sequence on reading entries.
 bookkeeperClientReorderReadSequenceEnabled=false
 

deployment/terraform-ansible/templates/broker.conf

@@ -634,6 +634,15 @@ bookkeeperClientRackawarePolicyEnabled=true
 # If enabled, the value of bookkeeperClientRackawarePolicyEnabled is ignored
 bookkeeperClientRegionawarePolicyEnabled=false
 
+# Minimum number of racks per write quorum. BK rack-aware bookie selection policy will try to
+# get bookies from at least 'bookkeeperClientMinNumRacksPerWriteQuorum' racks for a write quorum.
+bookkeeperClientMinNumRacksPerWriteQuorum=2
+
+# Enforces rack-aware bookie selection policy to pick bookies from 'bookkeeperClientMinNumRacksPerWriteQuorum'
+# racks for a writeQuorum.
+# If BK can't find bookie then it would throw BKNotEnoughBookiesException instead of picking random one.
+bookkeeperClientEnforceMinNumRacksPerWriteQuorum=false
+
 # Enable/disable reordering read sequence on reading entries.
 bookkeeperClientReorderReadSequenceEnabled=false
 

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java

@@ -1001,6 +1001,16 @@ public class ServiceConfiguration implements PulsarConfiguration {
         doc = "Enable region-aware bookie selection policy. \n\nBK will chose bookies from"
             + " different regions and racks when forming a new bookie ensemble")
     private boolean bookkeeperClientRegionawarePolicyEnabled = false;
+    @FieldContext(
+        category = CATEGORY_STORAGE_BK,
+        doc = "Minimum number of racks per write quorum. \n\nBK rack-aware bookie selection policy will try to"
+            + " get bookies from at least 'bookkeeperClientMinNumRacksPerWriteQuorum' racks for a write quorum.")
+    private int bookkeeperClientMinNumRacksPerWriteQuorum = 2;
+    @FieldContext(
+        category = CATEGORY_STORAGE_BK,
+        doc = "Enforces rack-aware bookie selection policy to pick bookies from 'bookkeeperClientMinNumRacksPerWriteQuorum' racks for "
+            + "a writeQuorum. \n\nIf BK can't find bookie then it would throw BKNotEnoughBookiesException instead of picking random one.")
+    private boolean bookkeeperClientEnforceMinNumRacksPerWriteQuorum = false;
     @FieldContext(
         category = CATEGORY_STORAGE_BK,
         doc = "Enable/disable reordering read sequence on reading entries")

pulsar-broker/src/main/java/org/apache/pulsar/broker/BookKeeperClientFactoryImpl.java

@@ -170,6 +170,10 @@ public class BookKeeperClientFactoryImpl implements BookKeeperClientFactory {
             } else {
                 bkConf.setEnsemblePlacementPolicy(RackawareEnsemblePlacementPolicy.class);
             }
+
+            bkConf.setMinNumRacksPerWriteQuorum(conf.getBookkeeperClientMinNumRacksPerWriteQuorum());
+            bkConf.setEnforceMinNumRacksPerWriteQuorum(conf.isBookkeeperClientEnforceMinNumRacksPerWriteQuorum());
+
             bkConf.setProperty(REPP_DNS_RESOLVER_CLASS,
                 conf.getProperties().getProperty(
                     REPP_DNS_RESOLVER_CLASS,

pulsar-broker/src/test/java/org/apache/pulsar/PulsarBrokerStarterTest.java

@@ -61,6 +61,8 @@ public class PulsarBrokerStarterTest {
         printWriter.println("bookkeeperClientHealthCheckErrorThresholdPerInterval=5");
         printWriter.println("bookkeeperClientRackawarePolicyEnabled=true");
         printWriter.println("bookkeeperClientRegionawarePolicyEnabled=false");
+        printWriter.println("bookkeeperClientMinNumRacksPerWriteQuorum=5");
+        printWriter.println("bookkeeperClientEnforceMinNumRacksPerWriteQuorum=true");
         printWriter.println("bookkeeperClientReorderReadSequenceEnabled=false");
         printWriter.println("bookkeeperClientIsolationGroups=group1,group2");
         printWriter.println("backlogQuotaDefaultLimitGB=18");
@@ -123,6 +125,8 @@ public class PulsarBrokerStarterTest {
         assertTrue(serviceConfig.isBookkeeperClientHealthCheckEnabled());
         assertEquals(serviceConfig.getBookkeeperClientHealthCheckErrorThresholdPerInterval(), 5);
         assertTrue(serviceConfig.isBookkeeperClientRackawarePolicyEnabled());
+        assertEquals(serviceConfig.getBookkeeperClientMinNumRacksPerWriteQuorum(), 5);
+        assertTrue(serviceConfig.isBookkeeperClientEnforceMinNumRacksPerWriteQuorum());
         assertFalse(serviceConfig.isBookkeeperClientRegionawarePolicyEnabled());
         assertFalse(serviceConfig.isBookkeeperClientReorderReadSequenceEnabled());
         assertEquals(serviceConfig.getBookkeeperClientIsolationGroups(), "group1,group2");

pulsar-broker/src/test/java/org/apache/pulsar/broker/BookKeeperClientFactoryImplTest.java

@@ -23,6 +23,8 @@ import static org.apache.bookkeeper.client.RegionAwareEnsemblePlacementPolicy.RE
 import static org.apache.bookkeeper.client.RegionAwareEnsemblePlacementPolicy.REPP_ENABLE_VALIDATION;
 import static org.apache.bookkeeper.client.RegionAwareEnsemblePlacementPolicy.REPP_MINIMUM_REGIONS_FOR_DURABILITY;
 import static org.apache.bookkeeper.client.RegionAwareEnsemblePlacementPolicy.REPP_REGIONS_TO_WRITE;
+import static org.apache.bookkeeper.conf.AbstractConfiguration.MIN_NUM_RACKS_PER_WRITE_QUORUM;
+import static org.apache.bookkeeper.conf.AbstractConfiguration.ENFORCE_MIN_NUM_RACKS_PER_WRITE_QUORUM;
 import static org.mockito.Mockito.mock;
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertFalse;
@@ -57,6 +59,8 @@ public class BookKeeperClientFactoryImplTest {
         assertNull(bkConf.getProperty(REPP_MINIMUM_REGIONS_FOR_DURABILITY));
         assertNull(bkConf.getProperty(REPP_ENABLE_DURABILITY_ENFORCEMENT_IN_REPLACE));
         assertNull(bkConf.getProperty(REPP_DNS_RESOLVER_CLASS));
+        assertNull(bkConf.getProperty(MIN_NUM_RACKS_PER_WRITE_QUORUM));
+        assertNull(bkConf.getProperty(ENFORCE_MIN_NUM_RACKS_PER_WRITE_QUORUM));
 
         BookKeeperClientFactoryImpl.setDefaultEnsemblePlacementPolicy(
             rackawarePolicyZkCache,
@@ -73,6 +77,8 @@ public class BookKeeperClientFactoryImplTest {
         assertEquals(
             bkConf.getProperty(REPP_DNS_RESOLVER_CLASS),
             ZkBookieRackAffinityMapping.class.getName());
+        assertFalse(bkConf.getEnforceMinNumRacksPerWriteQuorum());
+        assertEquals(2, bkConf.getMinNumRacksPerWriteQuorum());
 
         ((ZooKeeperCache) bkConf.getProperty(ZooKeeperCache.ZK_CACHE_INSTANCE)).stop();
     }
@@ -90,6 +96,8 @@ public class BookKeeperClientFactoryImplTest {
         assertNull(bkConf.getProperty(REPP_MINIMUM_REGIONS_FOR_DURABILITY));
         assertNull(bkConf.getProperty(REPP_ENABLE_DURABILITY_ENFORCEMENT_IN_REPLACE));
         assertNull(bkConf.getProperty(REPP_DNS_RESOLVER_CLASS));
+        assertNull(bkConf.getProperty(MIN_NUM_RACKS_PER_WRITE_QUORUM));
+        assertNull(bkConf.getProperty(ENFORCE_MIN_NUM_RACKS_PER_WRITE_QUORUM));
 
         conf.setBookkeeperClientRegionawarePolicyEnabled(true);
 
@@ -108,6 +116,8 @@ public class BookKeeperClientFactoryImplTest {
         assertEquals(
             bkConf.getProperty(REPP_DNS_RESOLVER_CLASS),
             ZkBookieRackAffinityMapping.class.getName());
+        assertFalse(bkConf.getEnforceMinNumRacksPerWriteQuorum());
+        assertEquals(2, bkConf.getMinNumRacksPerWriteQuorum());
 
         ((ZooKeeperCache) bkConf.getProperty(ZooKeeperCache.ZK_CACHE_INSTANCE)).stop();
     }
@@ -125,6 +135,8 @@ public class BookKeeperClientFactoryImplTest {
         assertNull(bkConf.getProperty(REPP_MINIMUM_REGIONS_FOR_DURABILITY));
         assertNull(bkConf.getProperty(REPP_ENABLE_DURABILITY_ENFORCEMENT_IN_REPLACE));
         assertNull(bkConf.getProperty(REPP_DNS_RESOLVER_CLASS));
+        assertNull(bkConf.getProperty(MIN_NUM_RACKS_PER_WRITE_QUORUM));
+        assertNull(bkConf.getProperty(ENFORCE_MIN_NUM_RACKS_PER_WRITE_QUORUM));
 
         conf.setBookkeeperClientRegionawarePolicyEnabled(true);
         conf.getProperties().setProperty(REPP_ENABLE_VALIDATION, "false");
@@ -132,6 +144,8 @@ public class BookKeeperClientFactoryImplTest {
         conf.getProperties().setProperty(REPP_MINIMUM_REGIONS_FOR_DURABILITY, "4");
         conf.getProperties().setProperty(REPP_ENABLE_DURABILITY_ENFORCEMENT_IN_REPLACE, "false");
         conf.getProperties().setProperty(REPP_DNS_RESOLVER_CLASS, CachedDNSToSwitchMapping.class.getName());
+        conf.setBookkeeperClientMinNumRacksPerWriteQuorum(20);
+        conf.setBookkeeperClientEnforceMinNumRacksPerWriteQuorum(true);
 
         BookKeeperClientFactoryImpl.setDefaultEnsemblePlacementPolicy(
             rackawarePolicyZkCache,
@@ -148,6 +162,8 @@ public class BookKeeperClientFactoryImplTest {
         assertEquals(
             bkConf.getProperty(REPP_DNS_RESOLVER_CLASS),
             CachedDNSToSwitchMapping.class.getName());
+        assertTrue(bkConf.getEnforceMinNumRacksPerWriteQuorum());
+        assertEquals(20, bkConf.getMinNumRacksPerWriteQuorum());
 
         ((ZooKeeperCache) bkConf.getProperty(ZooKeeperCache.ZK_CACHE_INSTANCE)).stop();
     }

pulsar-broker/src/test/resources/configurations/pulsar_broker_test.conf

@@ -60,6 +60,8 @@ bookkeeperClientHealthCheckErrorThresholdPerInterval=5
 bookkeeperClientHealthCheckQuarantineTimeInSeconds=1800
 bookkeeperClientRackawarePolicyEnabled=true
 bookkeeperClientRegionawarePolicyEnabled=false
+bookkeeperClientMinNumRacksPerWriteQuorum=2
+bookkeeperClientEnforceMinNumRacksPerWriteQuorum=false
 bookkeeperClientReorderReadSequenceEnabled=false
 bookkeeperClientIsolationGroups="test_group"
 managedLedgerDefaultEnsembleSize=3

pulsar-client-cpp/test-conf/standalone-ssl.conf

@@ -146,6 +146,15 @@ bookkeeperClientRackawarePolicyEnabled=true
 # If enabled, the value of bookkeeperClientRackawarePolicyEnabled is ignored
 bookkeeperClientRegionawarePolicyEnabled=false
 
+# Minimum number of racks per write quorum. BK rack-aware bookie selection policy will try to
+# get bookies from at least 'bookkeeperClientMinNumRacksPerWriteQuorum' racks for a write quorum.
+bookkeeperClientMinNumRacksPerWriteQuorum=1
+
+# Enforces rack-aware bookie selection policy to pick bookies from 'bookkeeperClientMinNumRacksPerWriteQuorum'
+# racks for a writeQuorum.
+# If BK can't find bookie then it would throw BKNotEnoughBookiesException instead of picking random one.
+bookkeeperClientEnforceMinNumRacksPerWriteQuorum=false
+
 # Enable/disable reordering read sequence on reading entries.
 bookkeeperClientReorderReadSequenceEnabled=false
 

pulsar-client-cpp/test-conf/standalone.conf

@@ -134,6 +134,15 @@ bookkeeperClientRackawarePolicyEnabled=true
 # If enabled, the value of bookkeeperClientRackawarePolicyEnabled is ignored
 bookkeeperClientRegionawarePolicyEnabled=false
 
+# Minimum number of racks per write quorum. BK rack-aware bookie selection policy will try to
+# get bookies from at least 'bookkeeperClientMinNumRacksPerWriteQuorum' racks for a write quorum.
+bookkeeperClientMinNumRacksPerWriteQuorum=1
+
+# Enforces rack-aware bookie selection policy to pick bookies from 'bookkeeperClientMinNumRacksPerWriteQuorum'
+# racks for a writeQuorum.
+# If BK can't find bookie then it would throw BKNotEnoughBookiesException instead of picking random one.
+bookkeeperClientEnforceMinNumRacksPerWriteQuorum=false
+
 # Enable/disable reordering read sequence on reading entries.
 bookkeeperClientReorderReadSequenceEnabled=false
 

pulsar-client-cpp/tests/authentication.conf

@@ -140,6 +140,15 @@ bookkeeperClientRackawarePolicyEnabled=true
 # If enabled, the value of bookkeeperClientRackawarePolicyEnabled is ignored
 bookkeeperClientRegionawarePolicyEnabled=false
 
+# Minimum number of racks per write quorum. BK rack-aware bookie selection policy will try to
+# get bookies from at least 'bookkeeperClientMinNumRacksPerWriteQuorum' racks for a write quorum.
+bookkeeperClientMinNumRacksPerWriteQuorum=1
+
+# Enforces rack-aware bookie selection policy to pick bookies from 'bookkeeperClientMinNumRacksPerWriteQuorum'
+# racks for a writeQuorum.
+# If BK can't find bookie then it would throw BKNotEnoughBookiesException instead of picking random one.
+bookkeeperClientEnforceMinNumRacksPerWriteQuorum=false
+
 # Enable/disable reordering read sequence on reading entries.
 bookkeeperClientReorderReadSequenceEnabled=false
 

pulsar-client-cpp/tests/standalone.conf

@@ -132,6 +132,15 @@ bookkeeperClientRackawarePolicyEnabled=true
 # If enabled, the value of bookkeeperClientRackawarePolicyEnabled is ignored
 bookkeeperClientRegionawarePolicyEnabled=false
 
+# Minimum number of racks per write quorum. BK rack-aware bookie selection policy will try to
+# get bookies from at least 'bookkeeperClientMinNumRacksPerWriteQuorum' racks for a write quorum.
+bookkeeperClientMinNumRacksPerWriteQuorum=1
+
+# Enforces rack-aware bookie selection policy to pick bookies from 'bookkeeperClientMinNumRacksPerWriteQuorum'
+# racks for a writeQuorum.
+# If BK can't find bookie then it would throw BKNotEnoughBookiesException instead of picking random one.
+bookkeeperClientEnforceMinNumRacksPerWriteQuorum=false
+
 # Enable/disable reordering read sequence on reading entries.
 bookkeeperClientReorderReadSequenceEnabled=false
 

site2/docs/reference-configuration.md

@@ -253,6 +253,8 @@ subscriptionExpirationTimeMinutes | How long to delete inactive subscriptions fr
 |bookkeeperClientHealthCheckQuarantineTimeInSeconds ||1800|
 |bookkeeperClientRackawarePolicyEnabled|  Enable rack-aware bookie selection policy. BK will chose bookies from different racks when forming a new bookie ensemble  |true|
 |bookkeeperClientRegionawarePolicyEnabled|  Enable region-aware bookie selection policy. BK will chose bookies from different regions and racks when forming a new bookie ensemble. If enabled, the value of bookkeeperClientRackawarePolicyEnabled is ignored  |false|
+|bookkeeperClientMinNumRacksPerWriteQuorum|  Minimum number of racks per write quorum. BK rack-aware bookie selection policy will try to get bookies from at least 'bookkeeperClientMinNumRacksPerWriteQuorum' racks for a write quorum.  |2|
+|bookkeeperClientEnforceMinNumRacksPerWriteQuorum|  Enforces rack-aware bookie selection policy to pick bookies from 'bookkeeperClientMinNumRacksPerWriteQuorum' racks for a writeQuorum. If BK can't find bookie then it would throw BKNotEnoughBookiesException instead of picking random one.  |false|
 |bookkeeperClientReorderReadSequenceEnabled|  Enable/disable reordering read sequence on reading entries.  |false|
 |bookkeeperClientIsolationGroups| Enable bookie isolation by specifying a list of bookie groups to choose from. Any bookie outside the specified groups will not be used by the broker  ||
 |bookkeeperClientSecondaryIsolationGroups| Enable bookie secondary-isolation group if bookkeeperClientIsolationGroups doesn't have enough bookie available.  ||
@@ -567,6 +569,8 @@ The value of 0 disables message-byte dispatch-throttling.|0|
 |bookkeeperGetBookieInfoRetryIntervalSeconds|Specify options for the GetBookieInfo check. This setting helps ensure the list of bookies that are up to date on the brokers.|60|
 |bookkeeperClientRackawarePolicyEnabled|    |true|
 |bookkeeperClientRegionawarePolicyEnabled|    |false|
+|bookkeeperClientMinNumRacksPerWriteQuorum|    |2|
+|bookkeeperClientMinNumRacksPerWriteQuorum|    |false|
 |bookkeeperClientReorderReadSequenceEnabled|    |false|
 |bookkeeperClientIsolationGroups|||
 |bookkeeperClientSecondaryIsolationGroups| Enable bookie secondary-isolation group if bookkeeperClientIsolationGroups doesn't have enough bookie available.  ||

site2/website/versioned_docs/version-2.6.2/reference-configuration.md

@@ -200,6 +200,8 @@ subscriptionExpirationTimeMinutes | How long to delete inactive subscriptions fr
 |bookkeeperClientHealthCheckQuarantineTimeInSeconds ||1800|
 |bookkeeperClientRackawarePolicyEnabled|  Enable rack-aware bookie selection policy. BK will chose bookies from different racks when forming a new bookie ensemble  |true|
 |bookkeeperClientRegionawarePolicyEnabled|  Enable region-aware bookie selection policy. BK will chose bookies from different regions and racks when forming a new bookie ensemble. If enabled, the value of bookkeeperClientRackawarePolicyEnabled is ignored  |false|
+|bookkeeperClientMinNumRacksPerWriteQuorum|  Minimum number of racks per write quorum. BK rack-aware bookie selection policy will try to get bookies from at least 'bookkeeperClientMinNumRacksPerWriteQuorum' racks for a write quorum.  |2|
+|bookkeeperClientEnforceMinNumRacksPerWriteQuorum|  Enforces rack-aware bookie selection policy to pick bookies from 'bookkeeperClientMinNumRacksPerWriteQuorum' racks for a writeQuorum. If BK can't find bookie then it would throw BKNotEnoughBookiesException instead of picking random one.  |false|
 |bookkeeperClientReorderReadSequenceEnabled|  Enable/disable reordering read sequence on reading entries.  |false|
 |bookkeeperClientIsolationGroups| Enable bookie isolation by specifying a list of bookie groups to choose from. Any bookie outside the specified groups will not be used by the broker  ||
 |bookkeeperClientSecondaryIsolationGroups| Enable bookie secondary-isolation group if bookkeeperClientIsolationGroups doesn't have enough bookie available.  ||