This is an automated email from the ASF dual-hosted git repository.
gianm pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new 79fb6155b4f fix: Prevent an AsyncRequest timeout race from wedging
HTTP connections (#20821)
79fb6155b4f is described below
commit 79fb6155b4f453a5b486c72deb8fdcf073af55e5
Author: Yahya Kisana <[email protected]>
AuthorDate: Tue Nov 3 12:35:17 2026 -0600
fix: Prevent an AsyncRequest timeout race from wedging HTTP connections
(#20821)
When the future of a long-polling AsyncRequest completed on another thread
while Jetty was timing the request out, the future's thread won the
at-most-once guard and wrote the response, but Jetty rejected its
AsyncContext.complete(), and the timeout listener returned without
completing the request either. Jetty then never finished the exchange, so
the next request sent on that keep-alive connection hung until the client's
read timeout. This made the Overlord's assignTask requests to an Indexer
hang for 60s, which made KafkaMultiSupervisorTest flaky.
Always complete the request from the timeout listener, which is the only
thread Jetty allows to complete a request that is timing out, and make
AsyncRequest.complete() synchronized so that the listener waits for a
response that is still being written.
---
.../org/apache/druid/server/http/AsyncRequest.java | 27 ++-
.../druid/server/http/AsyncRequestJettyTest.java | 255 +++++++++++++++++++++
.../apache/druid/server/http/AsyncRequestTest.java | 106 ++++++++-
3 files changed, 381 insertions(+), 7 deletions(-)
diff --git
a/server/src/main/java/org/apache/druid/server/http/AsyncRequest.java
b/server/src/main/java/org/apache/druid/server/http/AsyncRequest.java
index 04fd558278a..793d17e8fe1 100644
--- a/server/src/main/java/org/apache/druid/server/http/AsyncRequest.java
+++ b/server/src/main/java/org/apache/druid/server/http/AsyncRequest.java
@@ -28,6 +28,7 @@ import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.MoreExecutors;
+import com.google.errorprone.annotations.concurrent.GuardedBy;
import org.apache.druid.java.util.common.logger.Logger;
import javax.annotation.Nullable;
@@ -36,7 +37,6 @@ import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import javax.ws.rs.core.MediaType;
import java.lang.reflect.Type;
-import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Function;
import java.util.function.Supplier;
@@ -51,11 +51,13 @@ public class AsyncRequest
{
private static final Logger log = new Logger(AsyncRequest.class);
- private final AtomicBoolean completed = new AtomicBoolean(false);
private final AsyncContext context;
private final ObjectMapper objectMapper;
private final String contentType;
+ @GuardedBy("this")
+ private boolean completed = false;
+
/**
* Starts async processing of the request and immediately completes it with
the given response, writing any
* response entity as JSON using the given {@link ObjectMapper}.
@@ -96,13 +98,15 @@ public class AsyncRequest
/**
* Completes the request as described by the {@link AsyncResponse} obtained
from the given supplier, if it
* hasn't already been completed. The supplier is invoked only if the
request is not already complete.
+ * Synchronized so that a call made while another thread is writing the
response waits for that write to finish.
*/
- private <T> void complete(Supplier<AsyncResponse<T>> responseSupplier)
+ private synchronized <T> void complete(Supplier<AsyncResponse<T>>
responseSupplier)
{
- if (!completed.compareAndSet(false, true)) {
+ if (completed) {
// Request has already been completed
return;
}
+ completed = true;
try {
final AsyncResponse<T> response = responseSupplier.get();
@@ -242,6 +246,7 @@ public class AsyncRequest
/**
* Function invoked with the result of the future on success, to determine
how to respond. Required.
+ * Invoked while holding the lock that the timeout listener waits on, so
it must not block.
*/
public Builder<T> onSuccess(Function<T, AsyncResponse<?>> onSuccess)
{
@@ -251,6 +256,7 @@ public class AsyncRequest
/**
* Function invoked with the failure cause of the future, to determine how
to respond. Required.
+ * Invoked while holding the lock that the timeout listener waits on, so
it must not block.
*/
public Builder<T> onFailure(Function<Throwable, AsyncResponse<?>>
onFailure)
{
@@ -274,8 +280,17 @@ public class AsyncRequest
asyncRequest.context.setTimeout(timeoutMillis);
asyncRequest.context.addListener(
ServletResourceUtils.createAsyncTimeoutListener(event -> {
- asyncRequest.complete(() ->
AsyncResponse.status(onTimeoutStatusCode));
- future.cancel(true);
+ try {
+ asyncRequest.complete(() ->
AsyncResponse.status(onTimeoutStatusCode));
+ future.cancel(true);
+ }
+ finally {
+ // While a request is timing out, the container only allows it
to be completed from this thread, so if
+ // the future won the race to respond, its call to complete()
was rejected. Complete the request here,
+ // otherwise the connection is never reused. Since complete() is
synchronized, any response being written
+ // by the future's thread has been fully written by now. No-op
if the request is already complete.
+ event.getAsyncContext().complete();
+ }
})
);
diff --git
a/server/src/test/java/org/apache/druid/server/http/AsyncRequestJettyTest.java
b/server/src/test/java/org/apache/druid/server/http/AsyncRequestJettyTest.java
new file mode 100644
index 00000000000..724bbac2839
--- /dev/null
+++
b/server/src/test/java/org/apache/druid/server/http/AsyncRequestJettyTest.java
@@ -0,0 +1,255 @@
+/*
+ * 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.druid.server.http;
+
+import com.google.common.util.concurrent.SettableFuture;
+import io.netty.handler.codec.http.HttpMethod;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import org.apache.druid.common.utils.SocketUtil;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.lifecycle.Lifecycle;
+import org.apache.druid.java.util.http.client.HttpClient;
+import org.apache.druid.java.util.http.client.HttpClientConfig;
+import org.apache.druid.java.util.http.client.HttpClientInit;
+import org.apache.druid.java.util.http.client.Request;
+import org.apache.druid.java.util.http.client.response.StatusResponseHandler;
+import org.apache.druid.java.util.http.client.response.StatusResponseHolder;
+import org.apache.druid.segment.TestHelper;
+import org.apache.druid.server.http.AsyncRequest.AsyncResponse;
+import org.eclipse.jetty.ee8.servlet.ServletContextHandler;
+import org.eclipse.jetty.ee8.servlet.ServletHolder;
+import org.eclipse.jetty.server.Server;
+import org.joda.time.Duration;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import javax.servlet.AsyncContext;
+import javax.servlet.AsyncEvent;
+import javax.servlet.AsyncListener;
+import javax.servlet.http.HttpServlet;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletRequestWrapper;
+import javax.servlet.http.HttpServletResponse;
+import javax.ws.rs.core.MediaType;
+import java.io.IOException;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Proxy;
+import java.net.URI;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Tests {@link AsyncRequest} against a real Jetty server.
+ */
+public class AsyncRequestJettyTest
+{
+ private static final Duration READ_TIMEOUT = Duration.standardSeconds(5);
+
+ /**
+ * The future completes and wins the race to respond, but writes its
response only after the request has started
+ * timing out. Jetty then rejects the writing thread's call to complete the
request, so unless the timeout listener
+ * completes it, the connection is never read from again and the next
request sent on it hangs.
+ */
+ @Test
+ public void
test_connectionIsReusable_whenFutureRespondsWhileRequestIsTimingOut() throws
Exception
+ {
+ final CountDownLatch futureWonRace = new CountDownLatch(1);
+ final CountDownLatch timeoutStarted = new CountDownLatch(1);
+ final CountDownLatch responseWritten = new CountDownLatch(1);
+
+ final int port = SocketUtil.findOpenPort(0);
+ final Server server = new Server(port);
+ final ServletContextHandler handler = new ServletContextHandler();
+ final ServletHolder holder = new ServletHolder(new HttpServlet()
+ {
+ @Override
+ protected void doGet(HttpServletRequest req, HttpServletResponse resp)
throws IOException
+ {
+ if (!"/race".equals(req.getPathInfo())) {
+ resp.setStatus(HttpServletResponse.SC_OK);
+ resp.getWriter().write("pong");
+ return;
+ }
+
+ final SettableFuture<String> future = SettableFuture.create();
+ AsyncRequest.create(new TimeoutHookingRequest(req, timeoutStarted,
responseWritten), future, 100L)
+ .contentType(TestHelper.JSON_MAPPER,
MediaType.APPLICATION_JSON)
+ .onTimeout(HttpServletResponse.SC_NO_CONTENT)
+ .onSuccess(
+ result -> {
+ futureWonRace.countDown();
+ // Write the response only once the request has
started timing out
+ await(timeoutStarted);
+ return AsyncResponse.of(HttpServletResponse.SC_OK,
result);
+ }
+ )
+ .onFailure(t ->
AsyncResponse.error(HttpServletResponse.SC_INTERNAL_SERVER_ERROR,
t.getMessage()))
+ .start();
+
+ // Complete the future from another thread, as ChangeRequestHistory
does for long-polling sync requests
+ new Thread(() -> {
+ future.set("the-result");
+ responseWritten.countDown();
+ }).start();
+
+ // Jetty schedules the timeout only after this method returns, so this
ensures the future wins the race
+ await(futureWonRace);
+ }
+ });
+ holder.setAsyncSupported(true);
+ handler.addServlet(holder, "/*");
+ server.setHandler(handler);
+
+ final Lifecycle lifecycle = new Lifecycle();
+ try {
+ server.start();
+ lifecycle.start();
+
+ // A single connection, so that the second request is sent on the same
connection as the first
+ final HttpClient client = HttpClientInit.createClient(
+ HttpClientConfig.builder().withNumConnections(1).build(),
+ lifecycle
+ );
+
+ final StatusResponseHolder first = client.go(
+ new Request(HttpMethod.GET,
URI.create(StringUtils.format("http://localhost:%d/race";, port)).toURL()),
+ StatusResponseHandler.getInstance(),
+ READ_TIMEOUT
+ ).get();
+ Assertions.assertEquals(HttpResponseStatus.OK, first.getStatus());
+ Assertions.assertEquals("\"the-result\"", first.getContent());
+
+ final StatusResponseHolder second = client.go(
+ new Request(HttpMethod.GET,
URI.create(StringUtils.format("http://localhost:%d/ping";, port)).toURL()),
+ StatusResponseHandler.getInstance(),
+ READ_TIMEOUT
+ ).get();
+ Assertions.assertEquals(HttpResponseStatus.OK, second.getStatus());
+ Assertions.assertEquals("pong", second.getContent());
+ }
+ finally {
+ lifecycle.stop();
+ server.stop();
+ }
+ }
+
+ /**
+ * Waits for the latch. The bound is only a safety net, since the latch is
released promptly whether or not the
+ * timeout listener completes the request.
+ */
+ private static void await(CountDownLatch latch)
+ {
+ try {
+ if (!latch.await(10, TimeUnit.SECONDS)) {
+ throw new IllegalStateException("Timed out waiting for latch");
+ }
+ }
+ catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ }
+ }
+
+ /**
+ * Request whose {@link AsyncContext} wraps every listener added to it,
which is how the timeout listener registered
+ * by {@link AsyncRequest} gets hooked. The servlet API offers no other way
to run code around that listener.
+ */
+ private static class TimeoutHookingRequest extends HttpServletRequestWrapper
+ {
+ private final CountDownLatch timeoutStarted;
+ private final CountDownLatch responseWritten;
+
+ TimeoutHookingRequest(HttpServletRequest request, CountDownLatch
timeoutStarted, CountDownLatch responseWritten)
+ {
+ super(request);
+ this.timeoutStarted = timeoutStarted;
+ this.responseWritten = responseWritten;
+ }
+
+ @Override
+ public AsyncContext startAsync()
+ {
+ final AsyncContext delegate = super.startAsync();
+ return (AsyncContext) Proxy.newProxyInstance(
+ AsyncContext.class.getClassLoader(),
+ new Class<?>[]{AsyncContext.class},
+ (proxy, method, args) -> {
+ if ("addListener".equals(method.getName()) && args.length == 1) {
+ delegate.addListener(
+ new TimeoutHookingListener((AsyncListener) args[0],
timeoutStarted, responseWritten)
+ );
+ return null;
+ }
+ try {
+ return method.invoke(delegate, args);
+ }
+ catch (InvocationTargetException e) {
+ throw e.getCause();
+ }
+ }
+ );
+ }
+ }
+
+ /**
+ * Signals that the request has started timing out before running the
wrapped listener, and waits for the response
+ * to be written before returning control to Jetty. This forces the order of
events seen in production: the
+ * response is written while the request is timing out, before Jetty acts on
the outcome of the timeout listeners.
+ */
+ private static class TimeoutHookingListener implements AsyncListener
+ {
+ private final AsyncListener delegate;
+ private final CountDownLatch timeoutStarted;
+ private final CountDownLatch responseWritten;
+
+ TimeoutHookingListener(AsyncListener delegate, CountDownLatch
timeoutStarted, CountDownLatch responseWritten)
+ {
+ this.delegate = delegate;
+ this.timeoutStarted = timeoutStarted;
+ this.responseWritten = responseWritten;
+ }
+
+ @Override
+ public void onTimeout(AsyncEvent event) throws IOException
+ {
+ timeoutStarted.countDown();
+ delegate.onTimeout(event);
+ await(responseWritten);
+ }
+
+ @Override
+ public void onComplete(AsyncEvent event) throws IOException
+ {
+ delegate.onComplete(event);
+ }
+
+ @Override
+ public void onError(AsyncEvent event) throws IOException
+ {
+ delegate.onError(event);
+ }
+
+ @Override
+ public void onStartAsync(AsyncEvent event) throws IOException
+ {
+ delegate.onStartAsync(event);
+ }
+ }
+}
diff --git
a/server/src/test/java/org/apache/druid/server/http/AsyncRequestTest.java
b/server/src/test/java/org/apache/druid/server/http/AsyncRequestTest.java
index c5fd3afd317..bfb363c19f8 100644
--- a/server/src/test/java/org/apache/druid/server/http/AsyncRequestTest.java
+++ b/server/src/test/java/org/apache/druid/server/http/AsyncRequestTest.java
@@ -23,6 +23,7 @@ import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.SettableFuture;
+import com.google.common.util.concurrent.Uninterruptibles;
import org.apache.druid.segment.TestHelper;
import org.apache.druid.server.coordination.ChangeRequestHistory;
import org.apache.druid.server.coordination.ChangeRequestsSnapshot;
@@ -35,12 +36,16 @@ import
org.apache.druid.server.mocks.MockHttpServletResponse;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
import javax.servlet.AsyncEvent;
import javax.servlet.AsyncListener;
import javax.ws.rs.core.MediaType;
+import java.io.IOException;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
public class AsyncRequestTest
{
@@ -230,6 +235,94 @@ public class AsyncRequestTest
Assertions.assertEquals(0, response.baos.size());
}
+ @Test
+ public void
test_start_completesRequestOnTimeout_whenFutureRespondedButCouldNotComplete()
throws Exception
+ {
+ final SettableFuture<String> future = SettableFuture.create();
+
+ AsyncRequest.create(request, future, 10L)
+ .contentType(JSON_MAPPER, MediaType.APPLICATION_JSON)
+ .onTimeout(204)
+ .onSuccess(result -> AsyncResponse.of(200, result))
+ .onFailure(t -> AsyncResponse.error(500, t.getMessage()))
+ .start();
+
+ // The future completes just as the request starts timing out. The
response is written, but the container
+ // rejects complete() since it is not called from the thread running the
timeout listener.
+ asyncContext.rejectComplete = true;
+ future.set("the-result");
+
+ Assertions.assertEquals(200, response.getStatus());
+ Assertions.assertFalse(asyncContext.isCompleted());
+
+ asyncContext.rejectComplete = false;
+ asyncContext.capturedListener.onTimeout(new AsyncEvent(asyncContext));
+
+ // The timeout listener must complete the request, without overwriting the
response that was already written.
+ Assertions.assertTrue(asyncContext.isCompleted());
+ Assertions.assertEquals(200, response.getStatus());
+ Assertions.assertEquals("the-result", readJson(new TypeReference<String>()
{}));
+ }
+
+ @Test
+ @Timeout(value = 60_000L, unit = TimeUnit.MILLISECONDS)
+ public void test_start_timeoutWaitsForResponseBeingWrittenByAnotherThread()
throws Exception
+ {
+ final SettableFuture<String> future = SettableFuture.create();
+ final CountDownLatch writeStarted = new CountDownLatch(1);
+ final CountDownLatch finishWrite = new CountDownLatch(1);
+
+ AsyncRequest.create(request, future, 10L)
+ .contentType(JSON_MAPPER, MediaType.APPLICATION_JSON)
+ .onTimeout(204)
+ .onSuccess(result -> {
+ writeStarted.countDown();
+ Uninterruptibles.awaitUninterruptibly(finishWrite);
+ return AsyncResponse.of(200, result);
+ })
+ .onFailure(t -> AsyncResponse.error(500, t.getMessage()))
+ .start();
+
+ final Thread writer = new Thread(() -> future.set("the-result"));
+
+ // Times out the request while the response is still being written by the
writer thread
+ final Thread timeout = new Thread(() -> {
+ try {
+ asyncContext.capturedListener.onTimeout(new AsyncEvent(asyncContext));
+ }
+ catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ });
+
+ try {
+ writer.start();
+ Assertions.assertTrue(writeStarted.await(10, TimeUnit.SECONDS));
+ timeout.start();
+
+ // The timeout listener must block until the write has finished, rather
than complete the request mid-write.
+ // Wait for it to either block or finish, to avoid relying on a sleep.
+ final long startNanos = System.nanoTime();
+ while (timeout.isAlive()
+ && timeout.getState() != Thread.State.BLOCKED
+ && System.nanoTime() - startNanos < TimeUnit.SECONDS.toNanos(10))
{
+ Thread.sleep(1);
+ }
+ Assertions.assertEquals(Thread.State.BLOCKED, timeout.getState());
+ Assertions.assertFalse(asyncContext.isCompleted());
+ }
+ finally {
+ finishWrite.countDown();
+ }
+
+ writer.join();
+ timeout.join();
+
+ Assertions.assertTrue(asyncContext.isCompleted());
+ Assertions.assertEquals(200, response.getStatus());
+ Assertions.assertEquals("the-result", readJson(new TypeReference<String>()
{}));
+ }
+
@Test
public void
test_onSuccess_preservesPolymorphicTypeInfo_whenResultTypeIsGenericAndErased()
{
@@ -271,16 +364,27 @@ public class AsyncRequestTest
/**
* A {@link MockAsyncContext} that captures the single listener added to it,
so tests can manually fire a
- * timeout event.
+ * timeout event. If {@link #rejectComplete} is set, {@link #complete()}
throws, like a servlet container does
+ * when a request that is timing out is completed from a thread other than
the one running the timeout listener.
*/
private static class CapturingAsyncContext extends MockAsyncContext
{
private AsyncListener capturedListener;
+ private boolean rejectComplete;
@Override
public void addListener(AsyncListener listener)
{
capturedListener = listener;
}
+
+ @Override
+ public void complete()
+ {
+ if (rejectComplete) {
+ throw new IllegalStateException("Request is timing out");
+ }
+ super.complete();
+ }
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
