Expose bookkeeper expose explicit lac in broker.conf (apache#5822)

### Motivation

Expose bookkeeper expose explicit lac configuration in broker.conf
It's related to apache#3828 apache#4976, some Pulsar SQL users need to enable the explicitLacInterval, so that they can get the last message in Pulsar SQL.

conf/broker.conf

@@ -528,6 +528,10 @@ bookkeeperTLSTrustCertsFilePath=
 # Enable/disable disk weight based placement. Default is false
 bookkeeperDiskWeightBasedPlacementEnabled=false
 
+# Set the interval to check the need for sending an explicit LAC
+# A value of '0' disables sending any explicit LACs. Default is 0.
+bookkeeperExplicitLacIntervalInMills=0;
+
 ### --- Managed Ledger --- ###
 
 # Number of bookies to use when creating a ledger

conf/standalone.conf

@@ -334,6 +334,10 @@ bookkeeperTLSTrustCertsFilePath=
 # Enable/disable disk weight based placement. Default is false
 bookkeeperDiskWeightBasedPlacementEnabled=false
 
+# Set the interval to check the need for sending an explicit LAC
+# A value of '0' disables sending any explicit LACs. Default is 0.
+bookkeeperExplicitLacIntervalInMills=0;
+
 ### --- Managed Ledger --- ###
 
 # Number of bookies to use when creating a ledger

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

@@ -879,6 +879,9 @@ public class ServiceConfiguration implements PulsarConfiguration {
     @FieldContext(category = CATEGORY_STORAGE_BK, doc = "Enable/disable disk weight based placement. Default is false")
     private boolean bookkeeperDiskWeightBasedPlacementEnabled = false;
 
+    @FieldContext(category = CATEGORY_STORAGE_BK, doc = "Set the interval to check the need for sending an explicit LAC")
+    private int bookkeeperExplicitLacIntervalInMills = 0;
+
     /**** --- Managed Ledger --- ****/
     @FieldContext(
         minValue = 1,

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

@@ -115,6 +115,7 @@ ClientConfiguration createBkClientConfiguration(ServiceConfiguration conf) {
         }
 
         bkConf.setReorderReadSequenceEnabled(conf.isBookkeeperClientReorderReadSequenceEnabled());
+        bkConf.setExplictLacInterval(conf.getBookkeeperExplicitLacIntervalInMills());
 
         return bkConf;
     }

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

@@ -78,6 +78,7 @@ private File createValidBrokerConfigFile() throws FileNotFoundException {
         printWriter.println("bookkeeperClientTimeoutInSeconds=12345");
         printWriter.println("bookkeeperClientSpeculativeReadTimeoutInMillis=3000");
         printWriter.println("enableRunBookieTogether=true");
+        printWriter.println("bookkeeperExplicitLacIntervalInMills=5");
 
         printWriter.close();
         testConfigFile.deleteOnExit();
@@ -127,6 +128,7 @@ public void testLoadConfig() throws SecurityException, NoSuchMethodException, IO
         assertEquals(serviceConfig.getBookkeeperClientIsolationGroups(), "group1,group2");
         assertEquals(serviceConfig.getBookkeeperClientSpeculativeReadTimeoutInMillis(), 3000);
         assertEquals(serviceConfig.getBookkeeperClientTimeoutInSeconds(), 12345);
+        assertEquals(serviceConfig.getBookkeeperExplicitLacIntervalInMills(), 5);
     }
 
     @Test

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

@@ -159,4 +159,13 @@ public void testSetDiskWeightBasedPlacementEnabled() {
         assertTrue(factory.createBkClientConfiguration(conf).getDiskWeightBasedPlacementEnabled());
     }
 
+    @Test
+    public void testSetExplicitLacInterval() {
+        BookKeeperClientFactoryImpl factory = new BookKeeperClientFactoryImpl();
+        ServiceConfiguration conf = new ServiceConfiguration();
+        assertEquals(factory.createBkClientConfiguration(conf).getExplictLacInterval(), 0);
+        conf.setBookkeeperExplicitLacIntervalInMills(5);
+        assertEquals(factory.createBkClientConfiguration(conf).getExplictLacInterval(), 5);
+    }
+
 }