[fix] [bk-client] Fix bk client MinNumRacksPerWriteQuorum and Enforce…

…MinNumRacksPerWriteQuorum not work problem. (apache#21327)
tmdc-io · Oct 12, 2023 · d45a00d · d45a00d
1 parent ab54ea9
commit d45a00d
Show file tree

Hide file tree

Showing 2 changed files with 23 additions and 1 deletion.
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
@@ -219,14 +219,17 @@ static void setDefaultEnsemblePlacementPolicy(
         }
     }
 
-    private void setEnsemblePlacementPolicy(ClientConfiguration bkConf, ServiceConfiguration conf, MetadataStore store,
+    static void setEnsemblePlacementPolicy(ClientConfiguration bkConf, ServiceConfiguration conf, MetadataStore store,
                                             Class<? extends EnsemblePlacementPolicy> policyClass) {
         bkConf.setEnsemblePlacementPolicy(policyClass);
         bkConf.setProperty(BookieRackAffinityMapping.METADATA_STORE_INSTANCE, store);
         if (conf.isBookkeeperClientRackawarePolicyEnabled() || conf.isBookkeeperClientRegionawarePolicyEnabled()) {
             bkConf.setProperty(REPP_DNS_RESOLVER_CLASS, conf.getProperties().getProperty(REPP_DNS_RESOLVER_CLASS,
                     BookieRackAffinityMapping.class.getName()));
 
+            bkConf.setMinNumRacksPerWriteQuorum(conf.getBookkeeperClientMinNumRacksPerWriteQuorum());
+            bkConf.setEnforceMinNumRacksPerWriteQuorum(conf.isBookkeeperClientEnforceMinNumRacksPerWriteQuorum());
+
             bkConf.setProperty(NET_TOPOLOGY_SCRIPT_FILE_NAME_KEY,
                 conf.getProperties().getProperty(
                     NET_TOPOLOGY_SCRIPT_FILE_NAME_KEY,

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
@@ -41,6 +41,7 @@
 import org.apache.pulsar.bookie.rackawareness.BookieRackAffinityMapping;
 import org.apache.pulsar.metadata.api.MetadataStore;
 import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended;
+import org.apache.pulsar.zookeeper.ZkIsolatedBookieEnsemblePlacementPolicy;
 import org.testng.annotations.Test;
 
 /**
@@ -152,6 +153,24 @@ public void testSetDefaultEnsemblePlacementPolicyRackAwareEnabledChangedValues()
         assertEquals(20, bkConf.getMinNumRacksPerWriteQuorum());
     }
 
+    @Test
+    public void testSetEnsemblePlacementPolicys() {
+        ClientConfiguration bkConf = new ClientConfiguration();
+        ServiceConfiguration conf = new ServiceConfiguration();
+        conf.setBookkeeperClientMinNumRacksPerWriteQuorum(3);
+        conf.setBookkeeperClientEnforceMinNumRacksPerWriteQuorum(true);
+
+        MetadataStore store = mock(MetadataStore.class);
+
+        BookKeeperClientFactoryImpl.setEnsemblePlacementPolicy(
+                bkConf,
+                conf,
+                store,
+                ZkIsolatedBookieEnsemblePlacementPolicy.class);
+        assertEquals(bkConf.getMinNumRacksPerWriteQuorum(), 3);
+        assertTrue(bkConf.getEnforceMinNumRacksPerWriteQuorum());
+    }
+
     @Test
     public void testSetDiskWeightBasedPlacementEnabled() {
         BookKeeperClientFactoryImpl factory = new BookKeeperClientFactoryImpl();