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 19:05:06 UTC
(pulsar) branch master updated: [improve][broker] Write async REST responses on the web service executor (#26946)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new df9a90f4460e [improve][broker] Write async REST responses on the web service executor (#26946)
df9a90f4460e is described below
commit df9a90f4460e73b399d59e6d7c0e3c56cd2a5b17
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Fri Oct 16 12:05:00 2026 -0700
[improve][broker] Write async REST responses on the web service executor (#26946)
---
.../pulsar/broker/web/WebExecutorThreadPool.java | 14 +-
.../broker/web/WebExecutorThreadPoolTest.java | 20 +
.../broker/web/WebExecutorAsyncResponse.java | 201 ++++++++++
.../web/WebExecutorAsyncResponseFeature.java | 106 ++++++
.../org/apache/pulsar/broker/web/WebService.java | 2 +
.../broker/web/WebExecutorAsyncResponseTest.java | 419 +++++++++++++++++++++
.../broker/web/WebServiceAsyncResponseTest.java | 120 ++++++
7 files changed, 881 insertions(+), 1 deletion(-)
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/web/WebExecutorThreadPool.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/web/WebExecutorThreadPool.java
index c44901913dd3..9677e8b776a7 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/web/WebExecutorThreadPool.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/web/WebExecutorThreadPool.java
@@ -25,6 +25,8 @@ import org.eclipse.jetty.util.thread.ExecutorThreadPool;
public class WebExecutorThreadPool extends ExecutorThreadPool {
+ private static final ThreadLocal<WebExecutorThreadPool> CURRENT_POOL = new ThreadLocal<>();
+
private final ThreadFactory threadFactory;
private volatile boolean startAttempted;
@@ -39,7 +41,17 @@ public class WebExecutorThreadPool extends ExecutorThreadPool {
@Override
protected Thread newThread(Runnable job) {
- return threadFactory.newThread(job);
+ return threadFactory.newThread(() -> {
+ CURRENT_POOL.set(this);
+ job.run();
+ });
+ }
+
+ /**
+ * Returns whether the calling thread is one of this pool's threads.
+ */
+ public boolean isPoolThread() {
+ return CURRENT_POOL.get() == this;
}
@Override
diff --git a/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/web/WebExecutorThreadPoolTest.java b/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/web/WebExecutorThreadPoolTest.java
index 1bf4b372a82d..ccf037f917ad 100644
--- a/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/web/WebExecutorThreadPoolTest.java
+++ b/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/web/WebExecutorThreadPoolTest.java
@@ -61,6 +61,26 @@ public class WebExecutorThreadPoolTest {
assertJoinCompletes(pool);
}
+ @Test(timeOut = 30000)
+ public void testIsPoolThread() throws Exception {
+ WebExecutorThreadPool pool = new WebExecutorThreadPool(4, "test-web");
+ WebExecutorThreadPool otherPool = new WebExecutorThreadPool(4, "test-other-web");
+ pool.start();
+ otherPool.start();
+ try {
+ assertThat(pool.isPoolThread()).isFalse();
+ CompletableFuture<Boolean> onPoolThread = new CompletableFuture<>();
+ pool.execute(() -> onPoolThread.complete(pool.isPoolThread()));
+ assertThat(onPoolThread).succeedsWithin(10, TimeUnit.SECONDS).isEqualTo(true);
+ CompletableFuture<Boolean> onOtherPoolThread = new CompletableFuture<>();
+ otherPool.execute(() -> onOtherPoolThread.complete(pool.isPoolThread()));
+ assertThat(onOtherPoolThread).succeedsWithin(10, TimeUnit.SECONDS).isEqualTo(false);
+ } finally {
+ pool.stop();
+ otherPool.stop();
+ }
+ }
+
private static void assertJoinCompletes(WebExecutorThreadPool pool) {
CompletableFuture<Void> joined = CompletableFuture.runAsync(() -> {
try {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/WebExecutorAsyncResponse.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/WebExecutorAsyncResponse.java
new file mode 100644
index 000000000000..9e9a6b5cae41
--- /dev/null
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/WebExecutorAsyncResponse.java
@@ -0,0 +1,201 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.web;
+
+import jakarta.ws.rs.container.AsyncResponse;
+import jakarta.ws.rs.container.TimeoutHandler;
+import java.util.Collection;
+import java.util.Date;
+import java.util.Map;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.BooleanSupplier;
+import lombok.CustomLog;
+
+/**
+ * An {@link AsyncResponse} that completes the request on the web service executor.
+ *
+ * <p>Jersey writes the response on the thread that calls {@code resume} or {@code cancel}, and the servlet output
+ * stream blocks that thread while a slow client reads the response. Async REST endpoints usually resume in a
+ * {@code CompletableFuture} continuation, which runs on the thread that completed the future, such as a metadata
+ * store callback thread. This wrapper hands the completion to the web service executor, the pool meant for blocking
+ * response writing, unless the calling thread already belongs to it. When the executor rejects the task, the
+ * request is completed on the calling thread so that the response isn't lost.
+ *
+ * <p>The first {@code resume} or {@code cancel} call decides how the request completes, even while its completion
+ * is still queued. Later calls fail like they do on a completed response, except that a cancellation succeeds
+ * after an earlier cancellation. {@link #isSuspended()}, {@link #isCancelled()} and {@link #isDone()} report
+ * Jersey's state, so they change once the queued completion has run.
+ *
+ * <p>Jersey's timeout handling uses its own response, not this wrapper, so a timeout could complete the request
+ * while a completion is queued and the queued completion would be lost. Once a timeout or a timeout handler is
+ * set, the request is therefore completed on the calling thread. Pulsar's endpoints don't use timeouts.
+ */
+@CustomLog
+final class WebExecutorAsyncResponse implements AsyncResponse {
+
+ private final AsyncResponse delegate;
+ private final WebExecutorThreadPool executor;
+ private static final int NOT_REQUESTED = 0;
+ private static final int RESUME_REQUESTED = 1;
+ private static final int CANCEL_REQUESTED = 2;
+ private static final long REJECTION_WARNING_INTERVAL_NANOS = TimeUnit.MINUTES.toNanos(1);
+
+ // the responses that were completed on the calling thread since the last warning about it
+ private static final AtomicLong rejectedSinceWarning = new AtomicLong();
+ private static final AtomicLong nextRejectionWarningNanos = new AtomicLong(System.nanoTime());
+
+ // Set by the first resume or cancel call, so that later calls behave like they would on a completed response
+ // while the first completion is still waiting in the executor's queue.
+ private final AtomicInteger requestedCompletion = new AtomicInteger(NOT_REQUESTED);
+ private volatile boolean timeoutSet;
+
+ WebExecutorAsyncResponse(AsyncResponse delegate, WebExecutorThreadPool executor) {
+ this.delegate = delegate;
+ this.executor = executor;
+ }
+
+ @Override
+ public boolean resume(Object response) {
+ return complete(RESUME_REQUESTED, () -> delegate.resume(response));
+ }
+
+ @Override
+ public boolean resume(Throwable response) {
+ return complete(RESUME_REQUESTED, () -> delegate.resume(response));
+ }
+
+ @Override
+ public boolean cancel() {
+ return complete(CANCEL_REQUESTED, delegate::cancel);
+ }
+
+ @Override
+ public boolean cancel(int retryAfter) {
+ return complete(CANCEL_REQUESTED, () -> delegate.cancel(retryAfter));
+ }
+
+ @Override
+ public boolean cancel(Date retryAfter) {
+ return complete(CANCEL_REQUESTED, () -> delegate.cancel(retryAfter));
+ }
+
+ private boolean complete(int completionType, BooleanSupplier completion) {
+ int previousCompletion = requestedCompletion.compareAndExchange(NOT_REQUESTED, completionType);
+ if (previousCompletion != NOT_REQUESTED) {
+ // like on a completed response, cancelling again succeeds if the response was cancelled. Check isDone()
+ // first, since the queued cancellation can complete between the two checks.
+ return completionType == CANCEL_REQUESTED && previousCompletion == CANCEL_REQUESTED
+ && (!delegate.isDone() || delegate.isCancelled());
+ }
+ if (timeoutSet || delegate.isDone() || executor.isPoolThread()) {
+ // a completed response doesn't write anything, so there's no need to hand it to the executor
+ return completion.getAsBoolean();
+ }
+ try {
+ executor.execute(() -> {
+ try {
+ completion.getAsBoolean();
+ } catch (Throwable t) {
+ log.warn().exception(t).log("Failed to complete the async HTTP response");
+ }
+ });
+ return true;
+ } catch (RejectedExecutionException e) {
+ logRejection(e);
+ return completion.getAsBoolean();
+ }
+ }
+
+ private void logRejection(RejectedExecutionException e) {
+ if (!executor.isRunning()) {
+ // the web service is stopping
+ log.debug()
+ .exceptionMessage(e)
+ .log("The web service executor rejected the response, completing it on the calling thread");
+ return;
+ }
+ rejectedSinceWarning.incrementAndGet();
+ long now = System.nanoTime();
+ long nextWarningNanos = nextRejectionWarningNanos.get();
+ // warn at most once per interval, since the executor rejects tasks while it is overloaded
+ if (now - nextWarningNanos >= 0
+ && nextRejectionWarningNanos.compareAndSet(nextWarningNanos, now + REJECTION_WARNING_INTERVAL_NANOS)) {
+ log.warn()
+ .attr("rejectedResponses", rejectedSinceWarning.getAndSet(0))
+ .exceptionMessage(e)
+ .log("The web service executor's queue is full, so async HTTP responses are written on the"
+ + " threads that complete them, which can block those threads while clients read the"
+ + " responses. Consider increasing numHttpServerThreads or httpServerThreadPoolQueueSize");
+ }
+ }
+
+ @Override
+ public boolean isSuspended() {
+ return delegate.isSuspended();
+ }
+
+ @Override
+ public boolean isCancelled() {
+ return delegate.isCancelled();
+ }
+
+ @Override
+ public boolean isDone() {
+ return delegate.isDone();
+ }
+
+ @Override
+ public boolean setTimeout(long time, TimeUnit unit) {
+ timeoutSet = true;
+ if (requestedCompletion.get() != NOT_REQUESTED) {
+ // like on a resumed response
+ return false;
+ }
+ return delegate.setTimeout(time, unit);
+ }
+
+ @Override
+ public void setTimeoutHandler(TimeoutHandler handler) {
+ timeoutSet = true;
+ delegate.setTimeoutHandler(handler);
+ }
+
+ @Override
+ public Collection<Class<?>> register(Class<?> callback) {
+ return delegate.register(callback);
+ }
+
+ @Override
+ public Map<Class<?>, Collection<Class<?>>> register(Class<?> callback, Class<?>... callbacks) {
+ return delegate.register(callback, callbacks);
+ }
+
+ @Override
+ public Collection<Class<?>> register(Object callback) {
+ return delegate.register(callback);
+ }
+
+ @Override
+ public Map<Class<?>, Collection<Class<?>>> register(Object callback, Object... callbacks) {
+ return delegate.register(callback, callbacks);
+ }
+}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/WebExecutorAsyncResponseFeature.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/WebExecutorAsyncResponseFeature.java
new file mode 100644
index 000000000000..ed6ad643a35c
--- /dev/null
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/WebExecutorAsyncResponseFeature.java
@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.web;
+
+import jakarta.ws.rs.container.AsyncResponse;
+import jakarta.ws.rs.core.Feature;
+import jakarta.ws.rs.core.FeatureContext;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Objects;
+import java.util.function.Function;
+import org.glassfish.jersey.InjectionManagerProvider;
+import org.glassfish.jersey.internal.inject.AbstractBinder;
+import org.glassfish.jersey.internal.inject.InjectionManager;
+import org.glassfish.jersey.server.ContainerRequest;
+import org.glassfish.jersey.server.model.Parameter;
+import org.glassfish.jersey.server.spi.internal.ValueParamProvider;
+
+/**
+ * Makes the {@link AsyncResponse} injected for {@code @Suspended} parameters complete the request on the web service
+ * executor, see {@link WebExecutorAsyncResponse}.
+ *
+ * <p>It registers a {@link ValueParamProvider} with a higher priority than Jersey's own provider for
+ * {@code @Suspended} parameters, and wraps the {@link AsyncResponse} that Jersey's provider supplies.
+ */
+public class WebExecutorAsyncResponseFeature implements Feature {
+
+ private final WebExecutorThreadPool executor;
+
+ public WebExecutorAsyncResponseFeature(WebExecutorThreadPool executor) {
+ this.executor = Objects.requireNonNull(executor);
+ }
+
+ @Override
+ public boolean configure(FeatureContext context) {
+ InjectionManager injectionManager = InjectionManagerProvider.getInjectionManager(context);
+ context.register(new AbstractBinder() {
+ @Override
+ protected void configure() {
+ bind(new AsyncResponseValueParamProvider(injectionManager, executor)).to(ValueParamProvider.class);
+ }
+ });
+ return true;
+ }
+
+ private static final class AsyncResponseValueParamProvider implements ValueParamProvider {
+
+ private final InjectionManager injectionManager;
+ private final WebExecutorThreadPool executor;
+
+ AsyncResponseValueParamProvider(InjectionManager injectionManager, WebExecutorThreadPool executor) {
+ this.injectionManager = injectionManager;
+ this.executor = executor;
+ }
+
+ @Override
+ public Function<ContainerRequest, ?> getValueProvider(Parameter parameter) {
+ // Jersey's provider also supplies subtypes of AsyncResponse, such as its AsyncContext, which the
+ // wrapper doesn't implement, so leave those parameters to it
+ if (parameter.getSource() != Parameter.Source.SUSPENDED
+ || !parameter.getRawType().isAssignableFrom(WebExecutorAsyncResponse.class)) {
+ return null;
+ }
+ // Delegate to the provider that would be used without this one, which is Jersey's own provider
+ List<ValueParamProvider> providers = injectionManager.getAllInstances(ValueParamProvider.class);
+ Function<ContainerRequest, ?> delegateProvider = providers.stream()
+ .filter(provider -> provider != this)
+ .sorted(Comparator.comparingInt(
+ (ValueParamProvider provider) -> provider.getPriority().getWeight()).reversed())
+ .map(provider -> provider.getValueProvider(parameter))
+ .filter(Objects::nonNull)
+ .findFirst()
+ .orElse(null);
+ if (delegateProvider == null) {
+ return null;
+ }
+ return request -> {
+ Object asyncResponse = delegateProvider.apply(request);
+ return asyncResponse instanceof AsyncResponse
+ ? new WebExecutorAsyncResponse((AsyncResponse) asyncResponse, executor)
+ : asyncResponse;
+ };
+ }
+
+ @Override
+ public PriorityType getPriority() {
+ return Priority.HIGH;
+ }
+ }
+}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/WebService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/WebService.java
index 72a3ecd4b5ef..fac9c82b157f 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/WebService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/WebService.java
@@ -256,6 +256,8 @@ public class WebService implements AutoCloseable {
config.register(JsonMapperProvider.class);
}
config.register(MultiPartFeature.class);
+ // Write the responses of async endpoints on the web service executor instead of the thread that resumes them
+ config.register(new WebExecutorAsyncResponseFeature(webServiceExecutor));
ServletHolder servletHolder = new ServletHolder(new ServletContainer(config));
servletHolder.setAsyncSupported(true);
addServlet(basePath, servletHolder, requiresAuthentication, attributeMap);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/web/WebExecutorAsyncResponseTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/web/WebExecutorAsyncResponseTest.java
new file mode 100644
index 000000000000..e084c59f751f
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/web/WebExecutorAsyncResponseTest.java
@@ -0,0 +1,419 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.web;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.timeout;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import io.netty.util.concurrent.DefaultThreadFactory;
+import jakarta.ws.rs.GET;
+import jakarta.ws.rs.Path;
+import jakarta.ws.rs.WebApplicationException;
+import jakarta.ws.rs.container.AsyncResponse;
+import jakarta.ws.rs.container.CompletionCallback;
+import jakarta.ws.rs.container.ContainerRequestContext;
+import jakarta.ws.rs.container.Suspended;
+import jakarta.ws.rs.core.Context;
+import jakarta.ws.rs.core.Response;
+import jakarta.ws.rs.ext.WriterInterceptor;
+import jakarta.ws.rs.ext.WriterInterceptorContext;
+import java.io.IOException;
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+import org.eclipse.jetty.ee10.servlet.ServletContextHandler;
+import org.eclipse.jetty.ee10.servlet.ServletHolder;
+import org.eclipse.jetty.server.Server;
+import org.eclipse.jetty.server.ServerConnector;
+import org.glassfish.jersey.server.AsyncContext;
+import org.glassfish.jersey.server.ResourceConfig;
+import org.glassfish.jersey.servlet.ServletContainer;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+@Test(groups = "broker")
+public class WebExecutorAsyncResponseTest {
+
+ private static final String WEB_THREAD_PREFIX = "test-web";
+ private static final String OTHER_THREAD_PREFIX = "test-other";
+ private static final String WRITER_THREAD_HEADER = "X-Writer-Thread";
+ private static final String RESPONSE_WRITTEN_PROPERTY = "test.responseWritten";
+
+ private ExecutorService otherExecutor;
+ private WebExecutorThreadPool webExecutor;
+ private Server server;
+ private URI baseUri;
+ private HttpClient httpClient;
+ private TestResource testResource;
+
+ @BeforeClass
+ public void setup() throws Exception {
+ otherExecutor = Executors.newSingleThreadExecutor(new DefaultThreadFactory(OTHER_THREAD_PREFIX));
+ webExecutor = new WebExecutorThreadPool(8, WEB_THREAD_PREFIX);
+ server = new Server(webExecutor);
+ ServerConnector connector = new ServerConnector(server);
+ connector.setPort(0);
+ server.addConnector(connector);
+
+ ResourceConfig config = new ResourceConfig();
+ testResource = new TestResource(otherExecutor);
+ config.register(testResource);
+ config.register(WriterThreadInterceptor.class);
+ config.register(new WebExecutorAsyncResponseFeature(webExecutor));
+ ServletHolder servletHolder = new ServletHolder(new ServletContainer(config));
+ servletHolder.setAsyncSupported(true);
+ ServletContextHandler context = new ServletContextHandler();
+ context.addServlet(servletHolder, "/*");
+ server.setHandler(context);
+ server.start();
+
+ baseUri = URI.create("http://localhost:" + connector.getLocalPort());
+ httpClient = HttpClient.newHttpClient();
+ }
+
+ @AfterClass(alwaysRun = true)
+ public void cleanup() throws Exception {
+ if (httpClient != null) {
+ httpClient.close();
+ }
+ if (server != null) {
+ server.stop();
+ }
+ if (otherExecutor != null) {
+ otherExecutor.shutdownNow();
+ }
+ }
+
+ @Path("/test")
+ public static class TestResource {
+ private final ExecutorService otherExecutor;
+ private volatile CompletableFuture<String> completionCallbackResult = new CompletableFuture<>();
+
+ public TestResource(ExecutorService otherExecutor) {
+ this.otherExecutor = otherExecutor;
+ }
+
+ @GET
+ @Path("/resumeFromOtherThread")
+ public void resumeFromOtherThread(@Suspended AsyncResponse asyncResponse) {
+ otherExecutor.execute(() -> asyncResponse.resume("resumed"));
+ }
+
+ @GET
+ @Path("/resumeExceptionFromOtherThread")
+ public void resumeExceptionFromOtherThread(@Suspended AsyncResponse asyncResponse) {
+ otherExecutor.execute(() -> asyncResponse.resume(new WebApplicationException(
+ Response.status(Response.Status.CONFLICT).entity("failed").build())));
+ }
+
+ @GET
+ @Path("/resumeWithCompletionCallback")
+ public void resumeWithCompletionCallback(@Suspended AsyncResponse asyncResponse,
+ @Context ContainerRequestContext requestContext) {
+ CompletableFuture<String> result = completionCallbackResult;
+ // the interceptor sets it once it has written the response
+ AtomicBoolean responseWritten = new AtomicBoolean();
+ requestContext.setProperty(RESPONSE_WRITTEN_PROPERTY, responseWritten);
+ asyncResponse.register((CompletionCallback) throwable -> result.complete(
+ "written=" + responseWritten.get() + ", thread=" + Thread.currentThread().getName()));
+ otherExecutor.execute(() -> asyncResponse.resume("resumed"));
+ }
+
+ @GET
+ @Path("/resumeAsyncContextFromOtherThread")
+ public void resumeAsyncContextFromOtherThread(@Suspended AsyncContext asyncContext) {
+ otherExecutor.execute(() -> asyncContext.resume("resumed"));
+ }
+
+ @GET
+ @Path("/resumeOnRequestThread")
+ public void resumeOnRequestThread(@Suspended AsyncResponse asyncResponse) {
+ asyncResponse.resume("resumed");
+ }
+ }
+
+ public static class WriterThreadInterceptor implements WriterInterceptor {
+ @Override
+ public void aroundWriteTo(WriterInterceptorContext context) throws IOException {
+ context.getHeaders().add(WRITER_THREAD_HEADER, Thread.currentThread().getName());
+ context.proceed();
+ if (context.getProperty(RESPONSE_WRITTEN_PROPERTY) instanceof AtomicBoolean responseWritten) {
+ responseWritten.set(true);
+ }
+ }
+ }
+
+ private HttpResponse<String> get(String path) throws Exception {
+ return httpClient.send(HttpRequest.newBuilder(baseUri.resolve(path)).GET().build(),
+ HttpResponse.BodyHandlers.ofString());
+ }
+
+ @Test
+ public void testResumeFromOtherThreadWritesOnWebExecutor() throws Exception {
+ HttpResponse<String> response = get("/test/resumeFromOtherThread");
+ assertThat(response.statusCode()).isEqualTo(200);
+ assertThat(response.body()).isEqualTo("resumed");
+ assertThat(response.headers().firstValue(WRITER_THREAD_HEADER)).hasValueSatisfying(
+ threadName -> assertThat(threadName).startsWith(WEB_THREAD_PREFIX));
+ }
+
+ @Test
+ public void testResumeExceptionFromOtherThreadWritesOnWebExecutor() throws Exception {
+ HttpResponse<String> response = get("/test/resumeExceptionFromOtherThread");
+ assertThat(response.statusCode()).isEqualTo(409);
+ assertThat(response.body()).isEqualTo("failed");
+ assertThat(response.headers().firstValue(WRITER_THREAD_HEADER)).hasValueSatisfying(
+ threadName -> assertThat(threadName).startsWith(WEB_THREAD_PREFIX));
+ }
+
+ @Test
+ public void testCompletionCallbackRunsAfterResponseIsWritten() throws Exception {
+ // callbacks such as the topic list permit release in NamespacesBase rely on running after the write
+ CompletableFuture<String> callbackResult = new CompletableFuture<>();
+ testResource.completionCallbackResult = callbackResult;
+ HttpResponse<String> response = get("/test/resumeWithCompletionCallback");
+ assertThat(response.statusCode()).isEqualTo(200);
+ assertThat(response.body()).isEqualTo("resumed");
+ assertThat(callbackResult.get(10, TimeUnit.SECONDS))
+ .startsWith("written=true, thread=" + WEB_THREAD_PREFIX);
+ }
+
+ @Test
+ public void testAsyncContextParameterIsLeftToJersey() throws Exception {
+ // the wrapper doesn't implement Jersey's AsyncContext, so such a parameter gets Jersey's own object
+ HttpResponse<String> response = get("/test/resumeAsyncContextFromOtherThread");
+ assertThat(response.statusCode()).isEqualTo(200);
+ assertThat(response.body()).isEqualTo("resumed");
+ }
+
+ @Test
+ public void testResumeOnRequestThread() throws Exception {
+ HttpResponse<String> response = get("/test/resumeOnRequestThread");
+ assertThat(response.statusCode()).isEqualTo(200);
+ assertThat(response.body()).isEqualTo("resumed");
+ assertThat(response.headers().firstValue(WRITER_THREAD_HEADER)).hasValueSatisfying(
+ threadName -> assertThat(threadName).startsWith(WEB_THREAD_PREFIX));
+ }
+
+ @Test
+ public void testResumeOnPoolThreadRunsInline() throws Exception {
+ AsyncResponse delegate = mock(AsyncResponse.class);
+ when(delegate.resume(any(Object.class))).thenReturn(true);
+ WebExecutorAsyncResponse asyncResponse = new WebExecutorAsyncResponse(delegate, webExecutor);
+ AtomicReference<Boolean> result = new AtomicReference<>();
+ CountDownLatch done = new CountDownLatch(1);
+ webExecutor.execute(() -> {
+ result.set(asyncResponse.resume("value"));
+ // resumed before returning, on this thread
+ verify(delegate).resume("value");
+ done.countDown();
+ });
+ assertThat(done.await(10, TimeUnit.SECONDS)).isTrue();
+ assertThat(result.get()).isTrue();
+ }
+
+ @Test
+ public void testCompletionIsDispatchedOnceWhileQueued() throws Exception {
+ WebExecutorThreadPool pool = new WebExecutorThreadPool(2, "test-blocked");
+ pool.start();
+ CountDownLatch release = new CountDownLatch(1);
+ try {
+ blockPool(pool, 2, release);
+ AsyncResponse delegate = mock(AsyncResponse.class);
+ when(delegate.isSuspended()).thenReturn(true);
+ CompletableFuture<String> resumingThread = new CompletableFuture<>();
+ when(delegate.resume(any(Object.class))).thenAnswer(invocation -> {
+ resumingThread.complete(Thread.currentThread().getName());
+ return true;
+ });
+ WebExecutorAsyncResponse asyncResponse = new WebExecutorAsyncResponse(delegate, pool);
+
+ assertThat(asyncResponse.resume("first")).isTrue();
+ // Jersey's state is reported until the queued completion has run
+ assertThat(asyncResponse.isDone()).isFalse();
+ assertThat(asyncResponse.isSuspended()).isTrue();
+ assertThat(asyncResponse.resume("second")).isFalse();
+ assertThat(asyncResponse.resume(new RuntimeException("third"))).isFalse();
+ assertThat(asyncResponse.cancel()).isFalse();
+ assertThat(asyncResponse.setTimeout(1, TimeUnit.SECONDS)).isFalse();
+ verify(delegate, never()).resume(any(Object.class));
+
+ release.countDown();
+ // Mockito records the invocation before it runs the answer, so wait for the answer
+ assertThat(resumingThread.get(10, TimeUnit.SECONDS)).startsWith("test-blocked");
+ verify(delegate).resume("first");
+ verify(delegate, never()).resume("second");
+ verify(delegate, never()).resume(any(Throwable.class));
+ verify(delegate, never()).cancel();
+ verify(delegate, never()).setTimeout(anyLong(), any());
+ } finally {
+ release.countDown();
+ pool.stop();
+ }
+ }
+
+ @Test
+ public void testCancellingAgainSucceeds() throws Exception {
+ WebExecutorThreadPool pool = new WebExecutorThreadPool(2, "test-blocked");
+ pool.start();
+ CountDownLatch release = new CountDownLatch(1);
+ try {
+ blockPool(pool, 2, release);
+ AsyncResponse delegate = mock(AsyncResponse.class);
+ when(delegate.cancel()).thenReturn(true);
+ WebExecutorAsyncResponse asyncResponse = new WebExecutorAsyncResponse(delegate, pool);
+
+ assertThat(asyncResponse.cancel()).isTrue();
+ // cancelling again succeeds while the cancellation is queued
+ assertThat(asyncResponse.cancel()).isTrue();
+ assertThat(asyncResponse.cancel(10)).isTrue();
+ assertThat(asyncResponse.resume("value")).isFalse();
+
+ release.countDown();
+ verify(delegate, timeout(10000)).cancel();
+ when(delegate.isDone()).thenReturn(true);
+ when(delegate.isCancelled()).thenReturn(true);
+ // and after it has run
+ assertThat(asyncResponse.cancel()).isTrue();
+ verify(delegate).cancel();
+ verify(delegate, never()).cancel(anyInt());
+ verify(delegate, never()).resume(any(Object.class));
+ } finally {
+ release.countDown();
+ pool.stop();
+ }
+ }
+
+ @Test
+ public void testCancellingAgainSucceedsWhileQueuedCancellationCompletes() throws Exception {
+ WebExecutorThreadPool pool = new WebExecutorThreadPool(2, "test-blocked");
+ pool.start();
+ CountDownLatch release = new CountDownLatch(1);
+ try {
+ blockPool(pool, 2, release);
+ AsyncResponse delegate = mock(AsyncResponse.class);
+ AtomicBoolean cancelled = new AtomicBoolean();
+ when(delegate.isDone()).thenAnswer(invocation -> cancelled.get());
+ // the queued cancellation completes right after the first check of Jersey's state
+ when(delegate.isCancelled()).thenAnswer(invocation -> cancelled.getAndSet(true));
+ WebExecutorAsyncResponse asyncResponse = new WebExecutorAsyncResponse(delegate, pool);
+
+ assertThat(asyncResponse.cancel()).isTrue();
+ assertThat(asyncResponse.cancel()).isTrue();
+ assertThat(asyncResponse.cancel()).isTrue();
+ } finally {
+ release.countDown();
+ pool.stop();
+ }
+ }
+
+ @Test
+ public void testCompletionRunsOnCallingThreadWhenTimeoutIsSet() throws Exception {
+ WebExecutorThreadPool pool = new WebExecutorThreadPool(2, "test-blocked");
+ pool.start();
+ CountDownLatch release = new CountDownLatch(1);
+ try {
+ blockPool(pool, 2, release);
+ AsyncResponse delegate = mock(AsyncResponse.class);
+ when(delegate.setTimeout(anyLong(), any())).thenReturn(true);
+ AtomicReference<Thread> resumingThread = new AtomicReference<>();
+ when(delegate.resume(any(Object.class))).thenAnswer(invocation -> {
+ resumingThread.set(Thread.currentThread());
+ return true;
+ });
+ WebExecutorAsyncResponse asyncResponse = new WebExecutorAsyncResponse(delegate, pool);
+
+ assertThat(asyncResponse.setTimeout(1, TimeUnit.SECONDS)).isTrue();
+ // a Jersey timeout can't overtake a queued completion, since the completion isn't queued
+ assertThat(asyncResponse.resume("value")).isTrue();
+ verify(delegate).resume("value");
+ assertThat(resumingThread.get()).isSameAs(Thread.currentThread());
+ } finally {
+ release.countDown();
+ pool.stop();
+ }
+ }
+
+ @Test
+ public void testRejectedCompletionRunsOnCallingThread() throws Exception {
+ WebExecutorThreadPool pool = new WebExecutorThreadPool(2, "test-stopped");
+ pool.start();
+ pool.stop();
+ AsyncResponse delegate = mock(AsyncResponse.class);
+ AtomicReference<Thread> resumingThread = new AtomicReference<>();
+ when(delegate.resume(any(Object.class))).thenAnswer(invocation -> {
+ resumingThread.set(Thread.currentThread());
+ return true;
+ });
+ WebExecutorAsyncResponse asyncResponse = new WebExecutorAsyncResponse(delegate, pool);
+
+ assertThat(asyncResponse.resume("value")).isTrue();
+ verify(delegate).resume("value");
+ assertThat(resumingThread.get()).isSameAs(Thread.currentThread());
+ }
+
+ @Test
+ public void testCompletedDelegateIsCompletedOnCallingThread() {
+ AsyncResponse delegate = mock(AsyncResponse.class);
+ when(delegate.isDone()).thenReturn(true);
+ AtomicReference<Thread> resumingThread = new AtomicReference<>();
+ when(delegate.resume(any(Object.class))).thenAnswer(invocation -> {
+ resumingThread.set(Thread.currentThread());
+ return false;
+ });
+ WebExecutorAsyncResponse asyncResponse = new WebExecutorAsyncResponse(delegate, webExecutor);
+
+ // Jersey's response decides the result, and doesn't write anything since it's already done
+ assertThat(asyncResponse.resume("value")).isFalse();
+ verify(delegate).resume("value");
+ assertThat(resumingThread.get()).isSameAs(Thread.currentThread());
+ }
+
+ // occupies all the pool's threads so that completions stay queued until release is counted down
+ private static void blockPool(WebExecutorThreadPool pool, int poolSize, CountDownLatch release)
+ throws InterruptedException {
+ CountDownLatch poolBlocked = new CountDownLatch(poolSize);
+ for (int i = 0; i < poolSize; i++) {
+ pool.execute(() -> {
+ poolBlocked.countDown();
+ try {
+ release.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ });
+ }
+ assertThat(poolBlocked.await(10, TimeUnit.SECONDS)).isTrue();
+ }
+}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/web/WebServiceAsyncResponseTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/web/WebServiceAsyncResponseTest.java
new file mode 100644
index 000000000000..d60342b43a45
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/web/WebServiceAsyncResponseTest.java
@@ -0,0 +1,120 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.web;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import io.netty.util.concurrent.DefaultThreadFactory;
+import jakarta.ws.rs.GET;
+import jakarta.ws.rs.Path;
+import jakarta.ws.rs.container.AsyncResponse;
+import jakarta.ws.rs.container.Suspended;
+import jakarta.ws.rs.ext.WriterInterceptor;
+import jakarta.ws.rs.ext.WriterInterceptorContext;
+import java.io.IOException;
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+/**
+ * Tests that the REST resources that {@link WebService} adds, like the broker's own resources, write the responses
+ * of async endpoints on the web service executor.
+ */
+@Test(groups = "broker")
+public class WebServiceAsyncResponseTest extends MockedPulsarServiceBaseTest {
+
+ private static final String OTHER_THREAD_PREFIX = "test-other";
+ private static final String WRITER_THREAD_HEADER = "X-Writer-Thread";
+
+ private static ExecutorService otherExecutor;
+ private WebService webService;
+ private HttpClient httpClient;
+
+ @BeforeClass
+ @Override
+ protected void setup() throws Exception {
+ super.internalSetup();
+ otherExecutor = Executors.newSingleThreadExecutor(new DefaultThreadFactory(OTHER_THREAD_PREFIX));
+ // a second web service of the test broker, since resources can't be added once a web service has started
+ Optional<Integer> brokerWebServicePort = conf.getWebServicePort();
+ conf.setWebServicePort(Optional.of(0));
+ try {
+ webService = new WebService(pulsar);
+ webService.addRestResource("/", false, Map.of(), true, TestResource.class,
+ WriterThreadInterceptor.class);
+ webService.start();
+ } finally {
+ // the web service sets the port that it listens on in the configuration
+ conf.setWebServicePort(brokerWebServicePort);
+ }
+ httpClient = HttpClient.newHttpClient();
+ }
+
+ @AfterClass(alwaysRun = true)
+ @Override
+ protected void cleanup() throws Exception {
+ if (httpClient != null) {
+ httpClient.close();
+ }
+ if (webService != null) {
+ webService.close();
+ }
+ if (otherExecutor != null) {
+ otherExecutor.shutdownNow();
+ }
+ super.internalCleanup();
+ }
+
+ @Path("/test")
+ public static class TestResource {
+ @GET
+ @Path("/resumeFromOtherThread")
+ public void resumeFromOtherThread(@Suspended AsyncResponse asyncResponse) {
+ otherExecutor.execute(() -> asyncResponse.resume("resumed"));
+ }
+ }
+
+ public static class WriterThreadInterceptor implements WriterInterceptor {
+ @Override
+ public void aroundWriteTo(WriterInterceptorContext context) throws IOException {
+ context.getHeaders().add(WRITER_THREAD_HEADER, Thread.currentThread().getName());
+ context.proceed();
+ }
+ }
+
+ @Test
+ public void testResumeFromOtherThreadWritesOnWebServiceExecutor() throws Exception {
+ URI uri = URI.create("http://localhost:" + webService.getListenPortHTTP().orElseThrow()
+ + "/test/resumeFromOtherThread");
+ HttpResponse<String> response = httpClient.send(HttpRequest.newBuilder(uri).GET().build(),
+ HttpResponse.BodyHandlers.ofString());
+ assertThat(response.statusCode()).isEqualTo(200);
+ assertThat(response.body()).isEqualTo("resumed");
+ assertThat(response.headers().firstValue(WRITER_THREAD_HEADER)).hasValueSatisfying(
+ threadName -> assertThat(threadName).startsWith("pulsar-web"));
+ }
+}