From f1f4d74ddaf112e4df755dc6f1f4a57ee048190a Mon Sep 17 00:00:00 2001 From: croway Date: Fri, 18 Sep 2026 10:43:23 +0200 Subject: [PATCH] CAMEL-24811: camel-servlet - fix async servlet completing before the route finishes CamelServlet.doServiceAsync() (async=true without executorRef) completed the AsyncContext right after doService() returned, discarding the CompletionStage that doExecute() returns when the route continues on another thread. This races the container's async completion against the later writeResponse(), giving an IllegalStateException on the recycled response (Tomcat) or an empty/stale response. camel-jetty in async mode (async=true&useContinuation=false) hits the same path via super.doService(). doService() keeps its void signature (protected extension point, must stay binary compatible for backports); the in-flight stage is handed over via the CamelAsyncPromise request attribute and doServiceAsync() completes the AsyncContext only once it is done, mirroring doAsyncExecution(). Also: complete the AsyncContext in doAsyncExecution()'s error branch (onError always throws), and always write the response / finish the UoW when processAsync() fails synchronously or completes exceptionally. Co-Authored-By: Claude Sonnet 5 Co-Authored-By: Claude Fable 5.1 --- .../camel/http/common/CamelServlet.java | 58 ++++++++--- .../jetty/JettyAsyncDelayedRouteTest.java | 48 +++++++++ .../servlet/ServletAsyncErrorTest.java | 98 +++++++++++++++++++ .../ServletAsyncNoExecutorRefRaceTest.java | 56 +++++++++++ .../servlet/example-camelContext-race.xml | 44 +++++++++ 5 files changed, 289 insertions(+), 15 deletions(-) create mode 100644 components/camel-jetty/src/test/java/org/apache/camel/component/jetty/JettyAsyncDelayedRouteTest.java create mode 100644 components/camel-servlet/src/test/java/org/apache/camel/component/servlet/ServletAsyncErrorTest.java create mode 100644 components/camel-servlet/src/test/java/org/apache/camel/component/servlet/ServletAsyncNoExecutorRefRaceTest.java create mode 100644 components/camel-servlet/src/test/resources/org/apache/camel/component/servlet/example-camelContext-race.xml diff --git a/components/camel-http-common/src/main/java/org/apache/camel/http/common/CamelServlet.java b/components/camel-http-common/src/main/java/org/apache/camel/http/common/CamelServlet.java index db554a350ce99..2f7386a9817b5 100644 --- a/components/camel-http-common/src/main/java/org/apache/camel/http/common/CamelServlet.java +++ b/components/camel-http-common/src/main/java/org/apache/camel/http/common/CamelServlet.java @@ -59,6 +59,12 @@ public class CamelServlet extends HttpServlet implements HttpRegistryProvider { public static final String EXECUTOR_REF_PARAM = "executorRef"; public static final List METHODS = Arrays.asList("GET", "HEAD", "POST", "PUT", "DELETE", "TRACE", "OPTIONS", "CONNECT", "PATCH"); + /** + * Request attribute holding the {@link CompletionStage} of a request still being processed on another thread when + * {@link #doService(HttpServletRequest, HttpServletResponse)} returns. The {@link AsyncContext} must not be + * completed before it. + */ + protected static final String ASYNC_PROMISE_ATTRIBUTE_NAME = "CamelAsyncPromise"; private static final long serialVersionUID = -7061982839117697829L; @@ -141,14 +147,25 @@ private void doAsyncExecution( HttpServletRequest req, HttpServletResponse resp, HttpConsumer consumer, AsyncContext context) { try { final CompletionStage promise = doExecute(req, resp, consumer); - if (promise == null) { // early quit + completeOnCompletion(context, promise); + } catch (Exception e) { + try { + onError(resp, e); + } finally { context.complete(); - } else { - promise.whenComplete((r, e) -> context.complete()); } - } catch (Exception e) { - onError(resp, e); + } + } + + /** + * Completes the {@link AsyncContext} once the promise is done, or immediately if the request was handled + * synchronously (no promise). + */ + private static void completeOnCompletion(AsyncContext context, CompletionStage promise) { + if (promise == null) { context.complete(); + } else { + promise.whenComplete((r, e) -> context.complete()); } } @@ -213,11 +230,16 @@ protected void doServiceAsync(AsyncContext context) { final HttpServletResponse response = (HttpServletResponse) context.getResponse(); try { doService(request, response); + // doService is void (overridden by subclasses) so in-flight processing is handed over via the request + final CompletionStage promise = (CompletionStage) request.getAttribute(ASYNC_PROMISE_ATTRIBUTE_NAME); + completeOnCompletion(context, promise); } catch (Exception e) { //An error shouldn't occur as we should handle most of error in doService - onError(response, e); - } finally { - context.complete(); + try { + onError(response, e); + } finally { + context.complete(); + } } } @@ -232,7 +254,11 @@ protected void doService(HttpServletRequest request, HttpServletResponse respons log.trace("Service: {}", request); HttpConsumer consumer = doResolve(request, response); if (consumer != null) { - doExecute(request, response, consumer); + CompletionStage promise = doExecute(request, response, consumer); + if (promise != null) { + // still in-flight on another thread, which will write the response + request.setAttribute(ASYNC_PROMISE_ATTRIBUTE_NAME, promise); + } } } @@ -314,6 +340,8 @@ private CompletionStage doExecute(HttpServletRequest req, HttpServletResponse } } catch (Exception e) { exchange.setException(e); + // processAsync failed synchronously so write the response here + isAsync = false; } try { @@ -334,12 +362,12 @@ private CompletionStage tryAsyncProcess( .whenComplete((r, ex) -> { if (ex != null) { exchange.setException(ex); - } else { - try { - afterProcess(res, consumer, exchange, false); - } catch (Exception e) { - exchange.setException(e); - } + } + // always write the response (error or not) and finish the UoW + try { + afterProcess(res, consumer, exchange, false); + } catch (Exception e) { + exchange.setException(e); } }); return result; diff --git a/components/camel-jetty/src/test/java/org/apache/camel/component/jetty/JettyAsyncDelayedRouteTest.java b/components/camel-jetty/src/test/java/org/apache/camel/component/jetty/JettyAsyncDelayedRouteTest.java new file mode 100644 index 0000000000000..3d56c3c1a35a8 --- /dev/null +++ b/components/camel-jetty/src/test/java/org/apache/camel/component/jetty/JettyAsyncDelayedRouteTest.java @@ -0,0 +1,48 @@ +/* + * 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.camel.component.jetty; + +import org.apache.camel.builder.RouteBuilder; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +/** + * async=true must wait for a route step that resumes on another thread before completing the request. + */ +public class JettyAsyncDelayedRouteTest extends BaseJettyTest { + + @Test + public void testAsyncRouteCompletesBeforeResponse() { + String body = template.requestBody("http://localhost:{{port}}/racy", "hello", String.class); + assertEquals("delayed-response", body); + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + public void configure() { + // async and continuation is not compatible! + from("jetty:http://localhost:{{port}}/racy?async=true&useContinuation=false") + // resumes on another thread + .delay(300).asyncDelayed().end() + .transform().constant("delayed-response"); + } + }; + } + +} diff --git a/components/camel-servlet/src/test/java/org/apache/camel/component/servlet/ServletAsyncErrorTest.java b/components/camel-servlet/src/test/java/org/apache/camel/component/servlet/ServletAsyncErrorTest.java new file mode 100644 index 0000000000000..d72d807d6f745 --- /dev/null +++ b/components/camel-servlet/src/test/java/org/apache/camel/component/servlet/ServletAsyncErrorTest.java @@ -0,0 +1,98 @@ +/* + * 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.camel.component.servlet; + +import java.util.concurrent.CompletableFuture; + +import io.undertow.servlet.Servlets; +import io.undertow.servlet.api.DeploymentInfo; +import org.apache.camel.AsyncCallback; +import org.apache.camel.Exchange; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.support.AsyncProcessorSupport; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +/** + * async=true must still write an error response when the route fails after resuming on another thread, or when the + * processor's stage completes exceptionally. + */ +public class ServletAsyncErrorTest extends ServletCamelRouterTestSupport { + + @Test + public void testAsyncRouteThrows() throws Exception { + WebResponse response = query(new GetMethodWebRequest(contextUrl + "/services/async-error"), false); + + assertEquals(500, response.getResponseCode()); + } + + @Test + public void testAsyncStageCompletesExceptionally() throws Exception { + // a route processor never completes the stage exceptionally (the exception lands on the exchange), so + // plug a consumer with a custom AsyncProcessor directly + ServletEndpoint endpoint = context.getEndpoint("servlet:///async-failed", ServletEndpoint.class); + ServletConsumer consumer = (ServletConsumer) endpoint.createConsumer(new AsyncProcessorSupport() { + @Override + public boolean process(Exchange exchange, AsyncCallback callback) { + callback.done(true); + return true; + } + + @Override + public CompletableFuture processAsync(Exchange exchange) { + return CompletableFuture.failedFuture(new IllegalStateException("stage failed")); + } + }); + consumer.start(); + try { + WebResponse response = query(new GetMethodWebRequest(contextUrl + "/services/async-failed"), false); + + assertEquals(500, response.getResponseCode()); + } finally { + consumer.stop(); + } + } + + @Override + protected DeploymentInfo getDeploymentInfo() { + return Servlets.deployment() + .setClassLoader(getClass().getClassLoader()) + .setContextPath(CONTEXT) + .setDeploymentName(getClass().getName()) + .addServlet(Servlets.servlet("CamelServlet", CamelHttpTransportServlet.class) + .addInitParam("async", "true") + .setAsyncSupported(true) + .addMapping("/services/*")); + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + from("servlet:///async-error") + // resumes on another thread, then fails + .delay(100).asyncDelayed().end() + .process(e -> { + throw new IllegalStateException("boom"); + }); + } + }; + } + +} diff --git a/components/camel-servlet/src/test/java/org/apache/camel/component/servlet/ServletAsyncNoExecutorRefRaceTest.java b/components/camel-servlet/src/test/java/org/apache/camel/component/servlet/ServletAsyncNoExecutorRefRaceTest.java new file mode 100644 index 0000000000000..c3cbaa984a800 --- /dev/null +++ b/components/camel-servlet/src/test/java/org/apache/camel/component/servlet/ServletAsyncNoExecutorRefRaceTest.java @@ -0,0 +1,56 @@ +/* + * 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.camel.component.servlet; + +import io.undertow.servlet.Servlets; +import io.undertow.servlet.api.DeploymentInfo; +import org.junit.jupiter.api.Test; +import org.springframework.web.context.ContextLoaderListener; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +/** + * async=true without executorRef must wait for a route step that resumes on another thread before completing. + */ +public class ServletAsyncNoExecutorRefRaceTest extends ServletCamelRouterTestSupport { + + @Test + public void testAsyncRouteCompletesBeforeResponse() throws Exception { + WebRequest req = new GetMethodWebRequest(contextUrl + "/services/racy"); + WebResponse response = query(req, false); + + assertEquals(200, response.getResponseCode()); + assertEquals("delayed-response", response.getText()); + } + + @Override + protected DeploymentInfo getDeploymentInfo() { + return Servlets.deployment() + .setClassLoader(getClass().getClassLoader()) + .setContextPath(CONTEXT) + .setDeploymentName(getClass().getName()) + .addInitParameter("contextConfigLocation", + "classpath:org/apache/camel/component/servlet/example-camelContext-race.xml") + .addListener(Servlets.listener(ContextLoaderListener.class)) + .addServlet(Servlets.servlet("CamelServlet", CamelHttpTransportServlet.class) + .addInitParam("async", "true") + .setLoadOnStartup(1) + .setAsyncSupported(true) + .addMapping("/services/*")); + } + +} diff --git a/components/camel-servlet/src/test/resources/org/apache/camel/component/servlet/example-camelContext-race.xml b/components/camel-servlet/src/test/resources/org/apache/camel/component/servlet/example-camelContext-race.xml new file mode 100644 index 0000000000000..a9436e99ae1aa --- /dev/null +++ b/components/camel-servlet/src/test/resources/org/apache/camel/component/servlet/example-camelContext-race.xml @@ -0,0 +1,44 @@ + + + + + + + + + + + 300 + + + delayed-response + + + + +