You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "thetumbled (via GitHub)" <gi...@apache.org> on 2026/04/09 08:17:27 UTC
[PR] [fix][broker] Unthrottle producers immediately when publish rate limiting is disabled [pulsar]
thetumbled opened a new pull request, #25502:
URL: https://github.com/apache/pulsar/pull/25502
### Motivation
When publish rate limiting is disabled (for example by clearing the publish rate policy so update(null) is called), token buckets are nulled out immediately, but producers that were already throttled could remain blocked until a previously scheduled unthrottle task ran. That delayed task may have been scheduled with a very long wait time when the token bucket was deeply depleted, so disabling the limit in configuration did not restore traffic quickly. This change triggers the existing lock-free unthrottle path right away after disablement when an executor was already cached from a prior throttle on this limiter.
### Modifications
- In PublishRateLimiterImpl, cache the ScheduledExecutorService used when scheduling unthrottleQueuedProducers after the first throttling event on this limiter instance.
- When update(PublishRate) is called with a null PublishRate (rate limiting disabled), call scheduleUnthrottling(cachedExecutor, 0) so queued producers are processed without waiting on the old delayed task.
### Verifying this change
- [ ] Make sure that the change passes the CI checks.
*(Please pick either of the following options)*
This change is a trivial rework / code cleanup without any test coverage.
### Does this pull request potentially affect one of the following parts:
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
*If the box was checked, please highlight the changes*
- [ ] Dependencies (add or upgrade a dependency)
- [ ] The public API
- [ ] The schema
- [ ] The default values of configurations
- [ ] The threading model
- [ ] The binary protocol
- [ ] The REST endpoints
- [ ] The admin CLI options
- [ ] The metrics
- [ ] Anything that affects deployment
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Unthrottle producers immediately when publish rate limiting is disabled [pulsar]
Posted by "thetumbled (via GitHub)" <gi...@apache.org>.
thetumbled commented on PR #25502:
URL: https://github.com/apache/pulsar/pull/25502#issuecomment-4242845999
Tests have been added. You can comment out the call to `scheduleImmediateUnthrottling();` to verify the original behavior, and the test will output an error. @liangyepianzhou
<img width="832" height="206" alt="image" src="https://github.com/user-attachments/assets/d0a81671-8662-4e3e-8a0d-6124918491b7" />
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Unthrottle producers immediately when publish rate limiting is disabled [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #25502:
URL: https://github.com/apache/pulsar/pull/25502#discussion_r3056833100
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PublishRateLimiterImpl.java:
##########
@@ -88,6 +89,7 @@ private void scheduleDecrementThrottleCount(Producer producer) {
// this is to avoid scheduling unthrottling multiple times for concurrent producers
if (throttledProducersCount.incrementAndGet() == 1) {
ScheduledExecutorService executor = producer.getCnx().getBrokerService().executor().next();
Review Comment:
You should provide (and reuse) a dedicated `Executor` when creating the rate limiter, for example:
```java
private final Executor executor;
public PublishRateLimiterImpl(MonotonicClock monotonicClock,
Consumer<Producer> throttleAction,
Consumer<Producer> unthrottleAction,
Executor executor) {
this.monotonicClock = monotonicClock;
this.throttleAction = throttleAction;
this.unthrottleAction = unthrottleAction;
this.executor = executor;
}
```
This ensures that all scheduling (e.g., unthrottling tasks) runs on a controlled and consistent execution context, rather than relying on ad-hoc or shared threads.
When publish rate limiting is disabled, you can immediately release any throttled producers by invoking:
```java
scheduleUnthrottling(executor, 0L);
```
Using a delay of `0L` guarantees that the unthrottling task is executed asynchronously but without delay, ensuring producers are promptly resumed without blocking the caller thread.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Unthrottle producers immediately when publish rate limiting is disabled [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #25502:
URL: https://github.com/apache/pulsar/pull/25502#issuecomment-4225509138
@thetumbled did the problem happen also with the fix in #25262 and #25269?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Unthrottle producers immediately when publish rate limiting is disabled [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #25502:
URL: https://github.com/apache/pulsar/pull/25502#discussion_r3077607870
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PublishRateLimiterImpl.java:
##########
@@ -88,6 +89,7 @@ private void scheduleDecrementThrottleCount(Producer producer) {
// this is to avoid scheduling unthrottling multiple times for concurrent producers
if (throttledProducersCount.incrementAndGet() == 1) {
ScheduledExecutorService executor = producer.getCnx().getBrokerService().executor().next();
Review Comment:
Therefore I think that the solution in the PR is fine. Perhaps a comment would clarify it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Unthrottle producers immediately when publish rate limiting is disabled [pulsar]
Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou commented on code in PR #25502:
URL: https://github.com/apache/pulsar/pull/25502#discussion_r3063437578
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PublishRateLimiterImpl.java:
##########
Review Comment:
Based on the latest version of the code, this change appears to have a correctness issue. In the path where rate limiting is disabled, the value passed to the following method will never be null:
`org.apache.pulsar.broker.service.PublishRateLimiter#update(org.apache.pulsar.common.policies.data.PublishRate)`
Otherwise, the following method would throw an NPE:
`org.apache.pulsar.broker.service.AbstractTopic#updatePublishRateLimiter`.
Here it should fall back to the broker-level configuration, where the publishRate is never null.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PublishRateLimiterImpl.java:
##########
Review Comment:
```suggestion
if (maxPublishRate != null) {
if (PublishRate.normalize(maxPublishRate) != null) {
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Unthrottle producers immediately when publish rate limiting is disabled [pulsar]
Posted by "thetumbled (via GitHub)" <gi...@apache.org>.
thetumbled commented on code in PR #25502:
URL: https://github.com/apache/pulsar/pull/25502#discussion_r3077808698
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PublishRateLimiterImpl.java:
##########
@@ -189,6 +201,9 @@ protected void updateTokenBuckets(long publishThrottlingRateInMsg, long publishT
} else {
tokenBucketOnByte = null;
}
+ if (tokenBucketOnByte == null && tokenBucketOnMessage == null) {
Review Comment:
good point, I have updated the implementation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Unthrottle producers immediately when publish rate limiting is disabled [pulsar]
Posted by "thetumbled (via GitHub)" <gi...@apache.org>.
thetumbled commented on PR #25502:
URL: https://github.com/apache/pulsar/pull/25502#issuecomment-4234666490
> > That delayed task may have been scheduled with a very long wait time when the token bucket was deeply depleted,
>
> @thetumbled did the problem happen also with the fix in #25262 and #25269?
We haven't encountered this problem in our production environment yet. Since the bucket value is allowed to be negative, I am concerned that the producer may get stuck for a long time. We need a way to disable the throttling feature as quickly as possible.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Unthrottle producers immediately when publish rate limiting is disabled [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #25502:
URL: https://github.com/apache/pulsar/pull/25502#discussion_r3077548261
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PublishRateLimiterImpl.java:
##########
@@ -167,12 +169,22 @@ public void update(Policies policies, String clusterName) {
update(maxPublishRate);
}
+ private void unthrottleImmediately() {
Review Comment:
rename to `scheduleImmediateUnthrottling` so that it better conveys the meaning. In updateTokenBuckets, this method should be called whenever the rates change.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PublishRateLimiterImpl.java:
##########
@@ -189,6 +201,9 @@ protected void updateTokenBuckets(long publishThrottlingRateInMsg, long publishT
} else {
tokenBucketOnByte = null;
}
+ if (tokenBucketOnByte == null && tokenBucketOnMessage == null) {
Review Comment:
remove the condition since either limit might be impacting the throttling. Let's say if it's throttled on bytes and just that limit gets increased, it impacts the throttling decision. `scheduleUnthrottling` will handle it correctly.
The call to `scheduleImmediateUnthrottling` (new name) should be made in every case. The modification in `update` method also relies on this.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PublishRateLimiterImpl.java:
##########
@@ -167,12 +169,22 @@ public void update(Policies policies, String clusterName) {
update(maxPublishRate);
}
+ private void unthrottleImmediately() {
+ ScheduledExecutorService executor = lastUnthrottleExecutor;
+ if (executor != null) {
+ // Wake the existing unthrottle path without waiting on an old delay.
+ scheduleUnthrottling(executor, 0L);
+ }
+ // If executor is null, no throttle has happened yet on this limiter
+ }
+
public void update(PublishRate maxPublishRate) {
if (maxPublishRate != null) {
updateTokenBuckets(maxPublishRate.publishThrottlingRateInMsg, maxPublishRate.publishThrottlingRateInByte);
} else {
tokenBucketOnMessage = null;
tokenBucketOnByte = null;
+ unthrottleImmediately();
Review Comment:
this should be changed to `updateTokenBuckets(0, 0);` to avoid code duplication.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Unthrottle producers immediately when publish rate limiting is disabled [pulsar]
Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou commented on PR #25502:
URL: https://github.com/apache/pulsar/pull/25502#issuecomment-4241904838
Maybe a test should be added to verify that the fix actually works. Otherwise, there's no way to fully confirm whether issues similar to https://github.com/apache/pulsar/pull/25502#discussion_r3063424848 still exist — the scope of manual review is always limited.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Unthrottle producers immediately when publish rate limiting is disabled [pulsar]
Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou commented on PR #25502:
URL: https://github.com/apache/pulsar/pull/25502#issuecomment-4222080826
The scenario is valid, but the fix needs further evaluation. The root cause is that disabling rate limiting should **directly signal** throttled producers to be released, rather than waiting for the next scheduled cycle — i.e., the `disable` semantic itself should trigger an immediate (synchronous or async) unthrottle.
The current fix, via `update(null)`, appears to work around the root cause by adding an operational workaround rather than addressing the underlying code defect. This may fail to resolve the fundamental issue, and worse, increases cognitive overhead for users who need to understand why an extra `update(null)` call is necessary.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Unthrottle producers immediately when publish rate limiting is disabled [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #25502:
URL: https://github.com/apache/pulsar/pull/25502#discussion_r3077602939
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PublishRateLimiterImpl.java:
##########
@@ -88,6 +89,7 @@ private void scheduleDecrementThrottleCount(Producer producer) {
// this is to avoid scheduling unthrottling multiple times for concurrent producers
if (throttledProducersCount.incrementAndGet() == 1) {
ScheduledExecutorService executor = producer.getCnx().getBrokerService().executor().next();
Review Comment:
The current solution picks any executor from the broker service since ordered execution isn't needed for this implementation. `unthrottleQueuedProducers` will return immediately if there's another thread processing the throttled producers. To avoid race conditions, it will schedule one more execution after processing. The purpose of this is to avoid synchronization on hot paths and also to avoid a single thread becoming a bottleneck.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Unthrottle producers immediately when publish rate limiting is disabled [pulsar]
Posted by "thetumbled (via GitHub)" <gi...@apache.org>.
thetumbled merged PR #25502:
URL: https://github.com/apache/pulsar/pull/25502
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Unthrottle producers immediately when publish rate limiting is disabled [pulsar]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #25502:
URL: https://github.com/apache/pulsar/pull/25502#discussion_r3056453427
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PublishRateLimiterImpl.java:
##########
@@ -173,6 +175,12 @@ public void update(PublishRate maxPublishRate) {
} else {
tokenBucketOnMessage = null;
tokenBucketOnByte = null;
+ ScheduledExecutorService executor = lastUnthrottleExecutor;
+ if (executor != null) {
+ // Wake the existing unthrottle path without waiting on an old delay.
+ scheduleUnthrottling(executor, 0L);
+ }
+ // If executor is null, no throttle has happened yet on this limiter
}
Review Comment:
This adds new behavior (immediate unthrottling on `update(null)`) that is timing- and race-sensitive. Please add a test that (1) throttles at least one producer enough to schedule a long unthrottle delay, then (2) disables rate limiting via `update(null)`, and (3) asserts queued producers are unthrottled promptly (without waiting for the originally scheduled delay).
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PublishRateLimiterImpl.java:
##########
@@ -173,6 +175,12 @@ public void update(PublishRate maxPublishRate) {
} else {
tokenBucketOnMessage = null;
tokenBucketOnByte = null;
+ ScheduledExecutorService executor = lastUnthrottleExecutor;
+ if (executor != null) {
+ // Wake the existing unthrottle path without waiting on an old delay.
+ scheduleUnthrottling(executor, 0L);
+ }
Review Comment:
`lastUnthrottleExecutor` can become stale (e.g., executor shutdown during broker/namespace lifecycle changes). If `scheduleUnthrottling(...)` ultimately calls `ScheduledExecutorService#schedule`, this may throw `RejectedExecutionException` and break the update path when disabling rate limiting. Consider handling task rejection (catch `RejectedExecutionException`, clear `lastUnthrottleExecutor`, and/or fall back to a known-live executor) so disabling limits is best-effort and resilient.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Unthrottle producers immediately when publish rate limiting is disabled [pulsar]
Posted by "thetumbled (via GitHub)" <gi...@apache.org>.
thetumbled commented on code in PR #25502:
URL: https://github.com/apache/pulsar/pull/25502#discussion_r3061692646
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PublishRateLimiterImpl.java:
##########
@@ -88,6 +89,7 @@ private void scheduleDecrementThrottleCount(Producer producer) {
// this is to avoid scheduling unthrottling multiple times for concurrent producers
if (throttledProducersCount.incrementAndGet() == 1) {
ScheduledExecutorService executor = producer.getCnx().getBrokerService().executor().next();
Review Comment:
WDYT. @lhotari
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Unthrottle producers immediately when publish rate limiting is disabled [pulsar]
Posted by "thetumbled (via GitHub)" <gi...@apache.org>.
thetumbled commented on code in PR #25502:
URL: https://github.com/apache/pulsar/pull/25502#discussion_r3071670612
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PublishRateLimiterImpl.java:
##########
Review Comment:
Good point. The latest commit handle this situation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Unthrottle producers immediately when publish rate limiting is disabled [pulsar]
Posted by "thetumbled (via GitHub)" <gi...@apache.org>.
thetumbled commented on PR #25502:
URL: https://github.com/apache/pulsar/pull/25502#issuecomment-4242170344
> Maybe a test should be added to verify that the fix actually works. Otherwise, there's no way to fully confirm whether issues similar to [#25502 (comment)](https://github.com/apache/pulsar/pull/25502#discussion_r3063424848) still exist — the scope of manual review is always limited.
I will add one later.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org