You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "codelipenghui (via GitHub)" <gi...@apache.org> on 2026/10/16 02:13:55 UTC
[PR] [fix][broker] PIP-442: Fix unbounded executor queue growth and single-thread bottleneck in the topic list memory limiter [pulsar]
codelipenghui opened a new pull request, #26929:
URL: https://github.com/apache/pulsar/pull/26929
PIP: [PIP-442](https://github.com/apache/pulsar/blob/master/pip/pip-442.md)
### Motivation
A production broker running 4.0.x with the PIP-442 changes backported (#24833, #25066, #25070) ran out of heap because of tasks queued on the topic list limiter's `async-dual-memory-limiter` thread. The code involved is the same on `master`.
- `jmap -histo` showed 131M `ScheduledThreadPoolExecutor$ScheduledFutureTask` instances (12.6 GB) on a 28 GB heap. Heap in use after GC grew from about 5 GB to about 24 GB in three hours.
- ZGC allocation stalls of about 10 s followed, and publish latency reached p99 59 s. The broker only recovered after a restart.
- Meanwhile the permit queues were at their 10,000 limit while only about 51 MB of the 100 MB heap limit was in use, and the timeout counters stayed at 0.
The cause is in `AsyncSemaphoreImpl`:
1. Every `acquire()` goes through the queue and the single-threaded executor, even when permits are free. Each accepted request adds a timeout task and a `processQueue()` task.
2. `processQueue()` calls are never merged, so pending `processQueue()` tasks pile up.
3. When a permit is granted, the timeout task is cancelled with `cancel(false)`. The executor doesn't use `setRemoveOnCancelPolicy(true)`, so the cancelled task and the `PendingRequest` it references stay in the queue until the 25 s delay expires.
4. `internalProcessQueue()` completes the permit future on the limiter thread while holding the semaphore lock. The caller's continuation therefore runs on that one thread. For `CommandGetTopicsOfNamespace` that means listing and decoding the namespace's topics, filtering, hashing and serializing the response. One thread limits topic-list throughput for the whole broker, and timeouts can't fire while it's busy.
Once requests arrive faster than that thread can run them, the executor queue grows without bound. The 10,000-entry permit queue doesn't prevent this, because it limits pending requests, not executor tasks.
### Modifications
All changes are in `AsyncSemaphoreImpl`, which `AsyncDualMemoryLimiterImpl` and `TopicListMemoryLimiter` use:
- When no requests are waiting and enough permits are free, permits are granted on the calling thread. The returned future is already completed, and no executor tasks are created. A new request still waits behind queued ones (FIFO).
- At most one `processQueue()` task is pending at a time, using a flag that is cleared when the task starts. Nothing is scheduled when the queue is empty.
- The executor is created with `setRemoveOnCancelPolicy(true)`, so cancelled timeout tasks are removed right away. Timeout tasks are only created for requests that actually wait. The number of executor tasks is now bounded by the permit queue size instead of growing with the request rate.
- The futures of waiting requests (grants, cancellations and timeouts) are completed outside the lock on `ForkJoinPool.commonPool()`, so callers' continuations no longer run on the limiter thread.
- The common pool is used only if its parallelism is above 1, the same threshold as `CompletableFuture` and `OpReadEntry`. Otherwise the futures are completed on the limiter's executor.
- This matters because a pool disabled with `-Djava.util.concurrent.ForkJoinPool.common.parallelism=0` reports a parallelism of 1 and accepts tasks without running them.
- A request that is already cancelled is rejected immediately instead of being queued first.
### Verifying this change
- [ ] Make sure that the change passes the CI checks.
This change added tests and can be verified as follows:
- `AsyncSemaphoreImplTest.testAcquireWithAvailablePermitsDoesNotUseExecutor`: an uncontended acquire completes while the executor is blocked and adds no executor tasks.
- `AsyncSemaphoreImplTest.testQueuedRequestsDoNotAccumulateExecutorTasks`: 100 waiting requests add 100 timeout tasks and a single `processQueue()` task (202 tasks before this change). The cancelled timeout tasks are removed once the requests are granted.
- `AsyncSemaphoreImplTest.testQueuedRequestIsNotCompletedOnSemaphoreExecutorThread`: the continuation of a waiting request doesn't run on the executor thread. It's skipped when the common pool isn't used.
- `AsyncSemaphoreImplTest.testAcquireDoesNotBypassWaitingRequests`: a new request doesn't bypass a waiting one even when enough permits are free.
The first three fail without this change. The existing tests in the `pulsar-common` semaphore package also pass. So do `TopicListServiceTest`, `TopicListWatcherTest`, `ServerCnxTest`, `PatternConsumerBackPressureMultipleConsumersTest`, `PatternConsumerTopicWatcherBackPressureMultipleConsumersTest`, `PatternTopicsConsumerImplTest`, `LookupServiceTest` and `ProxyPatternConsumerBackPressureMultipleConsumersTest`. The semaphore tests also pass in a JVM started with `-Djava.util.concurrent.ForkJoinPool.common.parallelism=0`.
I also ran a local throughput check on an 18-core laptop, with 8 calling threads and 200 µs of CPU work in each continuation:
| Scenario | Before | After |
|---|---|---|
| Permits available | ~4,950 req/s | ~39,500 req/s |
| Permits for 4 concurrent holders | ~4,950 req/s | ~18,800 req/s |
Before this change, throughput is capped by the single limiter thread (one request per 200 µs). After it, throughput is limited by the memory permits.
### Notes for reviewers
- **Threading model:**
- When no requests are waiting, the caller's continuation runs on the thread that completed the previous stage, for example the connection's event loop. That was also the behavior before PIP-442, and it still is with an unbounded limiter (`maxTopicListInFlight*MemSizeMB=0`).
- Continuations of requests that had to wait run on the JVM's common pool.
- **Shared pool:** with the default `managedLedgerReadEntriesCallbackInline=true`, managed-ledger read completions also go to the common pool when nesting gets too deep (`OpReadEntry`). The memory permits limit how much topic-list work runs there, but the two could compete during a topic-list storm. I haven't measured this.
- **Not included:** metrics for the executor's queue size and busy time. The executor queue is now bounded, and the existing `topic_list_*_queue_size` metrics show the waiting requests.
- **Backport:** `branch-4.0`, `branch-4.1` and `branch-4.2` contain the same limiter logic.
### Does this pull request potentially affect one of the following parts:
*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
- [x] 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] PIP-442: Fix unbounded executor queue growth and single-thread bottleneck in the topic list memory limiter [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #26929:
URL: https://github.com/apache/pulsar/pull/26929#issuecomment-6219113180
yes, I agree that they are complementary and not exclusive.
--
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] PIP-442: Fix unbounded executor queue growth and single-thread bottleneck in the topic list memory limiter [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #26929:
URL: https://github.com/apache/pulsar/pull/26929#discussion_r4292534006
##########
pulsar-common/src/main/java/org/apache/pulsar/common/semaphore/AsyncSemaphoreImpl.java:
##########
@@ -232,57 +269,86 @@ private SemaphorePermit castToImplementation(AsyncSemaphorePermit permit) {
}
private void processQueue() {
- if (closed.get()) {
+ // a request is added to the queue before this is called, so skipping an empty queue doesn't lose a wakeup
+ if (closed.get() || queue.isEmpty()) {
return;
}
- executor.execute(processQueueRunnable);
+ // a pending task also handles the changes made before this call since it hasn't started yet
+ if (processQueueScheduled.compareAndSet(false, true)) {
+ executor.execute(processQueueRunnable);
+ }
}
- private synchronized void internalProcessQueue() {
- while (!closed.get() && !queue.isEmpty()) {
- long current = availablePermits.get();
- if (current <= 0) {
- break;
- }
-
- PendingRequest request = queue.peek();
- if (request == null) {
- break;
- }
-
- if (request.isCancelled.getAsBoolean()) {
- request.cancelTimeoutTask();
- queue.remove(request);
- request.future.completeExceptionally(
- new PermitAcquireCancelledException("Permit acquisition was cancelled"));
- continue;
- }
+ private synchronized boolean tryAcquireWithoutQueueing(PendingRequest request) {
+ // permits are granted in FIFO order, so a request can't bypass the waiting requests
+ return queue.isEmpty() && tryAcquirePermits(request);
+ }
- // request future has been completed by user code cancellation, remove it from the queue
- if (request.future.isDone()) {
- request.cancelTimeoutTask();
- queue.remove(request);
- continue;
- }
+ // must be called while holding the lock
+ private boolean tryAcquirePermits(PendingRequest request) {
+ long current = availablePermits.get();
+ if (current <= 0 || request.getRequiredPermits() > current) {
+ return false;
+ }
+ request.acquiredPermits = request.getRequiredPermitsReusingPrevious();
+ availablePermits.addAndGet(-request.acquiredPermits);
+ return true;
+ }
- if (request.getRequiredPermits() <= current) {
- long requiredPermitsReusingPrevious = request.getRequiredPermitsReusingPrevious();
- availablePermits.addAndGet(-requiredPermitsReusingPrevious);
- request.cancelTimeoutTask();
- queue.remove(request);
- SemaphorePermit permit = new SemaphorePermit(request.permits);
- recordQueueLatency(request.getAgeNanos());
- boolean futureCompleted = request.future.complete(permit);
- if (!futureCompleted) {
- // request was cancelled by user code, return permits
- availablePermits.addAndGet(requiredPermitsReusingPrevious);
+ private void internalProcessQueue() {
+ List<Runnable> completions = new ArrayList<>();
+ try {
+ synchronized (this) {
+ while (!closed.get()) {
+ PendingRequest request = queue.peek();
+ if (request == null) {
+ break;
+ }
+
+ if (request.isCancelled.getAsBoolean()) {
+ request.cancelTimeoutTask();
+ queue.remove(request);
+ completions.add(() -> request.future.completeExceptionally(
+ new PermitAcquireCancelledException("Permit acquisition was cancelled")));
+ continue;
+ }
+
+ // request future has been completed by user code cancellation, remove it from the queue
+ if (request.future.isDone()) {
+ request.cancelTimeoutTask();
+ queue.remove(request);
+ continue;
+ }
+
+ if (!tryAcquirePermits(request)) {
+ break;
+ }
+ request.cancelTimeoutTask();
+ queue.remove(request);
+ completions.add(() -> completeAcquiredRequest(request));
}
- } else {
- break;
}
+ } finally {
+ // complete outside the lock, also when an isCancelled supplier throws, so that granted permits don't leak
+ completions.forEach(this::completeAsync);
}
}
+ private void completeAcquiredRequest(PendingRequest request) {
+ recordQueueLatency(request.getAgeNanos());
+ if (!request.future.complete(new SemaphorePermit(request.permits))) {
+ // request was cancelled by user code, return permits
+ availablePermits.addAndGet(request.acquiredPermits);
+ processQueue();
+ }
+ }
+
+ // the continuations of waiting requests shouldn't run on the executor, otherwise they'd delay all other requests
+ private void completeAsync(Runnable completion) {
+ (USE_COMMON_POOL ? ForkJoinPool.commonPool() : executor)
Review Comment:
**[QUALITY] a granted completion can be lost when the executor is shut down during close() in the executor fallback**
Minor, shutdown-only, and only when `USE_COMMON_POOL` is false.
If `internalProcessQueue` grants a request (removing it from the queue) and `close()` runs before the `finally` block, `close()` can't fail that request because it's no longer queued, and `AsyncDualMemoryLimiterImpl.close()` ([`AsyncDualMemoryLimiterImpl.java:141-146`](https://github.com/apache/pulsar/blob/c9bf329e08cfdf00b21e2663e40eb41cb32ed4e8/pulsar-common/src/main/java/org/apache/pulsar/common/semaphore/AsyncDualMemoryLimiterImpl.java#L141-L146)) then shuts the executor down. `executor.execute` in `completeAsync` throws `RejectedExecutionException`, so the future is never completed and its permits are never returned. Before this change the completion ran inline, so this couldn't happen.
Catching `RejectedExecutionException` here and running the completion inline would close it.
##########
pulsar-common/src/main/java/org/apache/pulsar/common/semaphore/AsyncSemaphoreImpl.java:
##########
@@ -232,57 +269,86 @@ private SemaphorePermit castToImplementation(AsyncSemaphorePermit permit) {
}
private void processQueue() {
- if (closed.get()) {
+ // a request is added to the queue before this is called, so skipping an empty queue doesn't lose a wakeup
+ if (closed.get() || queue.isEmpty()) {
return;
}
- executor.execute(processQueueRunnable);
+ // a pending task also handles the changes made before this call since it hasn't started yet
+ if (processQueueScheduled.compareAndSet(false, true)) {
+ executor.execute(processQueueRunnable);
+ }
}
- private synchronized void internalProcessQueue() {
- while (!closed.get() && !queue.isEmpty()) {
- long current = availablePermits.get();
- if (current <= 0) {
- break;
- }
-
- PendingRequest request = queue.peek();
- if (request == null) {
- break;
- }
-
- if (request.isCancelled.getAsBoolean()) {
- request.cancelTimeoutTask();
- queue.remove(request);
- request.future.completeExceptionally(
- new PermitAcquireCancelledException("Permit acquisition was cancelled"));
- continue;
- }
+ private synchronized boolean tryAcquireWithoutQueueing(PendingRequest request) {
+ // permits are granted in FIFO order, so a request can't bypass the waiting requests
+ return queue.isEmpty() && tryAcquirePermits(request);
+ }
- // request future has been completed by user code cancellation, remove it from the queue
- if (request.future.isDone()) {
- request.cancelTimeoutTask();
- queue.remove(request);
- continue;
- }
+ // must be called while holding the lock
+ private boolean tryAcquirePermits(PendingRequest request) {
+ long current = availablePermits.get();
+ if (current <= 0 || request.getRequiredPermits() > current) {
+ return false;
+ }
+ request.acquiredPermits = request.getRequiredPermitsReusingPrevious();
+ availablePermits.addAndGet(-request.acquiredPermits);
+ return true;
+ }
- if (request.getRequiredPermits() <= current) {
- long requiredPermitsReusingPrevious = request.getRequiredPermitsReusingPrevious();
- availablePermits.addAndGet(-requiredPermitsReusingPrevious);
- request.cancelTimeoutTask();
- queue.remove(request);
- SemaphorePermit permit = new SemaphorePermit(request.permits);
- recordQueueLatency(request.getAgeNanos());
- boolean futureCompleted = request.future.complete(permit);
- if (!futureCompleted) {
- // request was cancelled by user code, return permits
- availablePermits.addAndGet(requiredPermitsReusingPrevious);
+ private void internalProcessQueue() {
+ List<Runnable> completions = new ArrayList<>();
+ try {
+ synchronized (this) {
+ while (!closed.get()) {
+ PendingRequest request = queue.peek();
+ if (request == null) {
+ break;
+ }
+
+ if (request.isCancelled.getAsBoolean()) {
+ request.cancelTimeoutTask();
+ queue.remove(request);
+ completions.add(() -> request.future.completeExceptionally(
+ new PermitAcquireCancelledException("Permit acquisition was cancelled")));
+ continue;
+ }
+
+ // request future has been completed by user code cancellation, remove it from the queue
+ if (request.future.isDone()) {
+ request.cancelTimeoutTask();
+ queue.remove(request);
+ continue;
+ }
+
+ if (!tryAcquirePermits(request)) {
+ break;
+ }
+ request.cancelTimeoutTask();
+ queue.remove(request);
+ completions.add(() -> completeAcquiredRequest(request));
}
- } else {
- break;
}
+ } finally {
+ // complete outside the lock, also when an isCancelled supplier throws, so that granted permits don't leak
+ completions.forEach(this::completeAsync);
}
}
+ private void completeAcquiredRequest(PendingRequest request) {
+ recordQueueLatency(request.getAgeNanos());
+ if (!request.future.complete(new SemaphorePermit(request.permits))) {
+ // request was cancelled by user code, return permits
+ availablePermits.addAndGet(request.acquiredPermits);
Review Comment:
**[QUALITY] failed completion of an update request does not return the reused permits**
This is not new, and no in-tree caller cancels these futures today, but the asynchronous handoff widens the window, so it may be worth fixing here.
For an `update()` that grows a permit, `tryAcquirePermits` ([`AsyncSemaphoreImpl.java:293`](https://github.com/apache/pulsar/blob/c9bf329e08cfdf00b21e2663e40eb41cb32ed4e8/pulsar-common/src/main/java/org/apache/pulsar/common/semaphore/AsyncSemaphoreImpl.java#L293)) sets `acquiredPermits = permits - previousPermit.releasePermits()`. This is only the *additional* amount, and the previous permit is zeroed. If the request's future has already been completed (user cancelled it) when `completeAcquiredRequest` runs, only `acquiredPermits` is refunded. The permits the old permit held are never returned, because the caller's old permit now reports 0.
Example: capacity 10, permit A holds 4, another holder has 6. `update(A, 8)` is queued; the 6 are released, the request is granted (4 deducted, A zeroed), and the caller cancels the future before the completion task runs. Only 4 are refunded and the original 4 stay unavailable for good.
Refunding `request.permits` (the full amount the new permit would have held) in this branch would fix it, and would make `acquiredPermits` unnecessary; a test cancelling a queued update between grant and completion would pin 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] PIP-442: Fix unbounded executor queue growth and single-thread bottleneck in the topic list memory limiter [pulsar]
Posted by "void-ptr974 (via GitHub)" <gi...@apache.org>.
void-ptr974 commented on code in PR #26929:
URL: https://github.com/apache/pulsar/pull/26929#discussion_r4301241151
##########
pulsar-common/src/main/java/org/apache/pulsar/common/semaphore/AsyncSemaphoreImpl.java:
##########
@@ -232,57 +269,86 @@ private SemaphorePermit castToImplementation(AsyncSemaphorePermit permit) {
}
private void processQueue() {
- if (closed.get()) {
+ // a request is added to the queue before this is called, so skipping an empty queue doesn't lose a wakeup
+ if (closed.get() || queue.isEmpty()) {
return;
}
- executor.execute(processQueueRunnable);
+ // a pending task also handles the changes made before this call since it hasn't started yet
+ if (processQueueScheduled.compareAndSet(false, true)) {
+ executor.execute(processQueueRunnable);
+ }
}
- private synchronized void internalProcessQueue() {
- while (!closed.get() && !queue.isEmpty()) {
- long current = availablePermits.get();
- if (current <= 0) {
- break;
- }
-
- PendingRequest request = queue.peek();
- if (request == null) {
- break;
- }
-
- if (request.isCancelled.getAsBoolean()) {
- request.cancelTimeoutTask();
- queue.remove(request);
- request.future.completeExceptionally(
- new PermitAcquireCancelledException("Permit acquisition was cancelled"));
- continue;
- }
+ private synchronized boolean tryAcquireWithoutQueueing(PendingRequest request) {
+ // permits are granted in FIFO order, so a request can't bypass the waiting requests
+ return queue.isEmpty() && tryAcquirePermits(request);
+ }
- // request future has been completed by user code cancellation, remove it from the queue
- if (request.future.isDone()) {
- request.cancelTimeoutTask();
- queue.remove(request);
- continue;
- }
+ // must be called while holding the lock
+ private boolean tryAcquirePermits(PendingRequest request) {
+ long current = availablePermits.get();
+ if (current <= 0 || request.getRequiredPermits() > current) {
+ return false;
+ }
+ request.acquiredPermits = request.getRequiredPermitsReusingPrevious();
+ availablePermits.addAndGet(-request.acquiredPermits);
+ return true;
+ }
- if (request.getRequiredPermits() <= current) {
- long requiredPermitsReusingPrevious = request.getRequiredPermitsReusingPrevious();
- availablePermits.addAndGet(-requiredPermitsReusingPrevious);
- request.cancelTimeoutTask();
- queue.remove(request);
- SemaphorePermit permit = new SemaphorePermit(request.permits);
- recordQueueLatency(request.getAgeNanos());
- boolean futureCompleted = request.future.complete(permit);
- if (!futureCompleted) {
- // request was cancelled by user code, return permits
- availablePermits.addAndGet(requiredPermitsReusingPrevious);
+ private void internalProcessQueue() {
+ List<Runnable> completions = new ArrayList<>();
+ try {
+ synchronized (this) {
+ while (!closed.get()) {
+ PendingRequest request = queue.peek();
+ if (request == null) {
+ break;
+ }
+
+ if (request.isCancelled.getAsBoolean()) {
+ request.cancelTimeoutTask();
+ queue.remove(request);
+ completions.add(() -> request.future.completeExceptionally(
+ new PermitAcquireCancelledException("Permit acquisition was cancelled")));
+ continue;
+ }
+
+ // request future has been completed by user code cancellation, remove it from the queue
+ if (request.future.isDone()) {
+ request.cancelTimeoutTask();
+ queue.remove(request);
+ continue;
+ }
+
+ if (!tryAcquirePermits(request)) {
+ break;
+ }
+ request.cancelTimeoutTask();
+ queue.remove(request);
+ completions.add(() -> completeAcquiredRequest(request));
}
- } else {
- break;
}
+ } finally {
+ // complete outside the lock, also when an isCancelled supplier throws, so that granted permits don't leak
+ completions.forEach(this::completeAsync);
Review Comment:
Thanks for the detailed explanation and for following up on this. That makes sense—I had conflated client disconnect handling with the JAX-RS CompletionCallback lifecycle. Since the response is only resumed after the permit is published, the scenario I described is not reachable here. The related update() cancellation fix looks good as well.
--
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] PIP-442: Fix unbounded executor queue growth and single-thread bottleneck in the topic list memory limiter [pulsar]
Posted by "void-ptr974 (via GitHub)" <gi...@apache.org>.
void-ptr974 commented on code in PR #26929:
URL: https://github.com/apache/pulsar/pull/26929#discussion_r4294068411
##########
pulsar-common/src/main/java/org/apache/pulsar/common/semaphore/AsyncSemaphoreImpl.java:
##########
@@ -232,57 +269,86 @@ private SemaphorePermit castToImplementation(AsyncSemaphorePermit permit) {
}
private void processQueue() {
- if (closed.get()) {
+ // a request is added to the queue before this is called, so skipping an empty queue doesn't lose a wakeup
+ if (closed.get() || queue.isEmpty()) {
return;
}
- executor.execute(processQueueRunnable);
+ // a pending task also handles the changes made before this call since it hasn't started yet
+ if (processQueueScheduled.compareAndSet(false, true)) {
+ executor.execute(processQueueRunnable);
+ }
}
- private synchronized void internalProcessQueue() {
- while (!closed.get() && !queue.isEmpty()) {
- long current = availablePermits.get();
- if (current <= 0) {
- break;
- }
-
- PendingRequest request = queue.peek();
- if (request == null) {
- break;
- }
-
- if (request.isCancelled.getAsBoolean()) {
- request.cancelTimeoutTask();
- queue.remove(request);
- request.future.completeExceptionally(
- new PermitAcquireCancelledException("Permit acquisition was cancelled"));
- continue;
- }
+ private synchronized boolean tryAcquireWithoutQueueing(PendingRequest request) {
+ // permits are granted in FIFO order, so a request can't bypass the waiting requests
+ return queue.isEmpty() && tryAcquirePermits(request);
+ }
- // request future has been completed by user code cancellation, remove it from the queue
- if (request.future.isDone()) {
- request.cancelTimeoutTask();
- queue.remove(request);
- continue;
- }
+ // must be called while holding the lock
+ private boolean tryAcquirePermits(PendingRequest request) {
+ long current = availablePermits.get();
+ if (current <= 0 || request.getRequiredPermits() > current) {
+ return false;
+ }
+ request.acquiredPermits = request.getRequiredPermitsReusingPrevious();
+ availablePermits.addAndGet(-request.acquiredPermits);
+ return true;
+ }
- if (request.getRequiredPermits() <= current) {
- long requiredPermitsReusingPrevious = request.getRequiredPermitsReusingPrevious();
- availablePermits.addAndGet(-requiredPermitsReusingPrevious);
- request.cancelTimeoutTask();
- queue.remove(request);
- SemaphorePermit permit = new SemaphorePermit(request.permits);
- recordQueueLatency(request.getAgeNanos());
- boolean futureCompleted = request.future.complete(permit);
- if (!futureCompleted) {
- // request was cancelled by user code, return permits
- availablePermits.addAndGet(requiredPermitsReusingPrevious);
+ private void internalProcessQueue() {
+ List<Runnable> completions = new ArrayList<>();
+ try {
+ synchronized (this) {
+ while (!closed.get()) {
+ PendingRequest request = queue.peek();
+ if (request == null) {
+ break;
+ }
+
+ if (request.isCancelled.getAsBoolean()) {
+ request.cancelTimeoutTask();
+ queue.remove(request);
+ completions.add(() -> request.future.completeExceptionally(
+ new PermitAcquireCancelledException("Permit acquisition was cancelled")));
+ continue;
+ }
+
+ // request future has been completed by user code cancellation, remove it from the queue
+ if (request.future.isDone()) {
+ request.cancelTimeoutTask();
+ queue.remove(request);
+ continue;
+ }
+
+ if (!tryAcquirePermits(request)) {
+ break;
+ }
+ request.cancelTimeoutTask();
+ queue.remove(request);
+ completions.add(() -> completeAcquiredRequest(request));
}
- } else {
- break;
}
+ } finally {
+ // complete outside the lock, also when an isCancelled supplier throws, so that granted permits don't leak
+ completions.forEach(this::completeAsync);
Review Comment:
Correction: I rechecked this path, and the scenario described above is not reachable—the permit references are populated before `CompletionCallback` can run. Please disregard this comment.
--
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] PIP-442: Fix unbounded executor queue growth and single-thread bottleneck in the topic list memory limiter [pulsar]
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui merged PR #26929:
URL: https://github.com/apache/pulsar/pull/26929
--
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] PIP-442: Fix unbounded executor queue growth and single-thread bottleneck in the topic list memory limiter [pulsar]
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on PR #26929:
URL: https://github.com/apache/pulsar/pull/26929#issuecomment-6218935454
Thanks, #26935 is a real hazard and worth fixing, but we found no evidence that it played a role in this incident. The load on the affected broker was binary protocol. Its limiter thread was RUNNABLE in ServerCnx.internalHandleGetTopicsOfNamespace, and in a 3 s log sample it logged 530 Failed to acquire direct memory permits for GetTopicsOfNamespace warnings from ServerCnx. In this version every grant runs on the
limiter executor, so any REST topic listing would have its RequestLog line written on the async-dual-memory-limiter thread. We found none, and no /namespaces/{tenant}/{ns}/topics requests at all. A blocked thread also can't explain 131M tasks: once the 10k permit queues fill, further requests are rejected without creating tasks. The backlog needs a sustained throughput deficit, which the executor changes in #26929
address. I see #26929 and #26935 as complementary.
--
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] PIP-442: Fix unbounded executor queue growth and single-thread bottleneck in the topic list memory limiter [pulsar]
Posted by "void-ptr974 (via GitHub)" <gi...@apache.org>.
void-ptr974 commented on code in PR #26929:
URL: https://github.com/apache/pulsar/pull/26929#discussion_r4294068411
##########
pulsar-common/src/main/java/org/apache/pulsar/common/semaphore/AsyncSemaphoreImpl.java:
##########
@@ -232,57 +269,86 @@ private SemaphorePermit castToImplementation(AsyncSemaphorePermit permit) {
}
private void processQueue() {
- if (closed.get()) {
+ // a request is added to the queue before this is called, so skipping an empty queue doesn't lose a wakeup
+ if (closed.get() || queue.isEmpty()) {
return;
}
- executor.execute(processQueueRunnable);
+ // a pending task also handles the changes made before this call since it hasn't started yet
+ if (processQueueScheduled.compareAndSet(false, true)) {
+ executor.execute(processQueueRunnable);
+ }
}
- private synchronized void internalProcessQueue() {
- while (!closed.get() && !queue.isEmpty()) {
- long current = availablePermits.get();
- if (current <= 0) {
- break;
- }
-
- PendingRequest request = queue.peek();
- if (request == null) {
- break;
- }
-
- if (request.isCancelled.getAsBoolean()) {
- request.cancelTimeoutTask();
- queue.remove(request);
- request.future.completeExceptionally(
- new PermitAcquireCancelledException("Permit acquisition was cancelled"));
- continue;
- }
+ private synchronized boolean tryAcquireWithoutQueueing(PendingRequest request) {
+ // permits are granted in FIFO order, so a request can't bypass the waiting requests
+ return queue.isEmpty() && tryAcquirePermits(request);
+ }
- // request future has been completed by user code cancellation, remove it from the queue
- if (request.future.isDone()) {
- request.cancelTimeoutTask();
- queue.remove(request);
- continue;
- }
+ // must be called while holding the lock
+ private boolean tryAcquirePermits(PendingRequest request) {
+ long current = availablePermits.get();
+ if (current <= 0 || request.getRequiredPermits() > current) {
+ return false;
+ }
+ request.acquiredPermits = request.getRequiredPermitsReusingPrevious();
+ availablePermits.addAndGet(-request.acquiredPermits);
+ return true;
+ }
- if (request.getRequiredPermits() <= current) {
- long requiredPermitsReusingPrevious = request.getRequiredPermitsReusingPrevious();
- availablePermits.addAndGet(-requiredPermitsReusingPrevious);
- request.cancelTimeoutTask();
- queue.remove(request);
- SemaphorePermit permit = new SemaphorePermit(request.permits);
- recordQueueLatency(request.getAgeNanos());
- boolean futureCompleted = request.future.complete(permit);
- if (!futureCompleted) {
- // request was cancelled by user code, return permits
- availablePermits.addAndGet(requiredPermitsReusingPrevious);
+ private void internalProcessQueue() {
+ List<Runnable> completions = new ArrayList<>();
+ try {
+ synchronized (this) {
+ while (!closed.get()) {
+ PendingRequest request = queue.peek();
+ if (request == null) {
+ break;
+ }
+
+ if (request.isCancelled.getAsBoolean()) {
+ request.cancelTimeoutTask();
+ queue.remove(request);
+ completions.add(() -> request.future.completeExceptionally(
+ new PermitAcquireCancelledException("Permit acquisition was cancelled")));
+ continue;
+ }
+
+ // request future has been completed by user code cancellation, remove it from the queue
+ if (request.future.isDone()) {
+ request.cancelTimeoutTask();
+ queue.remove(request);
+ continue;
+ }
+
+ if (!tryAcquirePermits(request)) {
+ break;
+ }
+ request.cancelTimeoutTask();
+ queue.remove(request);
+ completions.add(() -> completeAcquiredRequest(request));
}
- } else {
- break;
}
+ } finally {
+ // complete outside the lock, also when an isCancelled supplier throws, so that granted permits don't leak
+ completions.forEach(this::completeAsync);
Review Comment:
Correction: I rechecked this path, and the scenario described above is not reachable—the permit references are populated before `CompletionCallback` can run. Please disregard this comment.
--
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] PIP-442: Fix unbounded executor queue growth and single-thread bottleneck in the topic list memory limiter [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #26929:
URL: https://github.com/apache/pulsar/pull/26929#issuecomment-6214614087
This is an improvement. However, it doesn't address the root cause, which is explained in #26935.
The continuations of the permit futures run on whichever thread completes them. For the REST API, that continuation ends in `AsyncResponse.resume(...)`, which serializes and writes the response on the calling thread with blocking writes, so a slow HTTP client blocks that thread until the response has been written or the idle timeout expires. Completing the futures on `ForkJoinPool.commonPool()` instead of the limiter's executor moves that blocking to common pool threads, but doesn't remove it. #26935 proposes resuming REST responses on the web service executor and making that pool able to grow when its threads are blocked by slow readers. #26936 proposes bounding slow readers with Jetty's `MinimumDataRateHandler`.
--
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] PIP-442: Fix unbounded executor queue growth and single-thread bottleneck in the topic list memory limiter [pulsar]
Posted by "void-ptr974 (via GitHub)" <gi...@apache.org>.
void-ptr974 commented on code in PR #26929:
URL: https://github.com/apache/pulsar/pull/26929#discussion_r4291263661
##########
pulsar-common/src/main/java/org/apache/pulsar/common/semaphore/AsyncSemaphoreImpl.java:
##########
@@ -232,57 +269,86 @@ private SemaphorePermit castToImplementation(AsyncSemaphorePermit permit) {
}
private void processQueue() {
- if (closed.get()) {
+ // a request is added to the queue before this is called, so skipping an empty queue doesn't lose a wakeup
+ if (closed.get() || queue.isEmpty()) {
return;
}
- executor.execute(processQueueRunnable);
+ // a pending task also handles the changes made before this call since it hasn't started yet
+ if (processQueueScheduled.compareAndSet(false, true)) {
+ executor.execute(processQueueRunnable);
+ }
}
- private synchronized void internalProcessQueue() {
- while (!closed.get() && !queue.isEmpty()) {
- long current = availablePermits.get();
- if (current <= 0) {
- break;
- }
-
- PendingRequest request = queue.peek();
- if (request == null) {
- break;
- }
-
- if (request.isCancelled.getAsBoolean()) {
- request.cancelTimeoutTask();
- queue.remove(request);
- request.future.completeExceptionally(
- new PermitAcquireCancelledException("Permit acquisition was cancelled"));
- continue;
- }
+ private synchronized boolean tryAcquireWithoutQueueing(PendingRequest request) {
+ // permits are granted in FIFO order, so a request can't bypass the waiting requests
+ return queue.isEmpty() && tryAcquirePermits(request);
+ }
- // request future has been completed by user code cancellation, remove it from the queue
- if (request.future.isDone()) {
- request.cancelTimeoutTask();
- queue.remove(request);
- continue;
- }
+ // must be called while holding the lock
+ private boolean tryAcquirePermits(PendingRequest request) {
+ long current = availablePermits.get();
+ if (current <= 0 || request.getRequiredPermits() > current) {
+ return false;
+ }
+ request.acquiredPermits = request.getRequiredPermitsReusingPrevious();
+ availablePermits.addAndGet(-request.acquiredPermits);
+ return true;
+ }
- if (request.getRequiredPermits() <= current) {
- long requiredPermitsReusingPrevious = request.getRequiredPermitsReusingPrevious();
- availablePermits.addAndGet(-requiredPermitsReusingPrevious);
- request.cancelTimeoutTask();
- queue.remove(request);
- SemaphorePermit permit = new SemaphorePermit(request.permits);
- recordQueueLatency(request.getAgeNanos());
- boolean futureCompleted = request.future.complete(permit);
- if (!futureCompleted) {
- // request was cancelled by user code, return permits
- availablePermits.addAndGet(requiredPermitsReusingPrevious);
+ private void internalProcessQueue() {
+ List<Runnable> completions = new ArrayList<>();
+ try {
+ synchronized (this) {
+ while (!closed.get()) {
+ PendingRequest request = queue.peek();
+ if (request == null) {
+ break;
+ }
+
+ if (request.isCancelled.getAsBoolean()) {
+ request.cancelTimeoutTask();
+ queue.remove(request);
+ completions.add(() -> request.future.completeExceptionally(
+ new PermitAcquireCancelledException("Permit acquisition was cancelled")));
+ continue;
+ }
+
+ // request future has been completed by user code cancellation, remove it from the queue
+ if (request.future.isDone()) {
+ request.cancelTimeoutTask();
+ queue.remove(request);
+ continue;
+ }
+
+ if (!tryAcquirePermits(request)) {
+ break;
+ }
+ request.cancelTimeoutTask();
+ queue.remove(request);
+ completions.add(() -> completeAcquiredRequest(request));
}
- } else {
- break;
}
+ } finally {
+ // complete outside the lock, also when an isCancelled supplier throws, so that granted permits don't leak
+ completions.forEach(this::completeAsync);
}
}
+ private void completeAcquiredRequest(PendingRequest request) {
+ recordQueueLatency(request.getAgeNanos());
+ if (!request.future.complete(new SemaphorePermit(request.permits))) {
+ // request was cancelled by user code, return permits
+ availablePermits.addAndGet(request.acquiredPermits);
+ processQueue();
+ }
+ }
+
+ // the continuations of waiting requests shouldn't run on the executor, otherwise they'd delay all other requests
+ private void completeAsync(Runnable completion) {
+ (USE_COMMON_POOL ? ForkJoinPool.commonPool() : executor)
Review Comment:
Thanks for addressing the issue of business continuations running on the limiter’s single-threaded executor. CompletableFuture.complete() can execute non-async stages inline, and the current call paths include topic-list size estimation, filtering, hash calculation, and response serialization. Keeping this work away from the executor responsible for permit accounting, queue processing, and timeouts makes the execution boundary much clearer.
Making the completion executor an explicit dependency owned by BrokerService or ProxyService would preserve this separation consistently. On a two-CPU JVM, the common-pool parallelism is normally 1, so the current fallback routes completions back to the limiter executor and allows a slow business continuation to block permit and timeout processing again.
--
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] PIP-442: Fix unbounded executor queue growth and single-thread bottleneck in the topic list memory limiter [pulsar]
Posted by "void-ptr974 (via GitHub)" <gi...@apache.org>.
void-ptr974 commented on code in PR #26929:
URL: https://github.com/apache/pulsar/pull/26929#discussion_r4291273801
##########
pulsar-common/src/main/java/org/apache/pulsar/common/semaphore/AsyncSemaphoreImpl.java:
##########
@@ -232,57 +269,86 @@ private SemaphorePermit castToImplementation(AsyncSemaphorePermit permit) {
}
private void processQueue() {
- if (closed.get()) {
+ // a request is added to the queue before this is called, so skipping an empty queue doesn't lose a wakeup
+ if (closed.get() || queue.isEmpty()) {
return;
}
- executor.execute(processQueueRunnable);
+ // a pending task also handles the changes made before this call since it hasn't started yet
+ if (processQueueScheduled.compareAndSet(false, true)) {
+ executor.execute(processQueueRunnable);
+ }
}
- private synchronized void internalProcessQueue() {
- while (!closed.get() && !queue.isEmpty()) {
- long current = availablePermits.get();
- if (current <= 0) {
- break;
- }
-
- PendingRequest request = queue.peek();
- if (request == null) {
- break;
- }
-
- if (request.isCancelled.getAsBoolean()) {
- request.cancelTimeoutTask();
- queue.remove(request);
- request.future.completeExceptionally(
- new PermitAcquireCancelledException("Permit acquisition was cancelled"));
- continue;
- }
+ private synchronized boolean tryAcquireWithoutQueueing(PendingRequest request) {
+ // permits are granted in FIFO order, so a request can't bypass the waiting requests
+ return queue.isEmpty() && tryAcquirePermits(request);
+ }
- // request future has been completed by user code cancellation, remove it from the queue
- if (request.future.isDone()) {
- request.cancelTimeoutTask();
- queue.remove(request);
- continue;
- }
+ // must be called while holding the lock
+ private boolean tryAcquirePermits(PendingRequest request) {
+ long current = availablePermits.get();
+ if (current <= 0 || request.getRequiredPermits() > current) {
+ return false;
+ }
+ request.acquiredPermits = request.getRequiredPermitsReusingPrevious();
+ availablePermits.addAndGet(-request.acquiredPermits);
+ return true;
+ }
- if (request.getRequiredPermits() <= current) {
- long requiredPermitsReusingPrevious = request.getRequiredPermitsReusingPrevious();
- availablePermits.addAndGet(-requiredPermitsReusingPrevious);
- request.cancelTimeoutTask();
- queue.remove(request);
- SemaphorePermit permit = new SemaphorePermit(request.permits);
- recordQueueLatency(request.getAgeNanos());
- boolean futureCompleted = request.future.complete(permit);
- if (!futureCompleted) {
- // request was cancelled by user code, return permits
- availablePermits.addAndGet(requiredPermitsReusingPrevious);
+ private void internalProcessQueue() {
+ List<Runnable> completions = new ArrayList<>();
+ try {
+ synchronized (this) {
+ while (!closed.get()) {
+ PendingRequest request = queue.peek();
+ if (request == null) {
+ break;
+ }
+
+ if (request.isCancelled.getAsBoolean()) {
+ request.cancelTimeoutTask();
+ queue.remove(request);
+ completions.add(() -> request.future.completeExceptionally(
+ new PermitAcquireCancelledException("Permit acquisition was cancelled")));
+ continue;
+ }
+
+ // request future has been completed by user code cancellation, remove it from the queue
+ if (request.future.isDone()) {
+ request.cancelTimeoutTask();
+ queue.remove(request);
+ continue;
+ }
+
+ if (!tryAcquirePermits(request)) {
+ break;
+ }
+ request.cancelTimeoutTask();
+ queue.remove(request);
+ completions.add(() -> completeAcquiredRequest(request));
}
- } else {
- break;
}
+ } finally {
+ // complete outside the lock, also when an isCancelled supplier throws, so that granted permits don't leak
+ completions.forEach(this::completeAsync);
Review Comment:
One remaining concern is the permit ownership window introduced by this asynchronous handoff.
The concrete HTTP path is NamespacesBase.internalGetListOfTopics(). Its response.register(new CompletionCallback() { ... }) callback sets permitRequestCancelled and releases only the permits currently stored in initialPermitsRef and permitsRef. However, the acquired permit is stored later by thenCompose(initialPermits -> initialPermitsRef.set(initialPermits)), and the updated permit is stored later by thenApply(permits -> permitsRef.set(permits)).
If the HTTP request fails or the client disconnects after internalProcessQueue() has deducted the permits and removed the request, but before the completion task runs, CompletionCallback.onComplete() observes no newly delivered permit and releases nothing. The delayed completion then stores the permit after the callback has finished, and that callback will not run again.
Keeping the request in a granted-but-not-delivered state until the permit is either published or fully rolled back would close this ownership gap.
--
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] PIP-442: Fix unbounded executor queue growth and single-thread bottleneck in the topic list memory limiter [pulsar]
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on code in PR #26929:
URL: https://github.com/apache/pulsar/pull/26929#discussion_r4296666457
##########
pulsar-common/src/main/java/org/apache/pulsar/common/semaphore/AsyncSemaphoreImpl.java:
##########
@@ -232,57 +269,86 @@ private SemaphorePermit castToImplementation(AsyncSemaphorePermit permit) {
}
private void processQueue() {
- if (closed.get()) {
+ // a request is added to the queue before this is called, so skipping an empty queue doesn't lose a wakeup
+ if (closed.get() || queue.isEmpty()) {
return;
}
- executor.execute(processQueueRunnable);
+ // a pending task also handles the changes made before this call since it hasn't started yet
+ if (processQueueScheduled.compareAndSet(false, true)) {
+ executor.execute(processQueueRunnable);
+ }
}
- private synchronized void internalProcessQueue() {
- while (!closed.get() && !queue.isEmpty()) {
- long current = availablePermits.get();
- if (current <= 0) {
- break;
- }
-
- PendingRequest request = queue.peek();
- if (request == null) {
- break;
- }
-
- if (request.isCancelled.getAsBoolean()) {
- request.cancelTimeoutTask();
- queue.remove(request);
- request.future.completeExceptionally(
- new PermitAcquireCancelledException("Permit acquisition was cancelled"));
- continue;
- }
+ private synchronized boolean tryAcquireWithoutQueueing(PendingRequest request) {
+ // permits are granted in FIFO order, so a request can't bypass the waiting requests
+ return queue.isEmpty() && tryAcquirePermits(request);
+ }
- // request future has been completed by user code cancellation, remove it from the queue
- if (request.future.isDone()) {
- request.cancelTimeoutTask();
- queue.remove(request);
- continue;
- }
+ // must be called while holding the lock
+ private boolean tryAcquirePermits(PendingRequest request) {
+ long current = availablePermits.get();
+ if (current <= 0 || request.getRequiredPermits() > current) {
+ return false;
+ }
+ request.acquiredPermits = request.getRequiredPermitsReusingPrevious();
+ availablePermits.addAndGet(-request.acquiredPermits);
+ return true;
+ }
- if (request.getRequiredPermits() <= current) {
- long requiredPermitsReusingPrevious = request.getRequiredPermitsReusingPrevious();
- availablePermits.addAndGet(-requiredPermitsReusingPrevious);
- request.cancelTimeoutTask();
- queue.remove(request);
- SemaphorePermit permit = new SemaphorePermit(request.permits);
- recordQueueLatency(request.getAgeNanos());
- boolean futureCompleted = request.future.complete(permit);
- if (!futureCompleted) {
- // request was cancelled by user code, return permits
- availablePermits.addAndGet(requiredPermitsReusingPrevious);
+ private void internalProcessQueue() {
+ List<Runnable> completions = new ArrayList<>();
+ try {
+ synchronized (this) {
+ while (!closed.get()) {
+ PendingRequest request = queue.peek();
+ if (request == null) {
+ break;
+ }
+
+ if (request.isCancelled.getAsBoolean()) {
+ request.cancelTimeoutTask();
+ queue.remove(request);
+ completions.add(() -> request.future.completeExceptionally(
+ new PermitAcquireCancelledException("Permit acquisition was cancelled")));
+ continue;
+ }
+
+ // request future has been completed by user code cancellation, remove it from the queue
+ if (request.future.isDone()) {
+ request.cancelTimeoutTask();
+ queue.remove(request);
+ continue;
+ }
+
+ if (!tryAcquirePermits(request)) {
+ break;
+ }
+ request.cancelTimeoutTask();
+ queue.remove(request);
+ completions.add(() -> completeAcquiredRequest(request));
}
- } else {
- break;
}
+ } finally {
+ // complete outside the lock, also when an isCancelled supplier throws, so that granted permits don't leak
+ completions.forEach(this::completeAsync);
Review Comment:
Thanks for checking who owns the permit during the handoff. I looked at whether `CompletionCallback.onComplete()` can run while a granted permit is still being handed off, and with the current code it can't:
- JAX-RS runs the `CompletionCallback` when the processing of the resumed response completes. In `Namespaces.getTopics()`, the response is only resumed at the end of this same chain (`thenAccept(response::resume)` and the `exceptionally` branch), after `internalGetListOfTopics()` has completed. By then the permit has been delivered and stored in `initialPermitsRef` or `permitsRef`.
- Nothing completes the response earlier. Pulsar doesn't set an `AsyncResponse` timeout, and Jersey 3.1.x sets the servlet `AsyncContext` timeout to never expire (`AsyncContextDelegateProviderImpl`, `NEVER_TIMEOUT_VALUE = -1`). If the client disconnects, Jersey's servlet `onError` listener only completes the `AsyncContext`. The JAX-RS completion callbacks still run when the chain resumes the response, after the permits have been stored.
So the callback always sees the delivered permits. The same window also existed before this PR, between the grant on the limiter thread and `initialPermitsRef.set(...)` in the continuation. If `NamespacesBase` ever completes the response independently of this chain, the fix would belong there, for example storing the permit and then re-checking a completed flag, because the semaphore can't know when the caller has stored the permit.
Related: when a granted `update()` can't be completed because its future was already completed, all of its permits are now returned (3d00df992a).
--
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