You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2026/07/20 08:47:41 UTC
(camel) branch fix/CAMEL-24179 created (now 521a45e7c746)
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a change to branch fix/CAMEL-24179
in repository https://gitbox.apache.org/repos/asf/camel.git
at 521a45e7c746 CAMEL-24179: camel-cxf - Fix continuation timeout race with async processing
This branch includes the following new commits:
new 521a45e7c746 CAMEL-24179: camel-cxf - Fix continuation timeout race with async processing
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(camel) 01/01: CAMEL-24179: camel-cxf - Fix continuation timeout race with async processing
Posted by da...@apache.org.
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 <cl...@gmail.com>
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 <no...@anthropic.com>
Signed-off-by: Claus Ibsen <cl...@gmail.com>
---
.../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);
+ }
+
}