You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2026/10/16 22:31:47 UTC
(pulsar) 24/24: [fix][broker] PIP-442: Fix unbounded executor queue growth and single-thread bottleneck in the topic list memory limiter (#26929)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0-backports-2026-10-17
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 61320fe58f73196b7024bac0e5f95cfab1de3e33
Author: Penghui Li <pe...@apache.org>
AuthorDate: Fri Oct 16 10:45:06 2026 -0700
[fix][broker] PIP-442: Fix unbounded executor queue growth and single-thread bottleneck in the topic list memory limiter (#26929)
(cherry picked from commit 777c489c87d3142a55379eefe639fe69de0fb5f1)
---
.../pulsar/broker/admin/impl/NamespacesBase.java | 7 +-
.../semaphore/AsyncDualMemoryLimiterImpl.java | 35 ++--
.../common/semaphore/AsyncSemaphoreImpl.java | 179 +++++++++++++++------
.../semaphore/AsyncDualMemoryLimiterImplTest.java | 54 +++++++
.../common/semaphore/AsyncSemaphoreImplTest.java | 128 +++++++++++++++
5 files changed, 341 insertions(+), 62 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java
index 1b5a79bf334c..6da45267fcd3 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java
@@ -220,11 +220,14 @@ public abstract class NamespacesBase extends AdminResource {
.exceptionally(t -> {
throw new CompletionException(new RestException(Status.TOO_MANY_REQUESTS,
"Failed due to heap memory limit exceeded"));
- }).thenApply(permits -> {
+ // A permit granted without waiting completes on the calling thread, here a metadata store
+ // callback thread. Continue on the limiter's completion executor, so that writing the
+ // possibly large response doesn't block that thread.
+ }).thenApplyAsync(permits -> {
permitsRef.set(permits);
initialPermitsRef.set(null);
return topicList;
- });
+ }, maxTopicListInFlightLimiter.getCompletionExecutor());
});
}));
}
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/semaphore/AsyncDualMemoryLimiterImpl.java b/pulsar-common/src/main/java/org/apache/pulsar/common/semaphore/AsyncDualMemoryLimiterImpl.java
index 58603851662e..88a5e42b2175 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/semaphore/AsyncDualMemoryLimiterImpl.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/semaphore/AsyncDualMemoryLimiterImpl.java
@@ -21,20 +21,27 @@ package org.apache.pulsar.common.semaphore;
import com.google.common.annotations.VisibleForTesting;
import io.netty.util.concurrent.DefaultThreadFactory;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.function.BooleanSupplier;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* Implementation of AsyncDualMemoryLimiter with separate limits for heap and direct memory.
+ *
+ * <p>Requests that had to wait for permits are completed on a dedicated pool, so their continuations don't run on
+ * the executor that handles the permits and timeouts, also when the common pool isn't usable.
*/
public class AsyncDualMemoryLimiterImpl implements AsyncDualMemoryLimiter, AutoCloseable {
private static final Logger log = LoggerFactory.getLogger(AsyncDualMemoryLimiterImpl.class);
private final ScheduledExecutorService executor;
private final boolean shutdownExecutor;
+ private final ExecutorService completionExecutor;
private final AsyncSemaphoreImpl heapLimiter;
private final AsyncSemaphoreImpl directLimiter;
@@ -46,7 +53,8 @@ public class AsyncDualMemoryLimiterImpl implements AsyncDualMemoryLimiter, AutoC
* @param maxDirectMemory max direct memory available for allocation
* @param maxDirectQueueSize max number of requests that can be queued for direct memory allocation
* @param directTimeoutMillis timeout in milliseconds for direct memory allocation
- * @param executor executor service to use for scheduling timeouts, it is expected to be single threaded
+ * @param executor executor service to use for scheduling timeouts, it is expected to be single threaded and to
+ * remove cancelled tasks, see {@link ScheduledThreadPoolExecutor#setRemoveOnCancelPolicy(boolean)}
*/
public AsyncDualMemoryLimiterImpl(long maxHeapMemory, int maxHeapQueueSize, long heapTimeoutMillis,
long maxDirectMemory, int maxDirectQueueSize, long directTimeoutMillis,
@@ -67,7 +75,7 @@ public class AsyncDualMemoryLimiterImpl implements AsyncDualMemoryLimiter, AutoC
public AsyncDualMemoryLimiterImpl(long maxHeapMemory, int maxHeapQueueSize, long heapTimeoutMillis,
long maxDirectMemory, int maxDirectQueueSize, long directTimeoutMillis) {
this(maxHeapMemory, maxHeapQueueSize, heapTimeoutMillis, maxDirectMemory, maxDirectQueueSize,
- directTimeoutMillis, createExecutor(), true);
+ directTimeoutMillis, AsyncSemaphoreImpl.createExecutor("async-dual-memory-limiter"), true);
}
AsyncDualMemoryLimiterImpl(long maxHeapMemory, int maxHeapQueueSize, long heapTimeoutMillis,
@@ -75,15 +83,13 @@ public class AsyncDualMemoryLimiterImpl implements AsyncDualMemoryLimiter, AutoC
ScheduledExecutorService executor, boolean shutdownExecutor) {
this.executor = executor;
this.shutdownExecutor = shutdownExecutor;
+ // threads are only started when a request had to wait for permits
+ this.completionExecutor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors(),
+ new DefaultThreadFactory("async-dual-memory-limiter-completion"));
this.heapLimiter = new AsyncSemaphoreImpl(maxHeapMemory, maxHeapQueueSize, heapTimeoutMillis, executor,
- this::recordHeapWaitTime);
+ false, completionExecutor, this::recordHeapWaitTime);
this.directLimiter = new AsyncSemaphoreImpl(maxDirectMemory, maxDirectQueueSize, directTimeoutMillis, executor,
- this::recordDirectWaitTime);
- }
-
- private static ScheduledExecutorService createExecutor() {
- return Executors.newSingleThreadScheduledExecutor(
- new DefaultThreadFactory("async-dual-memory-limiter"));
+ false, completionExecutor, this::recordDirectWaitTime);
}
@Override
@@ -144,6 +150,15 @@ public class AsyncDualMemoryLimiterImpl implements AsyncDualMemoryLimiter, AutoC
}
+ /**
+ * Executor for continuations of a granted permit that may block, such as writing a large response. A permit
+ * that is granted without waiting completes on the calling thread, which could be a thread that must not
+ * block, such as a metadata store callback thread.
+ */
+ public Executor getCompletionExecutor() {
+ return completionExecutor;
+ }
+
@Override
public void close() {
heapLimiter.close();
@@ -151,6 +166,8 @@ public class AsyncDualMemoryLimiterImpl implements AsyncDualMemoryLimiter, AutoC
if (shutdownExecutor) {
executor.shutdown();
}
+ // already submitted completions still run after shutdown()
+ completionExecutor.shutdown();
}
private static class DualMemoryLimiterPermit implements AsyncDualMemoryLimiterPermit {
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/semaphore/AsyncSemaphoreImpl.java b/pulsar-common/src/main/java/org/apache/pulsar/common/semaphore/AsyncSemaphoreImpl.java
index 6ee9028711f8..02b17bb09edd 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/semaphore/AsyncSemaphoreImpl.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/semaphore/AsyncSemaphoreImpl.java
@@ -18,14 +18,20 @@
*/
package org.apache.pulsar.common.semaphore;
+import com.google.common.annotations.VisibleForTesting;
import io.netty.util.concurrent.DefaultThreadFactory;
+import java.util.ArrayList;
+import java.util.List;
import java.util.Queue;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.Executors;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
@@ -33,14 +39,20 @@ import java.util.concurrent.atomic.AtomicLongFieldUpdater;
import java.util.function.BooleanSupplier;
import java.util.function.LongConsumer;
import org.apache.pulsar.common.util.Runnables;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
/**
* Implementation of AsyncSemaphore with timeout and queue size limits.
+ *
+ * <p>Permits are granted on the calling thread when no other requests are waiting. Waiting requests are processed
+ * on the executor, which also handles the timeouts. Their futures are completed outside the lock on the completion
+ * executor, so the callers' continuations don't run on the executor thread. {@link AsyncDualMemoryLimiterImpl}
+ * provides a dedicated completion executor. Otherwise {@link ForkJoinPool#commonPool()} is used, or the executor when
+ * the common pool's parallelism is 1 or less, since a disabled common pool can accept tasks without running them.
*/
public class AsyncSemaphoreImpl implements AsyncSemaphore, AutoCloseable {
- private static final Logger log = LoggerFactory.getLogger(AsyncSemaphoreImpl.class);
+ // Same threshold as CompletableFuture and OpReadEntry: parallelism=0 reports 1 and never runs the tasks
+ @VisibleForTesting
+ static final boolean USE_COMMON_POOL = ForkJoinPool.getCommonPoolParallelism() > 1;
private final AtomicLong availablePermits;
private final Queue<PendingRequest> queue;
@@ -48,9 +60,15 @@ public class AsyncSemaphoreImpl implements AsyncSemaphore, AutoCloseable {
private final long timeoutMillis;
private final ScheduledExecutorService executor;
private final boolean shutdownExecutor;
+ private final Executor completionExecutor;
private final LongConsumer queueLatencyRecorder;
private final AtomicBoolean closed = new AtomicBoolean(false);
- private final Runnable processQueueRunnable = Runnables.catchingAndLoggingThrowables(this::internalProcessQueue);
+ // true while a processQueue task is pending, so that at most one such task is queued in the executor
+ private final AtomicBoolean processQueueScheduled = new AtomicBoolean(false);
+ private final Runnable processQueueRunnable = Runnables.catchingAndLoggingThrowables(() -> {
+ processQueueScheduled.set(false);
+ internalProcessQueue();
+ });
private final ScheduledFuture<?> processQueueScheduledFuture;
/**
@@ -61,7 +79,8 @@ public class AsyncSemaphoreImpl implements AsyncSemaphore, AutoCloseable {
* @param timeoutMillis timeout in milliseconds for acquiring permits
*/
public AsyncSemaphoreImpl(long maxPermits, int maxQueueSize, long timeoutMillis) {
- this(maxPermits, maxQueueSize, timeoutMillis, maxPermits > 0 ? createExecutor() : null, maxPermits > 0, null);
+ this(maxPermits, maxQueueSize, timeoutMillis,
+ maxPermits > 0 ? createExecutor("async-semaphore-executor") : null, maxPermits > 0, null);
}
/**
@@ -70,7 +89,8 @@ public class AsyncSemaphoreImpl implements AsyncSemaphore, AutoCloseable {
* recommended)
* @param maxQueueSize max number of requests that can be queued, set to <= 0 for unbounded queue (not recommended)
* @param timeoutMillis timeout in milliseconds for acquiring permits
- * @param executor executor service to use for scheduling timeouts, it is expected to be single threaded
+ * @param executor executor service to use for scheduling timeouts, it is expected to be single threaded and to
+ * remove cancelled tasks, see {@link ScheduledThreadPoolExecutor#setRemoveOnCancelPolicy(boolean)}
* @param queueLatencyRecorder consumer to record queue latency, Long.MAX_VALUE is used for requests that timed out
*/
public AsyncSemaphoreImpl(long maxPermits, int maxQueueSize, long timeoutMillis,
@@ -80,12 +100,19 @@ public class AsyncSemaphoreImpl implements AsyncSemaphore, AutoCloseable {
AsyncSemaphoreImpl(long maxPermits, int maxQueueSize, long timeoutMillis, ScheduledExecutorService executor,
boolean shutdownExecutor, LongConsumer queueLatencyRecorder) {
+ this(maxPermits, maxQueueSize, timeoutMillis, executor, shutdownExecutor,
+ USE_COMMON_POOL ? ForkJoinPool.commonPool() : executor, queueLatencyRecorder);
+ }
+
+ AsyncSemaphoreImpl(long maxPermits, int maxQueueSize, long timeoutMillis, ScheduledExecutorService executor,
+ boolean shutdownExecutor, Executor completionExecutor, LongConsumer queueLatencyRecorder) {
this.availablePermits = new AtomicLong(maxPermits);
this.maxPermits = maxPermits;
this.queue = maxQueueSize > 0 ? new ArrayBlockingQueue<>(maxQueueSize) : new LinkedBlockingQueue<>();
this.timeoutMillis = timeoutMillis;
this.executor = executor;
this.shutdownExecutor = shutdownExecutor;
+ this.completionExecutor = completionExecutor;
this.queueLatencyRecorder = queueLatencyRecorder;
// scheduled task that runs the processQueue method every half of the timeout
// this is to support cancellation in cases where the head of the queue request is blocking others
@@ -94,9 +121,15 @@ public class AsyncSemaphoreImpl implements AsyncSemaphore, AutoCloseable {
? executor.schedule(processQueueRunnable, timeoutMillis / 2, TimeUnit.MILLISECONDS) : null;
}
- private static ScheduledExecutorService createExecutor() {
- return Executors.newSingleThreadScheduledExecutor(
- new DefaultThreadFactory("async-semaphore-executor"));
+ /**
+ * Creates a single threaded executor for the semaphore. Timeout tasks get cancelled when permits are granted, so
+ * cancelled tasks are removed from the queue right away instead of being retained until their delay expires.
+ */
+ static ScheduledThreadPoolExecutor createExecutor(String threadNamePrefix) {
+ ScheduledThreadPoolExecutor executor =
+ new ScheduledThreadPoolExecutor(1, new DefaultThreadFactory(threadNamePrefix));
+ executor.setRemoveOnCancelPolicy(true);
+ return executor;
}
@Override
@@ -120,7 +153,17 @@ public class AsyncSemaphoreImpl implements AsyncSemaphore, AutoCloseable {
return future;
}
+ if (isCancelled.getAsBoolean()) {
+ future.completeExceptionally(new PermitAcquireCancelledException("Permit acquisition was cancelled"));
+ return future;
+ }
+
PendingRequest request = new PendingRequest(permits, previousPermit, future, isCancelled);
+ if (tryAcquireWithoutQueueing(request)) {
+ recordQueueLatency(request.getAgeNanos());
+ future.complete(new SemaphorePermit(permits));
+ return future;
+ }
if (!queue.offer(request)) {
future.completeExceptionally(new PermitAcquireQueueFullException(
"Semaphore queue is full"));
@@ -133,8 +176,8 @@ public class AsyncSemaphoreImpl implements AsyncSemaphore, AutoCloseable {
recordQueueLatency(Long.MAX_VALUE);
// also record the time in the queue
recordQueueLatency(request.getAgeNanos());
- future.completeExceptionally(new PermitAcquireTimeoutException(
- "Permit acquisition timed out"));
+ completeAsync(() -> future.completeExceptionally(new PermitAcquireTimeoutException(
+ "Permit acquisition timed out")));
// the next request might have smaller permits and that might be processed
processQueue();
}
@@ -234,54 +277,88 @@ public class AsyncSemaphoreImpl implements AsyncSemaphore, AutoCloseable {
}
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;
- }
+ 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);
+ }
- PendingRequest request = queue.peek();
- if (request == null) {
- break;
- }
+ // must be called while holding the lock
+ private boolean tryAcquirePermits(PendingRequest request) {
+ long current = availablePermits.get();
+ if (current <= 0 || request.getRequiredPermits() > current) {
+ return false;
+ }
+ availablePermits.addAndGet(-request.getRequiredPermitsReusingPrevious());
+ return true;
+ }
- if (request.isCancelled.getAsBoolean()) {
- request.cancelTimeoutTask();
- queue.remove(request);
- request.future.completeExceptionally(
- new PermitAcquireCancelledException("Permit acquisition was cancelled"));
- continue;
+ 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));
+ }
}
+ } finally {
+ // complete outside the lock, also when an isCancelled supplier throws, so that granted permits don't leak
+ completions.forEach(this::completeAsync);
+ }
+ }
- // request future has been completed by user code cancellation, remove it from the queue
- if (request.future.isDone()) {
- request.cancelTimeoutTask();
- queue.remove(request);
- continue;
- }
+ private void completeAcquiredRequest(PendingRequest request) {
+ recordQueueLatency(request.getAgeNanos());
+ if (!request.future.complete(new SemaphorePermit(request.permits))) {
+ // request was cancelled by user code, return all permits, including those taken over from the
+ // previous permit, which was released when the request was granted
+ availablePermits.addAndGet(request.permits);
+ processQueue();
+ }
+ }
- 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);
- }
- } else {
- break;
- }
+ // the continuations of waiting requests shouldn't run on the executor, otherwise they'd delay all other requests
+ private void completeAsync(Runnable completion) {
+ Runnable task = Runnables.catchingAndLoggingThrowables(completion);
+ try {
+ completionExecutor.execute(task);
+ } catch (RejectedExecutionException e) {
+ // the executors are shut down when closing, complete inline so that a granted request isn't lost
+ task.run();
}
}
diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/semaphore/AsyncDualMemoryLimiterImplTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/semaphore/AsyncDualMemoryLimiterImplTest.java
index 04bd6d61c0b3..617b3094f762 100644
--- a/pulsar-common/src/test/java/org/apache/pulsar/common/semaphore/AsyncDualMemoryLimiterImplTest.java
+++ b/pulsar-common/src/test/java/org/apache/pulsar/common/semaphore/AsyncDualMemoryLimiterImplTest.java
@@ -24,6 +24,7 @@ import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
+import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
@@ -33,6 +34,7 @@ import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.BooleanSupplier;
import org.apache.pulsar.common.semaphore.AsyncDualMemoryLimiter.AsyncDualMemoryLimiterPermit;
@@ -743,4 +745,56 @@ public class AsyncDualMemoryLimiterImplTest {
assertThat(limiter.getLimiter(LimitType.HEAP_MEMORY).getAvailablePermits()).isEqualTo(100000);
assertThat(limiter.getLimiter(LimitType.HEAP_MEMORY).getAcquiredPermits()).isEqualTo(0);
}
+
+ @Test
+ public void testQueuedRequestIsCompletedOnCompletionExecutor() throws Exception {
+ limiter = new AsyncDualMemoryLimiterImpl(100, 10, 5000, 100, 10, 5000);
+ AsyncDualMemoryLimiterPermit permit =
+ limiter.acquire(100, LimitType.HEAP_MEMORY, () -> false).get(1, TimeUnit.SECONDS);
+ CompletableFuture<String> completingThreadName =
+ limiter.acquire(50, LimitType.HEAP_MEMORY, () -> false).thenApply(p -> {
+ limiter.release(p);
+ return Thread.currentThread().getName();
+ });
+ assertThat(completingThreadName).isNotDone();
+
+ limiter.release(permit);
+
+ // independent of the number of CPUs, the continuation doesn't run on the limiter's executor
+ assertThat(completingThreadName).succeedsWithin(Duration.ofSeconds(1))
+ .asString().startsWith("async-dual-memory-limiter-completion");
+ }
+
+ @Test
+ public void testGrantedRequestIsCompletedWhenClosedDuringQueueProcessing() throws Exception {
+ limiter = new AsyncDualMemoryLimiterImpl(2, 10, 60_000, 2, 10, 60_000);
+ AsyncDualMemoryLimiterPermit permit =
+ limiter.acquire(2, LimitType.HEAP_MEMORY, () -> false).get(1, TimeUnit.SECONDS);
+ CompletableFuture<AsyncDualMemoryLimiterPermit> granted =
+ limiter.acquire(1, LimitType.HEAP_MEMORY, () -> false);
+ CountDownLatch processingQueue = new CountDownLatch(1);
+ CountDownLatch closed = new CountDownLatch(1);
+ AtomicInteger isCancelledCalls = new AtomicInteger();
+ limiter.acquire(1, LimitType.HEAP_MEMORY, () -> {
+ // the first call happens before queueing, the second one while processing the queue after the
+ // previous request has been granted
+ if (isCancelledCalls.incrementAndGet() == 2) {
+ processingQueue.countDown();
+ try {
+ closed.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ return false;
+ });
+ limiter.release(permit);
+ assertThat(processingQueue.await(5, TimeUnit.SECONDS)).isTrue();
+
+ // shuts down the executors before the granted request has been handed off for completion
+ limiter.close();
+ closed.countDown();
+
+ assertThat(granted).succeedsWithin(Duration.ofSeconds(5));
+ }
}
\ No newline at end of file
diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/semaphore/AsyncSemaphoreImplTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/semaphore/AsyncSemaphoreImplTest.java
index ec2e01e1a718..a55cf4b63254 100644
--- a/pulsar-common/src/test/java/org/apache/pulsar/common/semaphore/AsyncSemaphoreImplTest.java
+++ b/pulsar-common/src/test/java/org/apache/pulsar/common/semaphore/AsyncSemaphoreImplTest.java
@@ -29,14 +29,20 @@ import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
import org.apache.pulsar.common.semaphore.AsyncSemaphore.AsyncSemaphorePermit;
import org.apache.pulsar.common.semaphore.AsyncSemaphore.PermitAcquireAlreadyClosedException;
import org.apache.pulsar.common.semaphore.AsyncSemaphore.PermitAcquireCancelledException;
import org.apache.pulsar.common.semaphore.AsyncSemaphore.PermitAcquireQueueFullException;
import org.apache.pulsar.common.semaphore.AsyncSemaphore.PermitAcquireTimeoutException;
+import org.awaitility.Awaitility;
+import org.testng.SkipException;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.Test;
@@ -666,4 +672,126 @@ public class AsyncSemaphoreImplTest {
assertThat(semaphore.getAcquiredPermits()).isEqualTo(10);
}
+
+ @Test
+ public void testAcquireWithAvailablePermitsDoesNotUseExecutor() throws Exception {
+ ScheduledThreadPoolExecutor executor = AsyncSemaphoreImpl.createExecutor("test-async-semaphore");
+ CountDownLatch unblockExecutor = new CountDownLatch(1);
+ try {
+ semaphore = new AsyncSemaphoreImpl(10, 10, 5000, executor, null);
+ int initialTaskCount = executor.getQueue().size();
+ blockExecutor(executor, unblockExecutor);
+ for (int i = 0; i < 1000; i++) {
+ CompletableFuture<AsyncSemaphorePermit> future = semaphore.acquire(5, () -> false);
+ // the permits are granted on the calling thread although the executor is blocked
+ assertThat(future).isCompleted();
+ semaphore.release(future.join());
+ }
+ assertThat(executor.getQueue()).hasSize(initialTaskCount);
+ } finally {
+ unblockExecutor.countDown();
+ executor.shutdownNow();
+ }
+ }
+
+ @Test
+ public void testQueuedRequestsDoNotAccumulateExecutorTasks() throws Exception {
+ ScheduledThreadPoolExecutor executor = AsyncSemaphoreImpl.createExecutor("test-async-semaphore");
+ CountDownLatch unblockExecutor = new CountDownLatch(1);
+ try {
+ semaphore = new AsyncSemaphoreImpl(1, 100, 60_000, executor, null);
+ AsyncSemaphorePermit permit = semaphore.acquire(1, () -> false).get(1, TimeUnit.SECONDS);
+ int initialTaskCount = executor.getQueue().size();
+ blockExecutor(executor, unblockExecutor);
+ List<CompletableFuture<AsyncSemaphorePermit>> futures = new ArrayList<>();
+ for (int i = 0; i < 100; i++) {
+ futures.add(semaphore.acquire(1, () -> false));
+ }
+ // a timeout task for each queued request and a single pending task for processing the queue
+ assertThat(executor.getQueue()).hasSize(initialTaskCount + futures.size() + 1);
+
+ unblockExecutor.countDown();
+ semaphore.release(permit);
+ for (CompletableFuture<AsyncSemaphorePermit> future : futures) {
+ semaphore.release(future.get(1, TimeUnit.SECONDS));
+ }
+ // the timeout tasks of the granted requests are removed when they get cancelled
+ Awaitility.await().untilAsserted(() -> assertThat(executor.getQueue()).hasSize(initialTaskCount));
+ } finally {
+ unblockExecutor.countDown();
+ executor.shutdownNow();
+ }
+ }
+
+ @Test
+ public void testAcquireDoesNotBypassWaitingRequests() throws Exception {
+ semaphore = new AsyncSemaphoreImpl(10, 10, 5000);
+ AsyncSemaphorePermit permit = semaphore.acquire(6, () -> false).get(1, TimeUnit.SECONDS);
+ CompletableFuture<AsyncSemaphorePermit> large = semaphore.acquire(5, () -> false);
+ // 4 permits are free, but this request must wait behind the queued one
+ CompletableFuture<AsyncSemaphorePermit> small = semaphore.acquire(2, () -> false);
+ assertThat(small).isNotDone();
+
+ semaphore.release(permit);
+
+ assertThat(large).succeedsWithin(Duration.ofSeconds(1));
+ assertThat(small).succeedsWithin(Duration.ofSeconds(1));
+ }
+
+ @Test
+ public void testCancelledUpdateBetweenGrantAndCompletionReturnsAllPermits() throws Exception {
+ AtomicReference<CompletableFuture<AsyncSemaphorePermit>> toCancel = new AtomicReference<>();
+ // the queue latency is recorded after the request is granted and before its future is completed
+ semaphore = new AsyncSemaphoreImpl(10, 10, 5000, AsyncSemaphoreImpl.createExecutor("test-async-semaphore"),
+ true, latency -> {
+ CompletableFuture<AsyncSemaphorePermit> future = toCancel.get();
+ if (future != null) {
+ future.cancel(false);
+ }
+ });
+ AsyncSemaphorePermit permit = semaphore.acquire(4, () -> false).get(1, TimeUnit.SECONDS);
+ AsyncSemaphorePermit otherPermit = semaphore.acquire(6, () -> false).get(1, TimeUnit.SECONDS);
+ CompletableFuture<AsyncSemaphorePermit> update = semaphore.update(permit, 8, () -> false);
+ assertThat(update).isNotDone();
+ toCancel.set(update);
+
+ // grants the update, which takes over the 4 permits of the previous permit, and then it gets cancelled
+ semaphore.release(otherPermit);
+
+ Awaitility.await().untilAsserted(() -> assertThat(update).isCancelled());
+ Awaitility.await().untilAsserted(() -> assertThat(semaphore.getAvailablePermits()).isEqualTo(10));
+ }
+
+ @Test
+ public void testQueuedRequestIsNotCompletedOnSemaphoreExecutorThread() throws Exception {
+ if (!AsyncSemaphoreImpl.USE_COMMON_POOL) {
+ throw new SkipException("Queued requests are completed on the executor when the common pool is unusable");
+ }
+ semaphore = new AsyncSemaphoreImpl(1, 10, 5000);
+ AsyncSemaphorePermit permit = semaphore.acquire(1, () -> false).get(1, TimeUnit.SECONDS);
+ CompletableFuture<String> completingThreadName = semaphore.acquire(1, () -> false).thenApply(p -> {
+ semaphore.release(p);
+ return Thread.currentThread().getName();
+ });
+ assertThat(completingThreadName).isNotDone();
+
+ semaphore.release(permit);
+
+ // the caller's continuation must not run on the thread that manages the permits
+ assertThat(completingThreadName).succeedsWithin(Duration.ofSeconds(1))
+ .asString().doesNotStartWith("async-semaphore-executor");
+ }
+
+ private static void blockExecutor(ExecutorService executor, CountDownLatch unblock) throws Exception {
+ CountDownLatch blocked = new CountDownLatch(1);
+ executor.execute(() -> {
+ blocked.countDown();
+ try {
+ unblock.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ });
+ assertThat(blocked.await(5, TimeUnit.SECONDS)).isTrue();
+ }
}
\ No newline at end of file