From cff1b43620c5c0328b41ed9d1baaf39d3ef2c01a Mon Sep 17 00:00:00 2001 From: fengyubiao Date: Tue, 24 Sep 2024 00:19:54 +0800 Subject: [PATCH 1/4] [fix] Key_Shared mode comsumption latency when low traffic --- .../java/org/apache/pulsar/broker/ServiceConfiguration.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) 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 index 486587ec174a0..9fda3ed9efa6c 100644 --- 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 @@ -1231,14 +1231,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 = 0; @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 = 0; @FieldContext( dynamic = true, From 758a51a68cc330214b17567c29eed0c7b9522d1e Mon Sep 17 00:00:00 2001 From: Lari Hotari Date: Mon, 23 Sep 2024 10:41:20 -0700 Subject: [PATCH 2/4] Reduce defaults without disabling the feature so that CPU burning spinning loops are avoided --- .../java/org/apache/pulsar/broker/ServiceConfiguration.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) 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 index 9fda3ed9efa6c..33b4fbff5f5bb 100644 --- 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 @@ -1231,14 +1231,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 = 0; + 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 = 0; + private int dispatcherRetryBackoffMaxTimeInMs = 10; @FieldContext( dynamic = true, From cbef0853ea6ec5a9db01d8aa49dad2d94292f379 Mon Sep 17 00:00:00 2001 From: Lari Hotari Date: Mon, 23 Sep 2024 20:50:16 +0300 Subject: [PATCH 3/4] Adjust settings in config files --- conf/broker.conf | 4 ++-- conf/standalone.conf | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/conf/broker.conf b/conf/broker.conf index 125b2aa8c1b39..617e202e5ec65 100644 --- a/conf/broker.conf +++ b/conf/broker.conf @@ -489,12 +489,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 diff --git a/conf/standalone.conf b/conf/standalone.conf index 622949bf6c325..535800a43f3e0 100644 --- a/conf/standalone.conf +++ b/conf/standalone.conf @@ -305,12 +305,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 From 513498e6b4d7a2cf9613ce9725e97fd635ed4693 Mon Sep 17 00:00:00 2001 From: Lari Hotari Date: Mon, 23 Sep 2024 21:08:07 +0300 Subject: [PATCH 4/4] Reset the backoff whenever entries were dispatched --- .../persistent/PersistentDispatcherMultipleConsumers.java | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java index 450a446c85a78..8fdb65e7b3076 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java @@ -729,11 +729,13 @@ 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 || skipNextBackoff) { skipNextBackoff = false; - // 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) {