This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new f29a90e6f159 CAMEL-26696: camel-cxf: async producer may never complete
the exchange when the connection fails (#29450)
f29a90e6f159 is described below
commit f29a90e6f15905b8a5a3e0d810dd17d9758c9ddc
Author: Guillaume Nodet <[email protected]>
AuthorDate: Tue Nov 10 14:47:40 2026 +0100
CAMEL-26696: camel-cxf: async producer may never complete the exchange when
the connection fails (#29450)
The camel-cxf async producer always signals the Camel async callback
exactly once, even when the cookie store or the response binding throws, so the
calling thread no longer hangs in DefaultAsyncProcessorAwaitManager. Binding
failures are kept as suppressed exceptions on the root cause, a done-guard
prevents double completion, and the ConduitSelector branch leaves the exchange
clean when failover already succeeded.
CXF-9272 (the hang inside CXF 4.2.4) still needs a CXF release; CAMEL-26696
stays open until then. testInvokingAWrongServerAsync is disabled pending
CXF-9272.
Hermes Agent (Claude Sonnet 4.6) on behalf of Guillaume Nodet
Co-authored-by: Claude Sonnet 4.6 <[email protected]>
---
.../component/cxf/jaxws/CxfClientCallback.java | 135 +++++--
.../component/cxf/jaxws/CxfClientCallbackTest.java | 396 +++++++++++++++++++++
.../camel/component/cxf/jaxws/CxfProducerTest.java | 28 +-
3 files changed, 523 insertions(+), 36 deletions(-)
diff --git
a/components/camel-cxf/camel-cxf-soap/src/main/java/org/apache/camel/component/cxf/jaxws/CxfClientCallback.java
b/components/camel-cxf/camel-cxf-soap/src/main/java/org/apache/camel/component/cxf/jaxws/CxfClientCallback.java
index ad655051c868..724d3d297107 100644
---
a/components/camel-cxf/camel-cxf-soap/src/main/java/org/apache/camel/component/cxf/jaxws/CxfClientCallback.java
+++
b/components/camel-cxf/camel-cxf-soap/src/main/java/org/apache/camel/component/cxf/jaxws/CxfClientCallback.java
@@ -16,9 +16,9 @@
*/
package org.apache.camel.component.cxf.jaxws;
-import java.io.IOException;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.camel.AsyncCallback;
import org.apache.camel.Exchange;
@@ -38,6 +38,9 @@ public class CxfClientCallback extends ClientCallback {
private final org.apache.cxf.message.Exchange cxfExchange;
private final BindingOperationInfo boi;
private final CxfEndpoint endpoint;
+ // Guard against done(false) being called more than once: CXF may call
handleException
+ // after handleResponse throws (e.g. ClientImpl.onMessage catches and
re-dispatches).
+ private final AtomicBoolean doneFired = new AtomicBoolean();
public CxfClientCallback(AsyncCallback callback,
Exchange camelExchange,
@@ -56,23 +59,40 @@ public class CxfClientCallback extends ClientCallback {
try {
super.handleResponse(ctx, res);
} finally {
- // add cookies to the cookie store
- if (endpoint.getCookieHandler() != null &&
cxfExchange.getInMessage() != null) {
- try {
- Map<String, List<String>> cxfHeaders
- = CastUtils.cast((Map<?, ?>)
cxfExchange.getInMessage().get(CxfConstants.PROTOCOL_HEADERS));
- endpoint.getCookieHandler().storeCookies(camelExchange,
endpoint.getRequestUri(camelExchange), cxfHeaders);
- } catch (IOException e) {
- LOG.error("Cannot store cookies", e);
- }
- }
- // bind the CXF response to Camel exchange and
- // call camel callback
// for one way messages callback is already called in
// process method of org.apache.camel.component.cxf.CxfProducer
if (!boi.getOperationInfo().isOneWay()) {
-
endpoint.getCxfBinding().populateExchangeFromCxfResponse(camelExchange,
cxfExchange, ctx);
- camelAsyncCallback.done(false);
+ // always signal the Camel callback, even if cookie store or
binding population throws,
+ // to prevent the calling thread from hanging in
DefaultAsyncProcessorAwaitManager
+ try {
+ // add cookies to the cookie store; catch Throwable so an
unexpected runtime failure
+ // does not prevent populateExchangeFromCxfResponse from
running or done() from firing
+ if (endpoint.getCookieHandler() != null &&
cxfExchange.getInMessage() != null) {
+ try {
+ Map<String, List<String>> cxfHeaders
+ = CastUtils.cast((Map<?, ?>)
cxfExchange.getInMessage().get(CxfConstants.PROTOCOL_HEADERS));
+
endpoint.getCookieHandler().storeCookies(camelExchange,
endpoint.getRequestUri(camelExchange),
+ cxfHeaders);
+ } catch (Throwable e) {
+ LOG.error("Cannot store cookies", e);
+ camelExchange.setException(e);
+ }
+ }
+ // bind the CXF response to Camel exchange
+ try {
+
endpoint.getCxfBinding().populateExchangeFromCxfResponse(camelExchange,
cxfExchange, ctx);
+ } catch (Throwable t) {
+ // preserve any exception set by the cookie store;
record binding error as suppressed
+ Exception cur = camelExchange.getException();
+ if (cur != null && cur != t) {
+ cur.addSuppressed(t);
+ } else if (cur == null) {
+ camelExchange.setException(t);
+ }
+ }
+ } finally {
+ signalDone();
+ }
}
if (LOG.isDebugEnabled()) {
LOG.debug("{} calling handleResponse",
Thread.currentThread().getName());
@@ -82,42 +102,76 @@ public class CxfClientCallback extends ClientCallback {
@Override
public void handleException(Map<String, Object> ctx, Throwable ex) {
+ // If done(false) was already signalled (e.g. handleResponse ran first
and then CXF re-dispatched
+ // to handleException after it threw), skip all exchange mutations:
the exchange is already done.
+ if (doneFired.get()) {
+ return;
+ }
try {
super.handleException(ctx, ex);
// need to call the conduitSelector complete method to enable the
fail over feature
ConduitSelector conduitSelector =
cxfExchange.get(ConduitSelector.class);
if (conduitSelector != null) {
conduitSelector.complete(cxfExchange);
+ // For FailoverTargetSelector, conduitSelector.complete() runs
the retry itself.
+ // If failover succeeds, neither message will carry an
exception — do not set one,
+ // as the exchange should remain clean (failover already
succeeded).
+ // Only propagate an exception when a message explicitly
carries one.
+ Throwable resolved = null;
if (cxfExchange.getOutMessage() != null) {
- ex =
cxfExchange.getOutMessage().getContent(Exception.class);
+ resolved =
cxfExchange.getOutMessage().getContent(Exception.class);
}
- if (ex == null && cxfExchange.getInMessage() != null) {
- ex =
cxfExchange.getInMessage().getContent(Exception.class);
+ if (resolved == null && cxfExchange.getInMessage() != null) {
+ resolved =
cxfExchange.getInMessage().getContent(Exception.class);
}
- if (ex != null) {
- camelExchange.setException(ex);
+ if (resolved != null) {
+ camelExchange.setException(resolved);
}
} else {
camelExchange.setException(ex);
}
} finally {
- // add cookies to the cookie store
- if (endpoint.getCookieHandler() != null &&
cxfExchange.getInMessage() != null) {
- try {
- Map<String, List<String>> cxfHeaders
- = CastUtils.cast((Map<?, ?>)
cxfExchange.getInMessage().get(CxfConstants.PROTOCOL_HEADERS));
- endpoint.getCookieHandler().storeCookies(camelExchange,
endpoint.getRequestUri(camelExchange), cxfHeaders);
- } catch (IOException e) {
- LOG.error("Cannot store cookies", e);
- }
- }
- // copy the context information and
- // call camel callback
// for one way messages callback is already called in
// process method of org.apache.camel.component.cxf.CxfProducer
if (!boi.getOperationInfo().isOneWay()) {
-
endpoint.getCxfBinding().populateExchangeFromCxfResponse(camelExchange,
cxfExchange, ctx);
- camelAsyncCallback.done(false);
+ // always signal the Camel callback, even if cookie store or
binding population throws,
+ // to prevent the calling thread from hanging in
DefaultAsyncProcessorAwaitManager
+ try {
+ // add cookies to the cookie store; catch Throwable so an
unexpected runtime failure
+ // does not prevent populateExchangeFromCxfResponse from
running or done() from firing;
+ // use suppressed to preserve the root transport/fault
exception already on the exchange
+ if (endpoint.getCookieHandler() != null &&
cxfExchange.getInMessage() != null) {
+ try {
+ Map<String, List<String>> cxfHeaders
+ = CastUtils.cast((Map<?, ?>)
cxfExchange.getInMessage().get(CxfConstants.PROTOCOL_HEADERS));
+
endpoint.getCookieHandler().storeCookies(camelExchange,
endpoint.getRequestUri(camelExchange),
+ cxfHeaders);
+ } catch (Throwable e) {
+ LOG.error("Cannot store cookies", e);
+ Throwable root = camelExchange.getException();
+ if (root != null && root != e) {
+ root.addSuppressed(e);
+ } else if (root == null) {
+ camelExchange.setException(e);
+ }
+ }
+ }
+ // copy the context information
+ try {
+
endpoint.getCxfBinding().populateExchangeFromCxfResponse(camelExchange,
cxfExchange, ctx);
+ } catch (Throwable t) {
+ // preserve the root transport/fault exception; record
the binding error as suppressed;
+ // guard against self-suppression (cur == t) which
throws IllegalArgumentException
+ Exception cur = camelExchange.getException();
+ if (cur != null && cur != t) {
+ cur.addSuppressed(t);
+ } else if (cur == null) {
+ camelExchange.setException(t);
+ }
+ }
+ } finally {
+ signalDone();
+ }
}
if (LOG.isDebugEnabled()) {
LOG.debug("{} calling handleException",
Thread.currentThread().getName());
@@ -125,4 +179,17 @@ public class CxfClientCallback extends ClientCallback {
}
}
+ /**
+ * Signals the Camel async callback exactly once. Guards against CXF
calling both {@code handleResponse} and
+ * {@code handleException} on the same callback instance (e.g. when {@code
handleResponse} throws and
+ * {@code ClientImpl.onMessage} re-dispatches).
+ */
+ private void signalDone() {
+ if (doneFired.compareAndSet(false, true)) {
+ camelAsyncCallback.done(false);
+ } else {
+ LOG.warn("done(false) already fired for this callback — duplicate
call suppressed");
+ }
+ }
+
}
diff --git
a/components/camel-cxf/camel-cxf-soap/src/test/java/org/apache/camel/component/cxf/jaxws/CxfClientCallbackTest.java
b/components/camel-cxf/camel-cxf-soap/src/test/java/org/apache/camel/component/cxf/jaxws/CxfClientCallbackTest.java
new file mode 100644
index 000000000000..20ec68636d4e
--- /dev/null
+++
b/components/camel-cxf/camel-cxf-soap/src/test/java/org/apache/camel/component/cxf/jaxws/CxfClientCallbackTest.java
@@ -0,0 +1,396 @@
+/*
+ * 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.cxf.jaxws;
+
+import java.net.URI;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.Exchange;
+import org.apache.camel.component.cxf.common.CxfBinding;
+import org.apache.camel.http.base.cookie.CookieHandler;
+import org.apache.cxf.endpoint.ConduitSelector;
+import org.apache.cxf.message.Message;
+import org.apache.cxf.service.model.BindingOperationInfo;
+import org.apache.cxf.service.model.OperationInfo;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Unit tests for {@link CxfClientCallback} verifying that {@code
camelAsyncCallback.done(false)} is always called
+ * exactly once, and that exception handling preserves the root cause.
+ *
+ * CAMEL-26696
+ */
+class CxfClientCallbackTest {
+
+ private Exchange camelExchange;
+ private org.apache.cxf.message.Exchange cxfExchange;
+ private BindingOperationInfo boi;
+ private CxfEndpoint endpoint;
+ private CxfBinding cxfBinding;
+
+ @BeforeEach
+ void setUp() {
+ camelExchange = mock(Exchange.class);
+ cxfExchange = mock(org.apache.cxf.message.Exchange.class);
+ boi = mock(BindingOperationInfo.class);
+ endpoint = mock(CxfEndpoint.class);
+ cxfBinding = mock(CxfBinding.class);
+
+ OperationInfo operationInfo = mock(OperationInfo.class);
+ when(boi.getOperationInfo()).thenReturn(operationInfo);
+ when(operationInfo.isOneWay()).thenReturn(false);
+
+ when(endpoint.getCxfBinding()).thenReturn(cxfBinding);
+ when(endpoint.getCookieHandler()).thenReturn(null); // no cookie
handler
+ }
+
+ // --- handleResponse tests ---
+
+ @Test
+ void handleResponse_normalPath_callsDoneExactlyOnce() {
+ AtomicInteger doneCount = new AtomicInteger();
+ AsyncCallback callback = sync -> doneCount.incrementAndGet();
+ Map<String, Object> ctx = new HashMap<>();
+
+ CxfClientCallback cb = new CxfClientCallback(callback, camelExchange,
cxfExchange, boi, endpoint);
+ cb.handleResponse(ctx, new Object[0]);
+
+ assertEquals(1, doneCount.get(), "done(false) must be called exactly
once");
+ }
+
+ @Test
+ void handleResponse_bindingThrows_callsDoneAndSetsException() {
+ RuntimeException bindingError = new RuntimeException("binding
failure");
+
doThrow(bindingError).when(cxfBinding).populateExchangeFromCxfResponse(any(),
any(), any());
+
+ AtomicInteger doneCount = new AtomicInteger();
+ AsyncCallback callback = sync -> doneCount.incrementAndGet();
+
+ // Capture the exception set on the exchange
+ AtomicReference<Throwable> captured = new AtomicReference<>();
+ camelExchange = mock(Exchange.class);
+ doAnswer(inv -> {
+ captured.set((Throwable) inv.getArgument(0));
+ return null;
+ }).when(camelExchange).setException(any(Throwable.class));
+
+ Map<String, Object> ctx = new HashMap<>();
+ CxfClientCallback cb = new CxfClientCallback(callback, camelExchange,
cxfExchange, boi, endpoint);
+ cb.handleResponse(ctx, new Object[0]);
+
+ assertEquals(1, doneCount.get(), "done(false) must be called even when
binding throws");
+ assertSame(bindingError, captured.get());
+ }
+
+ @Test
+ void handleResponse_bindingThrowsError_callsDoneAndSetsException() {
+ OutOfMemoryError bindingError = new OutOfMemoryError("simulated OOM in
binding");
+
doThrow(bindingError).when(cxfBinding).populateExchangeFromCxfResponse(any(),
any(), any());
+
+ AtomicInteger doneCount = new AtomicInteger();
+ AsyncCallback callback = sync -> doneCount.incrementAndGet();
+
+ AtomicReference<Throwable> captured = new AtomicReference<>();
+ camelExchange = mock(Exchange.class);
+ doAnswer(inv -> {
+ captured.set((Throwable) inv.getArgument(0));
+ return null;
+ }).when(camelExchange).setException(any(Throwable.class));
+
+ Map<String, Object> ctx = new HashMap<>();
+ CxfClientCallback cb = new CxfClientCallback(callback, camelExchange,
cxfExchange, boi, endpoint);
+ cb.handleResponse(ctx, new Object[0]);
+
+ assertEquals(1, doneCount.get(), "done(false) must be called even when
binding throws an Error");
+ assertSame(bindingError, captured.get());
+ }
+
+ // --- handleException tests ---
+
+ @Test
+ void handleException_normalPath_callsDoneExactlyOnce() {
+ AtomicInteger doneCount = new AtomicInteger();
+ AsyncCallback callback = sync -> doneCount.incrementAndGet();
+
+ RuntimeException transportEx = new RuntimeException("connection
refused");
+
+ Map<String, Object> ctx = new HashMap<>();
+ CxfClientCallback cb = new CxfClientCallback(callback, camelExchange,
cxfExchange, boi, endpoint);
+ cb.handleException(ctx, transportEx);
+
+ assertEquals(1, doneCount.get(), "done(false) must be called exactly
once");
+ verify(camelExchange).setException(transportEx);
+ }
+
+ @Test
+ void
handleException_withConduitSelector_usesResolvedExceptionFromOutMessage() {
+ // When a ConduitSelector is set and outMessage carries a richer
exception, that takes priority
+ RuntimeException transportEx = new RuntimeException("connection
refused");
+ RuntimeException resolvedEx = new RuntimeException("resolved from
outMessage");
+
+ ConduitSelector conduitSelector = mock(ConduitSelector.class);
+ Message outMessage = mock(Message.class);
+
when(cxfExchange.get(ConduitSelector.class)).thenReturn(conduitSelector);
+ when(cxfExchange.getOutMessage()).thenReturn(outMessage);
+ when(outMessage.getContent(Exception.class)).thenReturn(resolvedEx);
+
+ AtomicInteger doneCount = new AtomicInteger();
+ AsyncCallback callback = sync -> doneCount.incrementAndGet();
+
+ Map<String, Object> ctx = new HashMap<>();
+ CxfClientCallback cb = new CxfClientCallback(callback, camelExchange,
cxfExchange, boi, endpoint);
+ cb.handleException(ctx, transportEx);
+
+ assertEquals(1, doneCount.get(), "done(false) must be called exactly
once");
+ verify(camelExchange).setException(resolvedEx);
+ }
+
+ @Test
+ void
handleException_withConduitSelector_noExceptionOnMessages_doesNotSetException()
{
+ // When ConduitSelector is set and neither message carries an
exception, the failover
+ // succeeded — do NOT fall back to the original transport exception,
leave the exchange
+ // clean (failover already succeeded).
+ RuntimeException transportEx = new RuntimeException("connection
refused");
+
+ ConduitSelector conduitSelector = mock(ConduitSelector.class);
+ Message outMessage = mock(Message.class);
+
when(cxfExchange.get(ConduitSelector.class)).thenReturn(conduitSelector);
+ when(cxfExchange.getOutMessage()).thenReturn(outMessage);
+ when(outMessage.getContent(Exception.class)).thenReturn(null);
+ when(cxfExchange.getInMessage()).thenReturn(null);
+
+ AtomicInteger doneCount = new AtomicInteger();
+ AsyncCallback callback = sync -> doneCount.incrementAndGet();
+
+ Map<String, Object> ctx = new HashMap<>();
+ CxfClientCallback cb = new CxfClientCallback(callback, camelExchange,
cxfExchange, boi, endpoint);
+ cb.handleException(ctx, transportEx);
+
+ assertEquals(1, doneCount.get(), "done(false) must be called exactly
once");
+ // failover succeeded: setException must NOT be called (exchange
should remain clean)
+ verify(camelExchange, never()).setException(any());
+ }
+
+ @Test
+ void handleException_bindingThrows_preservesRootCauseAndCallsDone() {
+ RuntimeException transportEx = new RuntimeException("connection
refused");
+ RuntimeException bindingEx = new RuntimeException("binding failure");
+
doThrow(bindingEx).when(cxfBinding).populateExchangeFromCxfResponse(any(),
any(), any());
+
+ AtomicInteger doneCount = new AtomicInteger();
+ AsyncCallback callback = sync -> doneCount.incrementAndGet();
+
+ // Simulate exchange already has the transport exception set
+ when(camelExchange.getException()).thenReturn(transportEx);
+
+ Map<String, Object> ctx = new HashMap<>();
+ CxfClientCallback cb = new CxfClientCallback(callback, camelExchange,
cxfExchange, boi, endpoint);
+ cb.handleException(ctx, transportEx);
+
+ assertEquals(1, doneCount.get(), "done(false) must be called even when
binding throws");
+ // binding error should be suppressed on the root cause, not replace it
+ assertEquals(1, transportEx.getSuppressed().length);
+ assertSame(bindingEx, transportEx.getSuppressed()[0]);
+ }
+
+ @Test
+ void handleException_bindingThrows_noRootCause_setsBindingException() {
+ RuntimeException bindingEx = new RuntimeException("binding failure");
+
doThrow(bindingEx).when(cxfBinding).populateExchangeFromCxfResponse(any(),
any(), any());
+
+ AtomicInteger doneCount = new AtomicInteger();
+ AsyncCallback callback = sync -> doneCount.incrementAndGet();
+
+ // No prior exception on the exchange
+ when(camelExchange.getException()).thenReturn(null);
+ AtomicReference<Throwable> captured = new AtomicReference<>();
+ doAnswer(inv -> {
+ captured.set((Throwable) inv.getArgument(0));
+ return null;
+ }).when(camelExchange).setException(any(Throwable.class));
+
+ RuntimeException transportEx = new RuntimeException("connection
refused");
+ Map<String, Object> ctx = new HashMap<>();
+ CxfClientCallback cb = new CxfClientCallback(callback, camelExchange,
cxfExchange, boi, endpoint);
+ cb.handleException(ctx, transportEx);
+
+ assertEquals(1, doneCount.get());
+ // when there's no root cause, the binding exception should be set
+ assertSame(bindingEx, captured.get());
+ }
+
+ @Test
+ void handleException_bindingThrowsError_callsDoneAndPreservesRootCause() {
+ RuntimeException transportEx = new RuntimeException("connection
refused");
+ OutOfMemoryError bindingError = new OutOfMemoryError("simulated OOM in
binding");
+
doThrow(bindingError).when(cxfBinding).populateExchangeFromCxfResponse(any(),
any(), any());
+
+ AtomicInteger doneCount = new AtomicInteger();
+ AsyncCallback callback = sync -> doneCount.incrementAndGet();
+
+ when(camelExchange.getException()).thenReturn(transportEx);
+
+ Map<String, Object> ctx = new HashMap<>();
+ CxfClientCallback cb = new CxfClientCallback(callback, camelExchange,
cxfExchange, boi, endpoint);
+ cb.handleException(ctx, transportEx);
+
+ assertEquals(1, doneCount.get(), "done(false) must be called even when
binding throws an Error");
+ assertEquals(1, transportEx.getSuppressed().length);
+ assertSame(bindingError, transportEx.getSuppressed()[0]);
+ }
+
+ @Test
+ void doneFiredOnlyOnce_whenHandleExceptionCalledAfterHandleResponse() {
+ // Simulate the scenario where CXF calls handleException after
handleResponse
+ // (e.g. because ClientImpl.onMessage catches a thrown exception and
re-dispatches).
+ // The AtomicBoolean guard must prevent done(false) from firing twice.
+ AtomicInteger doneCount = new AtomicInteger();
+ AsyncCallback callback = sync -> doneCount.incrementAndGet();
+
+ RuntimeException transportEx = new RuntimeException("connection
refused");
+ Map<String, Object> ctx = new HashMap<>();
+
+ CxfClientCallback cb = new CxfClientCallback(callback, camelExchange,
cxfExchange, boi, endpoint);
+ cb.handleResponse(ctx, new Object[0]);
+ cb.handleException(ctx, transportEx);
+
+ assertEquals(1, doneCount.get(),
+ "done(false) must fire exactly once even if both
handleResponse and handleException are called");
+ }
+
+ // --- cookie handler tests ---
+
+ @Test
+ void handleResponse_cookieStoreThrows_callsDoneAndSetsException() throws
Exception {
+ // CookieHandler.storeCookies declares IOException but could throw
RuntimeException too;
+ // any Throwable from the cookie store must be recorded and must not
prevent done() from firing.
+ RuntimeException cookieError = new RuntimeException("cookie store
failure");
+ CookieHandler cookieHandler = mock(CookieHandler.class);
+ doThrow(cookieError).when(cookieHandler).storeCookies(any(),
any(URI.class), any());
+
+ Message inMessage = mock(Message.class);
+ when(inMessage.get(any())).thenReturn(Collections.emptyMap());
+ when(cxfExchange.getInMessage()).thenReturn(inMessage);
+ when(endpoint.getCookieHandler()).thenReturn(cookieHandler);
+
when(endpoint.getRequestUri(any())).thenReturn(URI.create("http://wrong-server/";));
+
+ AtomicInteger doneCount = new AtomicInteger();
+ AsyncCallback callback = sync -> doneCount.incrementAndGet();
+
+ AtomicReference<Throwable> captured = new AtomicReference<>();
+ camelExchange = mock(Exchange.class);
+ doAnswer(inv -> {
+ captured.set((Throwable) inv.getArgument(0));
+ return null;
+ }).when(camelExchange).setException(any(Throwable.class));
+
+ when(endpoint.getCxfBinding()).thenReturn(cxfBinding);
+
+ Map<String, Object> ctx = new HashMap<>();
+ CxfClientCallback cb = new CxfClientCallback(callback, camelExchange,
cxfExchange, boi, endpoint);
+ cb.handleResponse(ctx, new Object[0]);
+
+ assertEquals(1, doneCount.get(), "done(false) must be called even when
cookie store throws");
+ assertSame(cookieError, captured.get());
+ }
+
+ @Test
+ void handleException_cookieStoreThrows_suppressesOnRootCauseAndCallsDone()
throws Exception {
+ // Cookie store failure in the exception path: must be suppressed on
the root transport exception.
+ RuntimeException transportEx = new RuntimeException("connection
refused");
+ RuntimeException cookieError = new RuntimeException("cookie store
failure");
+ CookieHandler cookieHandler = mock(CookieHandler.class);
+ doThrow(cookieError).when(cookieHandler).storeCookies(any(),
any(URI.class), any());
+
+ Message inMessage = mock(Message.class);
+ when(inMessage.get(any())).thenReturn(Collections.emptyMap());
+ when(cxfExchange.getInMessage()).thenReturn(inMessage);
+ when(endpoint.getCookieHandler()).thenReturn(cookieHandler);
+
when(endpoint.getRequestUri(any())).thenReturn(URI.create("http://wrong-server/";));
+
+ AtomicInteger doneCount = new AtomicInteger();
+ AsyncCallback callback = sync -> doneCount.incrementAndGet();
+
+ when(camelExchange.getException()).thenReturn(transportEx);
+ when(endpoint.getCxfBinding()).thenReturn(cxfBinding);
+
+ Map<String, Object> ctx = new HashMap<>();
+ CxfClientCallback cb = new CxfClientCallback(callback, camelExchange,
cxfExchange, boi, endpoint);
+ cb.handleException(ctx, transportEx);
+
+ assertEquals(1, doneCount.get(), "done(false) must be called even when
cookie store throws");
+ // cookie error must be suppressed on root transport exception, not
replace it
+ assertEquals(1, transportEx.getSuppressed().length);
+ assertSame(cookieError, transportEx.getSuppressed()[0]);
+ }
+
+ @Test
+ void
handleResponse_cookieAndBindingThrow_bindingErrorSuppressedOnCookieError()
throws Exception {
+ // Both cookie store and binding throw in the success path; the
binding error must be suppressed
+ // on the cookie error (not silently lost), and done(false) must still
fire.
+ RuntimeException cookieError = new RuntimeException("cookie store
failure");
+ RuntimeException bindingError = new RuntimeException("binding
failure");
+ CookieHandler cookieHandler = mock(CookieHandler.class);
+ doThrow(cookieError).when(cookieHandler).storeCookies(any(),
any(URI.class), any());
+
doThrow(bindingError).when(cxfBinding).populateExchangeFromCxfResponse(any(),
any(), any());
+
+ Message inMessage = mock(Message.class);
+ when(inMessage.get(any())).thenReturn(Collections.emptyMap());
+ when(cxfExchange.getInMessage()).thenReturn(inMessage);
+ when(endpoint.getCookieHandler()).thenReturn(cookieHandler);
+
when(endpoint.getRequestUri(any())).thenReturn(URI.create("http://wrong-server/";));
+
+ AtomicInteger doneCount = new AtomicInteger();
+ AsyncCallback callback = sync -> doneCount.incrementAndGet();
+
+ AtomicReference<Throwable> captured = new AtomicReference<>();
+ camelExchange = mock(Exchange.class);
+ doAnswer(inv -> {
+ captured.set((Throwable) inv.getArgument(0));
+ return null;
+ }).when(camelExchange).setException(any(Throwable.class));
+ when(camelExchange.getException()).thenAnswer(inv -> captured.get());
+
+ when(endpoint.getCxfBinding()).thenReturn(cxfBinding);
+
+ Map<String, Object> ctx = new HashMap<>();
+ CxfClientCallback cb = new CxfClientCallback(callback, camelExchange,
cxfExchange, boi, endpoint);
+ cb.handleResponse(ctx, new Object[0]);
+
+ assertEquals(1, doneCount.get(), "done(false) must be called even when
both cookie store and binding throw");
+ // cookie error is the primary exception; binding error must be
suppressed on it, not silently lost
+ assertSame(cookieError, captured.get());
+ assertEquals(1, cookieError.getSuppressed().length);
+ assertSame(bindingError, cookieError.getSuppressed()[0]);
+ }
+}
diff --git
a/components/camel-cxf/camel-cxf-soap/src/test/java/org/apache/camel/component/cxf/jaxws/CxfProducerTest.java
b/components/camel-cxf/camel-cxf-soap/src/test/java/org/apache/camel/component/cxf/jaxws/CxfProducerTest.java
index e444b9e04017..3b83162c250c 100644
---
a/components/camel-cxf/camel-cxf-soap/src/test/java/org/apache/camel/component/cxf/jaxws/CxfProducerTest.java
+++
b/components/camel-cxf/camel-cxf-soap/src/test/java/org/apache/camel/component/cxf/jaxws/CxfProducerTest.java
@@ -48,6 +48,7 @@ import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
import org.junit.jupiter.api.Timeout.ThreadMode;
@@ -192,7 +193,7 @@ public class CxfProducerTest {
}
@Test
- public void testInvokingAWrongServer() throws Exception {
+ void testInvokingAWrongServer() throws Exception {
Exchange reply = sendSimpleMessage(getWrongEndpointUri());
assertNotNull(reply.getException(), "We should get the exception
here");
assertInstanceOf(IOException.class, reply.getException().getCause());
@@ -208,6 +209,22 @@ public class CxfProducerTest {
assertInstanceOf(IOException.class, reply.getException().getCause());
}
+ // CAMEL-26696: async producer must always complete the exchange when the
connection fails.
+ // The @Timeout(30s) on the class guards against the exchange hanging in
+ // DefaultAsyncProcessorAwaitManager if camelAsyncCallback.done() is never
called.
+ // For the callback exception-handling semantics (binding throws, Error
thrown, etc.)
+ // see CxfClientCallbackTest which tests CxfClientCallback directly via
Mockito.
+ // Disabled: this test passes with or without the fix (CXF goes through
handleException, not
+ // through populateExchangeFromCxfResponse, on a connection failure), and
CXF-9272 makes it
+ // hang occasionally under load. The CxfClientCallbackTest unit tests
cover the invariants.
+ @Disabled("CXF-9272: may hang under load; callback invariants covered by
CxfClientCallbackTest")
+ @Test
+ void testInvokingAWrongServerAsync() throws Exception {
+ Exchange reply = sendSimpleMessage(getWrongEndpointUriAsync());
+ assertNotNull(reply.getException(), "Async wrong-server call must
complete with an exception");
+ assertInstanceOf(IOException.class, reply.getException().getCause());
+ }
+
@Test
public void testInvokingJaxWsServerWithParams() throws Exception {
Exchange exchange = sendJaxWsMessage();
@@ -238,13 +255,20 @@ public class CxfProducerTest {
protected String getWrongEndpointUri() {
// CAMEL-26600: synchronous=true, as with the CXF async client the
callback sometimes never completes
// when the connection fails, leaving the exchange waiting forever.
- // The async error path is not covered anymore, see CAMEL-26696
+ // The async error path is not covered here; see CxfClientCallbackTest
and CAMEL-26696.
return "cxf://" + getWrongServerAddress()
+
"?serviceClass=org.apache.camel.component.cxf.jaxws.HelloService"
+ "&cxfConfigurer=#wrongServerConfigurer"
+ "&synchronous=true";
}
+ // CAMEL-26696: async (synchronous=false) wrong-server endpoint — callback
must always complete
+ protected String getWrongEndpointUriAsync() {
+ return "cxf://" + getWrongServerAddress()
+ +
"?serviceClass=org.apache.camel.component.cxf.jaxws.HelloService"
+ + "&cxfConfigurer=#wrongServerConfigurer";
+ }
+
protected Exchange sendSimpleMessage() {
return sendSimpleMessage(getSimpleEndpointUri());
}
