This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch fix/CAMEL-24179 in repository https://gitbox.apache.org/repos/asf/camel.git
commit 521a45e7c7462467fd42213b84c7fc1b4590a5b2 Author: Claus Ibsen <[email protected]> AuthorDate: Mon Jul 20 10:47:25 2026 +0200 CAMEL-24179: camel-cxf - Fix continuation timeout race with async processing Use an AtomicBoolean to ensure exactly one of {timeout, async completion} owns the response and UoW lifecycle. When timeout wins the race, the late async callback closes the UoW after the worker finishes. When the async callback wins, the timeout branch is skipped. This prevents double continuation.resume(), concurrent Exchange mutation, and double UoW close that could corrupt pooled exchanges. Co-Authored-By: Claude Opus 4.6 <[email protected]> Signed-off-by: Claus Ibsen <[email protected]> --- .../camel/component/cxf/jaxrs/CxfRsInvoker.java | 44 +++++++++++++++------ .../camel/component/cxf/jaxws/CxfConsumer.java | 46 ++++++++++++++++------ .../jaxws/CxfConsumerContinuationTimeoutTest.java | 11 ++++++ 3 files changed, 76 insertions(+), 25 deletions(-) diff --git a/components/camel-cxf/camel-cxf-rest/src/main/java/org/apache/camel/component/cxf/jaxrs/CxfRsInvoker.java b/components/camel-cxf/camel-cxf-rest/src/main/java/org/apache/camel/component/cxf/jaxrs/CxfRsInvoker.java index dec381a8fcf3..daa83b583355 100644 --- a/components/camel-cxf/camel-cxf-rest/src/main/java/org/apache/camel/component/cxf/jaxrs/CxfRsInvoker.java +++ b/components/camel-cxf/camel-cxf-rest/src/main/java/org/apache/camel/component/cxf/jaxrs/CxfRsInvoker.java @@ -17,6 +17,7 @@ package org.apache.camel.component.cxf.jaxrs; import java.lang.reflect.Method; +import java.util.concurrent.atomic.AtomicBoolean; import jakarta.ws.rs.WebApplicationException; import jakarta.ws.rs.core.HttpHeaders; @@ -43,6 +44,7 @@ import org.slf4j.LoggerFactory; public class CxfRsInvoker extends JAXRSInvoker { private static final Logger LOG = LoggerFactory.getLogger(CxfRsInvoker.class); private static final String SUSPENDED = "org.apache.camel.component.cxf.jaxrs.suspend"; + private static final String COMPLETED = "org.apache.camel.component.cxf.jaxrs.completed"; private final CxfRsConsumer cxfRsConsumer; private final CxfRsEndpoint endpoint; @@ -91,8 +93,10 @@ public class CxfRsInvoker extends JAXRSInvoker { final org.apache.camel.Exchange camelExchange = prepareExchange(cxfExchange, method, paramArray, response); // we want to handle the UoW cxfRsConsumer.createUoW(camelExchange); - // Now we don't set up the timeout value LOG.trace("Suspending continuation of exchangeId: {}", camelExchange.getExchangeId()); + // guard so exactly one of {timeout, async completion} owns the response and UoW lifecycle + final AtomicBoolean completed = new AtomicBoolean(false); + cxfExchange.put(COMPLETED, completed); // The continuation could be called before the suspend is called continuation.suspend(endpoint.getContinuationTimeout()); cxfExchange.put(SUSPENDED, Boolean.TRUE); @@ -101,9 +105,16 @@ public class CxfRsInvoker extends JAXRSInvoker { public void done(boolean doneSync) { // make sure the continuation resume will not be called before the suspend method in other thread synchronized (continuation) { - LOG.trace("Resuming continuation of exchangeId: {}", camelExchange.getExchangeId()); - // resume processing after both, sync and async callbacks - continuation.resume(); + if (completed.compareAndSet(false, true)) { + LOG.trace("Resuming continuation of exchangeId: {}", camelExchange.getExchangeId()); + // resume processing after both, sync and async callbacks + continuation.resume(); + } else { + // timeout already sent the response; close the UoW now that the worker has finished + LOG.trace("Timeout already handled response for exchangeId: {}; closing UoW", + camelExchange.getExchangeId()); + cxfRsConsumer.doneUoW(camelExchange); + } } } }); @@ -120,14 +131,23 @@ public class CxfRsInvoker extends JAXRSInvoker { } } else { if (continuation.isTimeout() || !continuation.isPending()) { - cxfExchange.put(SUSPENDED, Boolean.FALSE); - org.apache.camel.Exchange camelExchange = (org.apache.camel.Exchange) continuation.getObject(); - camelExchange.setException(new ExchangeTimedOutException(camelExchange, endpoint.getContinuationTimeout())); - try { - return returnResponse(cxfExchange, camelExchange); - } catch (Exception ex) { - cxfRsConsumer.doneUoW(camelExchange); - throw ex; + AtomicBoolean completed = (AtomicBoolean) cxfExchange.get(COMPLETED); + if (completed != null && completed.compareAndSet(false, true)) { + cxfExchange.put(SUSPENDED, Boolean.FALSE); + org.apache.camel.Exchange camelExchange = (org.apache.camel.Exchange) continuation.getObject(); + camelExchange + .setException(new ExchangeTimedOutException(camelExchange, endpoint.getContinuationTimeout())); + try { + Object result = returnResponse(cxfExchange, camelExchange); + // detach so UnitOfWorkCloserInterceptor won't close UoW; + // the late async callback will close it after the worker finishes + cxfExchange.put(org.apache.camel.Exchange.class, null); + return result; + } catch (Exception ex) { + cxfExchange.put(org.apache.camel.Exchange.class, null); + cxfRsConsumer.doneUoW(camelExchange); + throw ex; + } } } } diff --git a/components/camel-cxf/camel-cxf-soap/src/main/java/org/apache/camel/component/cxf/jaxws/CxfConsumer.java b/components/camel-cxf/camel-cxf-soap/src/main/java/org/apache/camel/component/cxf/jaxws/CxfConsumer.java index 73a43095ba4b..16b63a98c308 100644 --- a/components/camel-cxf/camel-cxf-soap/src/main/java/org/apache/camel/component/cxf/jaxws/CxfConsumer.java +++ b/components/camel-cxf/camel-cxf-soap/src/main/java/org/apache/camel/component/cxf/jaxws/CxfConsumer.java @@ -20,6 +20,7 @@ import java.lang.reflect.Method; import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.concurrent.atomic.AtomicBoolean; import jakarta.xml.ws.WebFault; @@ -151,6 +152,8 @@ public class CxfConsumer extends DefaultConsumer implements Suspendable { private class CxfConsumerInvoker implements Invoker { + private static final String COMPLETED = "org.apache.camel.component.cxf.jaxws.completed"; + private final CxfEndpoint endpoint; CxfConsumerInvoker(CxfEndpoint endpoint) { @@ -180,9 +183,12 @@ public class CxfConsumer extends DefaultConsumer implements Suspendable { if (continuation.isNew()) { final org.apache.camel.Exchange camelExchange = prepareCamelExchange(cxfExchange); - // Now we don't set up the timeout value LOG.trace("Suspending continuation of exchangeId: {}", camelExchange.getExchangeId()); + // guard so exactly one of {timeout, async completion} owns the response and UoW lifecycle + final AtomicBoolean completed = new AtomicBoolean(false); + cxfExchange.put(COMPLETED, completed); + // The continuation could be called before the suspend is called continuation.suspend(cxfEndpoint.getContinuationTimeout()); @@ -193,9 +199,16 @@ public class CxfConsumer extends DefaultConsumer implements Suspendable { public void done(boolean doneSync) { // make sure the continuation resume will not be called before the suspend method in other thread synchronized (continuation) { - LOG.trace("Resuming continuation of exchangeId: {}", camelExchange.getExchangeId()); - // resume processing after both, sync and async callbacks - continuation.resume(); + if (completed.compareAndSet(false, true)) { + LOG.trace("Resuming continuation of exchangeId: {}", camelExchange.getExchangeId()); + // resume processing after both, sync and async callbacks + continuation.resume(); + } else { + // timeout already sent the response; close the UoW now that the worker has finished + LOG.trace("Timeout already handled response for exchangeId: {}; closing UoW", + camelExchange.getExchangeId()); + CxfConsumer.this.doneUoW(camelExchange); + } } } }); @@ -210,16 +223,23 @@ public class CxfConsumer extends DefaultConsumer implements Suspendable { } } else if (continuation.isTimeout() || !continuation.isResumed() && !continuation.isPending()) { - org.apache.camel.Exchange camelExchange = (org.apache.camel.Exchange) continuation.getObject(); - try { - if (!continuation.isPending()) { - camelExchange.setException( - new ExchangeTimedOutException(camelExchange, cxfEndpoint.getContinuationTimeout())); + AtomicBoolean completed = (AtomicBoolean) cxfExchange.get(COMPLETED); + if (completed != null && completed.compareAndSet(false, true)) { + org.apache.camel.Exchange camelExchange = (org.apache.camel.Exchange) continuation.getObject(); + try { + if (!continuation.isPending()) { + camelExchange.setException( + new ExchangeTimedOutException(camelExchange, cxfEndpoint.getContinuationTimeout())); + } + setResponseBack(cxfExchange, camelExchange); + // detach so UnitOfWorkCloserInterceptor won't close UoW; + // the late async callback will close it after the worker finishes + cxfExchange.put(org.apache.camel.Exchange.class, null); + } catch (Exception ex) { + cxfExchange.put(org.apache.camel.Exchange.class, null); + CxfConsumer.this.doneUoW(camelExchange); + throw ex; } - setResponseBack(cxfExchange, camelExchange); - } catch (Exception ex) { - CxfConsumer.this.doneUoW(camelExchange); - throw ex; } } } diff --git a/components/camel-cxf/camel-cxf-soap/src/test/java/org/apache/camel/component/cxf/jaxws/CxfConsumerContinuationTimeoutTest.java b/components/camel-cxf/camel-cxf-soap/src/test/java/org/apache/camel/component/cxf/jaxws/CxfConsumerContinuationTimeoutTest.java index 5e53e5a8792d..da7c70a50440 100644 --- a/components/camel-cxf/camel-cxf-soap/src/test/java/org/apache/camel/component/cxf/jaxws/CxfConsumerContinuationTimeoutTest.java +++ b/components/camel-cxf/camel-cxf-soap/src/test/java/org/apache/camel/component/cxf/jaxws/CxfConsumerContinuationTimeoutTest.java @@ -114,4 +114,15 @@ public class CxfConsumerContinuationTimeoutTest extends CamelTestSupport { assertTrue(out.contains("The OUT message was not received within: 5000 millis.")); } + @Test + public void testTimeoutThenNormalRequest() throws Exception { + // first request times out + String out = template.requestBodyAndHeader("direct:start", "Bye World", "priority", "slow", String.class); + assertTrue(out.contains("The OUT message was not received within: 5000 millis.")); + + // subsequent normal request must succeed without state corruption + Object out2 = template.requestBody("direct:start", "Hello World", String.class); + assertEquals(ECHO_BOOLEAN_RESPONSE, out2); + } + }
