Revert "[fix] Key_Shared mode consumption latency when low traffic (#…

…23340)"

This reverts commit 6404cc1.

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=1
+dispatcherRetryBackoffInitialTimeInMs=100
 
 # 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=10
+dispatcherRetryBackoffMaxTimeInMs=1000
 
 # 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=1
+dispatcherRetryBackoffInitialTimeInMs=100
 
 # 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=10
+dispatcherRetryBackoffMaxTimeInMs=1000
 
 # 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 = 1;
+    private int dispatcherRetryBackoffInitialTimeInMs = 100;
 
     @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 = 10;
+    private int dispatcherRetryBackoffMaxTimeInMs = 1000;
 
     @FieldContext(
             dynamic = true,

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

@@ -696,12 +696,10 @@ 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) {