Expose bookkeeperDiskWeightBasedPlacementEnabled in broker.conf (apac…

…he#5254) ### Motivation Allow users to enable bookkeeper DiskWeightBasedPlacement in broker.conf
teazj · Oct 2, 2019 · 6542c65 · 6542c65
1 parent 2ed17a6
commit 6542c65
Show file tree

Hide file tree

Showing 6 changed files with 23 additions and 1 deletion.
diff --git a/conf/broker.conf b/conf/broker.conf
@@ -491,6 +491,9 @@ bookkeeperTLSCertificateFilePath=
 # Path for the trusted TLS certificate file
 bookkeeperTLSTrustCertsFilePath=
 
+# Enable/disable disk weight based placement. Default is false
+bookkeeperDiskWeightBasedPlacementEnabled=false
+
 ### --- Managed Ledger --- ###
 
 # Number of bookies to use when creating a ledger

diff --git a/conf/standalone.conf b/conf/standalone.conf
@@ -308,6 +308,8 @@ bookkeeperTLSCertificateFilePath=
 # Path for the trusted TLS certificate file
 bookkeeperTLSTrustCertsFilePath=
 
+# Enable/disable disk weight based placement. Default is false
+bookkeeperDiskWeightBasedPlacementEnabled=false
 
 ### --- Managed Ledger --- ###
 

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
@@ -817,6 +817,9 @@ public class ServiceConfiguration implements PulsarConfiguration {
     @FieldContext(category = CATEGORY_STORAGE_BK, doc = "Path for the trusted TLS certificate file")
     private String bookkeeperTLSTrustCertsFilePath;
 
+    @FieldContext(category = CATEGORY_STORAGE_BK, doc = "Enable/disable disk weight based placement. Default is false")
+    private boolean bookkeeperDiskWeightBasedPlacementEnabled = false;
+
     /**** --- Managed Ledger --- ****/
     @FieldContext(
         minValue = 1,

diff --git a/...n/src/test/java/org/apache/pulsar/common/configuration/PulsarConfigurationLoaderTest.java b/...n/src/test/java/org/apache/pulsar/common/configuration/PulsarConfigurationLoaderTest.java
@@ -108,6 +108,7 @@ public void testPulsarConfiguraitonLoadingStream() throws Exception {
         printWriter.println("managedLedgerDefaultMarkDeleteRateLimit=5.0");
         printWriter.println("managedLedgerDigestType=CRC32C");
         printWriter.println("managedLedgerCacheSizeMB=");
+        printWriter.println("bookkeeperDiskWeightBasedPlacementEnabled=true");
         printWriter.close();
         testConfigFile.deleteOnExit();
         InputStream stream = new FileInputStream(testConfigFile);
@@ -124,6 +125,7 @@ public void testPulsarConfiguraitonLoadingStream() throws Exception {
         assertFalse(serviceConfig.getWebServicePortTls().isPresent());
         assertEquals(serviceConfig.getManagedLedgerDigestType(), DigestType.CRC32C);
         assertTrue(serviceConfig.getManagedLedgerCacheSizeMB() > 0);
+        assertTrue(serviceConfig.isBookkeeperDiskWeightBasedPlacementEnabled());
     }
 
     @Test

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
@@ -30,6 +30,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 
+import com.google.common.annotations.VisibleForTesting;
 import org.apache.bookkeeper.client.BKException;
 import org.apache.bookkeeper.client.BookKeeper;
 import org.apache.bookkeeper.client.EnsemblePlacementPolicy;
@@ -72,7 +73,8 @@ public BookKeeper create(ServiceConfiguration conf, ZooKeeper zkClient,
         }
     }
 
-    private ClientConfiguration createBkClientConfiguration(ServiceConfiguration conf) {
+    @VisibleForTesting
+    ClientConfiguration createBkClientConfiguration(ServiceConfiguration conf) {
         ClientConfiguration bkConf = new ClientConfiguration();
         if (conf.getBookkeeperClientAuthenticationPlugin() != null
                 && conf.getBookkeeperClientAuthenticationPlugin().trim().length() > 0) {
@@ -102,6 +104,7 @@ private ClientConfiguration createBkClientConfiguration(ServiceConfiguration con
         bkConf.setEnableDigestTypeAutodetection(true);
         bkConf.setStickyReadsEnabled(conf.isBookkeeperEnableStickyReads());
         bkConf.setNettyMaxFrameSizeBytes(conf.getMaxMessageSize() + Commands.MESSAGE_SIZE_FRAME_PADDING);
+        bkConf.setDiskWeightBasedPlacementEnabled(conf.isBookkeeperDiskWeightBasedPlacementEnabled());
 
         if (conf.isBookkeeperClientHealthCheckEnabled()) {
             bkConf.enableBookieHealthCheck();

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
@@ -147,4 +147,13 @@ public void testSetDefaultEnsemblePlacementPolicyRackAwareEnabledChangedValues()
 
     }
 
+    @Test
+    public void testSetDiskWeightBasedPlacementEnabled() {
+        BookKeeperClientFactoryImpl factory = new BookKeeperClientFactoryImpl();
+        ServiceConfiguration conf = new ServiceConfiguration();
+        assertFalse(factory.createBkClientConfiguration(conf).getDiskWeightBasedPlacementEnabled());
+        conf.setBookkeeperDiskWeightBasedPlacementEnabled(true);
+        assertTrue(factory.createBkClientConfiguration(conf).getDiskWeightBasedPlacementEnabled());
+    }
+
 }