Expose new entries check delay in the broker.conf (apache#7154)

Motivation
Expose new entries check delay in the broker.conf

conf/broker.conf

@@ -784,6 +784,12 @@ managedLedgerReadEntryTimeoutSeconds=0
 # Add entry timeout when broker tries to publish message to bookkeeper (0 to disable it).
 managedLedgerAddEntryTimeoutSeconds=0
 
+# New entries check delay for the cursor under the managed ledger.
+# If no new messages in the topic, the cursor will try to check again after the delay time.
+# For consumption latency sensitive scenario, can set to a smaller value or set to 0.
+# Of course, use a smaller value may degrade consumption throughput. Default is 10ms.
+managedLedgerNewEntriesCheckDelayInMillis=10
+
 ### --- Load balancer --- ###
 
 # Enable load balancer

conf/standalone.conf

@@ -567,6 +567,12 @@ managedLedgerReadEntryTimeoutSeconds=0
 # Add entry timeout when broker tries to publish message to bookkeeper (0 to disable it).
 managedLedgerAddEntryTimeoutSeconds=0
 
+# New entries check delay for the cursor under the managed ledger.
+# If no new messages in the topic, the cursor will try to check again after the delay time.
+# For consumption latency sensitive scenario, can set to a smaller value or set to 0.
+# Of course, use a smaller value may degrade consumption throughput. Default is 10ms.
+managedLedgerNewEntriesCheckDelayInMillis=10
+
 # Use Open Range-Set to cache unacked messages
 managedLedgerUnackedRangesOpenCacheSetEnabled=true
 

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java

@@ -70,6 +70,7 @@ public class ManagedLedgerConfig {
     private Class<? extends EnsemblePlacementPolicy>  bookKeeperEnsemblePlacementPolicyClassName;
     private Map<String, Object> bookKeeperEnsemblePlacementPolicyProperties;
     private LedgerOffloader ledgerOffloader = NullLedgerOffloader.INSTANCE;
+    private int newEntriesCheckDelayInMillis = 10;
     private Clock clock = Clock.systemUTC();
 
     public boolean isCreateIfMissing() {
@@ -602,4 +603,12 @@ public boolean isDeletionAtBatchIndexLevelEnabled() {
     public void setDeletionAtBatchIndexLevelEnabled(boolean deletionAtBatchIndexLevelEnabled) {
         this.deletionAtBatchIndexLevelEnabled = deletionAtBatchIndexLevelEnabled;
     }
+
+    public int getNewEntriesCheckDelayInMillis() {
+        return newEntriesCheckDelayInMillis;
+    }
+
+    public void setNewEntriesCheckDelayInMillis(int newEntriesCheckDelayInMillis) {
+        this.newEntriesCheckDelayInMillis = newEntriesCheckDelayInMillis;
+    }
 }

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java

@@ -722,7 +722,7 @@ public void asyncReadEntriesOrWait(int maxEntries, long maxSizeBytes, ReadEntrie
                     // If the managed ledger was indeed terminated, we need to notify the cursor
                     callback.readEntriesFailed(new NoMoreEntriesToReadException("Topic was terminated"), ctx);
                 }
-            }), 10, TimeUnit.MILLISECONDS);
+            }), config.getNewEntriesCheckDelayInMillis(), TimeUnit.MILLISECONDS);
         }
     }
 

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

@@ -1224,6 +1224,13 @@ public class ServiceConfiguration implements PulsarConfiguration {
             doc = "Add entry timeout when broker tries to publish message to bookkeeper.(0 to disable it)")
     private long managedLedgerAddEntryTimeoutSeconds = 0;
 
+    @FieldContext(category = CATEGORY_STORAGE_ML,
+            doc = "New entries check delay for the cursor under the managed ledger. \n"
+                    + "If no new messages in the topic, the cursor will try to check again after the delay time. \n"
+                    + "For consumption latency sensitive scenario, can set to a smaller value or set to 0.\n"
+                    + "Of course, this may degrade consumption throughput. Default is 10ms.")
+    private int managedLedgerNewEntriesCheckDelayInMillis = 10;
+
     /*** --- Load balancer --- ****/
     @FieldContext(
         category = CATEGORY_LOAD_BALANCER,

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java

@@ -1116,6 +1116,7 @@ public CompletableFuture<ManagedLedgerConfig> getManagedLedgerConfig(TopicName t
             managedLedgerConfig.setLedgerOffloader(pulsar.getManagedLedgerOffloader(namespace, offloadPolicies));
 
             managedLedgerConfig.setDeletionAtBatchIndexLevelEnabled(serviceConfig.isAcknowledgmentAtBatchIndexLevelEnabled());
+            managedLedgerConfig.setNewEntriesCheckDelayInMillis(serviceConfig.getManagedLedgerNewEntriesCheckDelayInMillis());
 
             future.complete(managedLedgerConfig);
         }, (exception) -> future.completeExceptionally(exception)));