[fix] Key_Shared mode consumption latency when low traffic (#23340)

Co-authored-by: Lari Hotari <lhotari@apache.org>
(cherry picked from commit 4ce0c75)

conf/broker.conf

@@ -470,12 +470,12 @@ dispatcherReadFailureBackoffMandatoryStopTimeInMs=0
 # On Shared and KeyShared subscriptions, if all available messages in the subscription are filtered
 # out and not dispatched to any consumer, message dispatching will be rescheduled with a backoff
 # delay. This parameter sets the initial backoff delay in milliseconds.
-dispatcherRetryBackoffInitialTimeInMs=100
+dispatcherRetryBackoffInitialTimeInMs=1
 
 # On Shared and KeyShared subscriptions, if all available messages in the subscription are filtered
 # out and not dispatched to any consumer, message dispatching will be rescheduled with a backoff
 # delay. This parameter sets the maximum backoff delay in milliseconds.
-dispatcherRetryBackoffMaxTimeInMs=1000
+dispatcherRetryBackoffMaxTimeInMs=10
 
 # Precise dispatcher flow control according to history message number of each entry
 preciseDispatcherFlowControl=false

conf/standalone.conf

@@ -286,12 +286,12 @@ dispatcherReadFailureBackoffMandatoryStopTimeInMs=0
 # On Shared and KeyShared subscriptions, if all available messages in the subscription are filtered
 # out and not dispatched to any consumer, message dispatching will be rescheduled with a backoff
 # delay. This parameter sets the initial backoff delay in milliseconds.
-dispatcherRetryBackoffInitialTimeInMs=100
+dispatcherRetryBackoffInitialTimeInMs=1
 
 # On Shared and KeyShared subscriptions, if all available messages in the subscription are filtered
 # out and not dispatched to any consumer, message dispatching will be rescheduled with a backoff
 # delay. This parameter sets the maximum backoff delay in milliseconds.
-dispatcherRetryBackoffMaxTimeInMs=1000
+dispatcherRetryBackoffMaxTimeInMs=10
 
 # Precise dispatcher flow control according to history message number of each entry
 preciseDispatcherFlowControl=false

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

@@ -1198,14 +1198,14 @@ The max allowed delay for delayed delivery (in milliseconds). If the broker rece
             doc = "On Shared and KeyShared subscriptions, if all available messages in the subscription are filtered "
                     + "out and not dispatched to any consumer, message dispatching will be rescheduled with a backoff "
                     + "delay. This parameter sets the initial backoff delay in milliseconds.")
-    private int dispatcherRetryBackoffInitialTimeInMs = 100;
+    private int dispatcherRetryBackoffInitialTimeInMs = 1;
 
     @FieldContext(
             category = CATEGORY_POLICIES,
             doc = "On Shared and KeyShared subscriptions, if all available messages in the subscription are filtered "
                     + "out and not dispatched to any consumer, message dispatching will be rescheduled with a backoff "
                     + "delay. This parameter sets the maximum backoff delay in milliseconds.")
-    private int dispatcherRetryBackoffMaxTimeInMs = 1000;
+    private int dispatcherRetryBackoffMaxTimeInMs = 10;
 
     @FieldContext(
             dynamic = true,

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java

@@ -696,10 +696,12 @@ private synchronized void handleSendingMessagesAndReadingMore(ReadType readType,
         boolean triggerReadingMore = sendMessagesToConsumers(readType, entries, needAcquireSendInProgress);
         int entriesDispatched = lastNumberOfEntriesDispatched;
         updatePendingBytesToDispatch(-totalBytesSize);
+        if (entriesDispatched > 0) {
+            // Reset the backoff when we successfully dispatched messages
+            retryBackoff.reset();
+        }
         if (triggerReadingMore) {
             if (entriesDispatched > 0) {
-                // Reset the backoff when we successfully dispatched messages
-                retryBackoff.reset();
                 // Call readMoreEntries in the same thread to trigger the next read
                 readMoreEntries();
             } else if (entriesDispatched == 0) {