CAMEL-24811: camel-servlet - fix async servlet completing before the route finishes (backport 4.22.x)

components/camel-http-common/src/main/java/org/apache/camel/http/common/CamelServlet.java

@@ -61,6 +61,12 @@ public class CamelServlet extends HttpServlet implements HttpRegistryProvider {
     public static final String EXECUTOR_REF_PARAM = "executorRef";
     public static final List<String> 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;
 
@@ -143,14 +149,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());
         }
     }
 
@@ -215,11 +232,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();
+            }
         }
     }
 
@@ -234,7 +256,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);
+            }
         }
     }
 
@@ -325,6 +351,8 @@ private CompletionStage<?> doExecute(HttpServletRequest req, HttpServletResponse
             }
         } catch (Exception e) {
             exchange.setException(e);
+            // processAsync failed synchronously so write the response here
+            isAsync = false;
         }
 
         try {
@@ -373,12 +401,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;

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");
+            }
+        };
+    }
+
+}

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<Exchange> 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");
+                        });
+            }
+        };
+    }
+
+}

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/*"));
+    }
+
+}

components/camel-servlet/src/test/resources/org/apache/camel/component/servlet/example-camelContext-race.xml

@@ -0,0 +1,44 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+    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.
+
+-->
+<beans xmlns="http://www.springframework.org/schema/beans"
+       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+       xmlns:camel="http://camel.apache.org/schema/spring"
+       xsi:schemaLocation="
+       http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+       http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd
+    ">
+
+   <camelContext id="camel" streamCache="true" xmlns="http://camel.apache.org/schema/spring" >
+    <route id="raceRoute">
+      <!-- incoming requests from the servlet is routed -->
+      <from uri="servlet:racy"/>
+      <!-- genuinely asynchronous: resumes on the delayer's scheduled-executor thread,
+           simulating a downstream call (e.g. CXF SOAP client) that completes on a
+           background thread instead of the calling thread -->
+      <delay asyncDelayed="true">
+        <constant>300</constant>
+      </delay>
+      <transform>
+        <simple>delayed-response</simple>
+      </transform>
+    </route>
+   </camelContext>
+
+</beans>