Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

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

Merged
merged 4 commits into from
Sep 23, 2024
Merged

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
cff1b43
[fix] Key_Shared mode comsumption latency when low traffic
poorbarcode Sep 23, 2024
758a51a
Reduce defaults without disabling the feature so that CPU burning spi…
lhotari Sep 23, 2024
cbef085
Adjust settings in config files
lhotari Sep 23, 2024
513498e
Reset the backoff whenever entries were dispatched
lhotari Sep 23, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions conf/broker.conf
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
4 changes: 2 additions & 2 deletions conf/standalone.conf
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
4 changes: 2 additions & 2 deletions pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 = 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,
Expand Down
6 changes: 4 additions & 2 deletions ...va/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand Down
Loading