This is an automated email from the ASF dual-hosted git repository.
ffang pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/cxf.git
The following commit(s) were added to refs/heads/main by this push:
new c3837b37948 CXF-7232: Report a failure of the retransmits of an
asynchronous call… (#3706)
c3837b37948 is described below
commit c3837b37948ccaa6e7be338d87d3fdc8cb2da3d6
Author: Freeman(Yue) Fang <[email protected]>
AuthorDate: Thu Nov 12 09:59:12 2026 -0500
CXF-7232: Report a failure of the retransmits of an asynchronous call…
(#3706)
* CXF-7232: Report a failure of the retransmits of an asynchronous call as
before
Since the retransmits of an asynchronous call are processed on the thread
that
handles the response, a failure of them, e.g. a refused connection with
MaxRetransmits set, reached the fault observer as the raw exception. When
they
were processed by the caller's close(), it was mapped and reported as the
fault
of sending the message ("Could not send Message."). Report it the same way
again.
DispatchImpl.mapException() threw the WebServiceException of such a fault
instead of returning it, so an asynchronous Dispatch call that got it never
completed.
* Rename handleRetransmitsAndResponseAsync to
handleRetransmitsAndResponseOnWorkqueue
It is only called by the runnable of handleResponseOnWorkqueue.
---
.../java/org/apache/cxf/jaxws/DispatchImpl.java | 2 +-
.../org/apache/cxf/transport/http/HTTPConduit.java | 69 +++++++++++++++++++---
.../cxf/systest/hc5/http/AsyncRetransmitTest.java | 36 +++++++++++
.../cxf/systest/http/AsyncRetransmitTest.java | 36 +++++++++++
4 files changed, 133 insertions(+), 10 deletions(-)
diff --git
a/rt/frontend/jaxws/src/main/java/org/apache/cxf/jaxws/DispatchImpl.java
b/rt/frontend/jaxws/src/main/java/org/apache/cxf/jaxws/DispatchImpl.java
index 702a3411dc7..3b4d9b2681c 100644
--- a/rt/frontend/jaxws/src/main/java/org/apache/cxf/jaxws/DispatchImpl.java
+++ b/rt/frontend/jaxws/src/main/java/org/apache/cxf/jaxws/DispatchImpl.java
@@ -264,7 +264,7 @@ public class DispatchImpl<T> implements Dispatch<T>,
BindingProvider, Closeable
}
private RuntimeException mapException(Exception ex) {
if (ex instanceof Fault && ex.getCause() instanceof IOException) {
- throw new WebServiceException(ex.getMessage(), ex.getCause());
+ return new WebServiceException(ex.getMessage(), ex.getCause());
}
if (getBinding() instanceof HTTPBinding) {
diff --git
a/rt/transports/http/src/main/java/org/apache/cxf/transport/http/HTTPConduit.java
b/rt/transports/http/src/main/java/org/apache/cxf/transport/http/HTTPConduit.java
index 4bf882924fd..f3bfe4293b8 100644
---
a/rt/transports/http/src/main/java/org/apache/cxf/transport/http/HTTPConduit.java
+++
b/rt/transports/http/src/main/java/org/apache/cxf/transport/http/HTTPConduit.java
@@ -30,6 +30,7 @@ import java.lang.ref.WeakReference;
import java.net.HttpRetryException;
import java.net.HttpURLConnection;
import java.net.MalformedURLException;
+import java.net.SocketTimeoutException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.Arrays;
@@ -38,6 +39,7 @@ import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.ResourceBundle;
import java.util.Set;
import java.util.concurrent.Executor;
import java.util.concurrent.RejectedExecutionException;
@@ -47,6 +49,7 @@ import java.util.logging.Logger;
import javax.xml.namespace.QName;
import org.apache.cxf.Bus;
+import org.apache.cxf.common.i18n.BundleUtils;
import org.apache.cxf.common.injection.NoJSR250Annotations;
import org.apache.cxf.common.logging.LogUtils;
import org.apache.cxf.common.util.PropertyUtils;
@@ -62,6 +65,8 @@ import org.apache.cxf.helpers.CastUtils;
import org.apache.cxf.helpers.HttpHeaderHelper;
import org.apache.cxf.helpers.IOUtils;
import org.apache.cxf.helpers.LoadingByteArrayOutputStream;
+import org.apache.cxf.interceptor.Fault;
+import org.apache.cxf.interceptor.MessageSenderInterceptor;
import org.apache.cxf.io.AbstractThresholdOutputStream;
import org.apache.cxf.io.CacheAndWriteOutputStream;
import org.apache.cxf.io.CachedOutputStream;
@@ -223,6 +228,11 @@ public abstract class HTTPConduit
*/
private static final String KEY_AUTH_URLS = "AuthURLs";
+ /**
+ * The messages of the faults of the out chain for a failure of the conduit
+ */
+ private static final ResourceBundle SENDER_BUNDLE =
BundleUtils.getBundle(MessageSenderInterceptor.class);
+
/**
* This field holds a reference to the CXF bus associated this conduit.
*/
@@ -1235,7 +1245,7 @@ public abstract class HTTPConduit
Runnable runnable = new Runnable() {
public void run() {
try {
- handleRetransmitsAndResponse();
+ handleRetransmitsAndResponseOnWorkqueue();
} catch (Throwable e) {
((PhaseInterceptorChain)outMessage.getInterceptorChain()).abort();
outMessage.setContent(Exception.class, e);
@@ -1428,14 +1438,7 @@ public abstract class HTTPConduit
} catch (HttpRetryException e) {
handleHttpRetryException(e);
} catch (IOException e) {
- String origMessage = e.getMessage();
- if (origMessage != null &&
origMessage.contains(url.toString())) {
- throw e;
- }
- throw mapException(e.getClass().getSimpleName()
- + " invoking " + url + ": "
- + getExceptionMessage(e), e,
- IOException.class);
+ throw mapException(e);
} catch (RuntimeException e) {
throw mapException(e.getClass().getSimpleName()
+ " invoking " + url + ": "
@@ -1444,6 +1447,17 @@ public abstract class HTTPConduit
}
}
+ private IOException mapException(IOException e) {
+ String origMessage = e.getMessage();
+ if (origMessage != null && origMessage.contains(url.toString())) {
+ return e;
+ }
+ return mapException(e.getClass().getSimpleName()
+ + " invoking " + url + ": "
+ + getExceptionMessage(e), e,
+ IOException.class);
+ }
+
protected String getExceptionMessage(Throwable t) {
return t.getMessage();
}
@@ -1645,6 +1659,43 @@ public abstract class HTTPConduit
handleResponseInternal();
}
+ /**
+ * Processes the retransmits and the response on the workqueue, see
+ * {@link #handleResponseOnWorkqueue(boolean, boolean)}. A failure of
the retransmits is
+ * reported as it was when they were processed by the caller: as
close() maps it, and the
+ * out chain then reports it.
+ */
+ private void handleRetransmitsAndResponseOnWorkqueue() throws
IOException {
+ try {
+ handleRetransmits();
+ } catch (HttpRetryException e) {
+ try {
+ handleHttpRetryException(e);
+ } catch (IOException ex) {
+ throw sendFault(ex);
+ }
+ } catch (IOException e) {
+ throw sendFault(mapException(e));
+ } catch (RuntimeException e) {
+ throw mapException(e.getClass().getSimpleName()
+ + " invoking " + url + ": "
+ + getExceptionMessage(e), e,
+ RuntimeException.class);
+ } finally {
+ releaseCachedStream();
+ }
+ handleResponseInternal();
+ }
+
+ /**
+ * The fault of the out chain for a failure of the conduit, as
reported by
+ * {@link MessageSenderInterceptor.MessageSenderEndingInterceptor}
+ */
+ private Fault sendFault(IOException e) {
+ return new Fault(new org.apache.cxf.common.i18n.Message(e
instanceof SocketTimeoutException
+ ? "COULD_NOT_RECEIVE" : "COULD_NOT_SEND", SENDER_BUNDLE), e);
+ }
+
private void releaseCachedStream() throws IOException {
if (cachingForRetransmission && cachedStream != null) {
cachedStream.close();
diff --git
a/systests/transport-hc5/src/test/java/org/apache/cxf/systest/hc5/http/AsyncRetransmitTest.java
b/systests/transport-hc5/src/test/java/org/apache/cxf/systest/hc5/http/AsyncRetransmitTest.java
index 19a279ec5a5..14467be7640 100644
---
a/systests/transport-hc5/src/test/java/org/apache/cxf/systest/hc5/http/AsyncRetransmitTest.java
+++
b/systests/transport-hc5/src/test/java/org/apache/cxf/systest/hc5/http/AsyncRetransmitTest.java
@@ -25,6 +25,7 @@ import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.io.StringReader;
+import java.net.ConnectException;
import java.net.InetSocketAddress;
import java.net.URI;
import java.nio.charset.StandardCharsets;
@@ -50,6 +51,7 @@ import com.sun.net.httpserver.HttpServer;
import jakarta.xml.ws.Dispatch;
import jakarta.xml.ws.Response;
import jakarta.xml.ws.Service;
+import jakarta.xml.ws.WebServiceException;
import jakarta.xml.ws.soap.SOAPBinding;
import org.apache.cxf.Bus;
import org.apache.cxf.BusFactory;
@@ -89,6 +91,8 @@ import static org.junit.Assert.fail;
@RunWith(Parameterized.class)
public class AsyncRetransmitTest {
public static final String PORT =
TestUtil.getPortNumber(AsyncRetransmitTest.class);
+ // nothing listens on it
+ public static final String REFUSED_PORT =
TestUtil.getPortNumber(AsyncRetransmitTest.class, 2);
private static final QName SERVICE = new QName("urn:cxf7232",
"EchoService");
private static final QName PORT_NAME = new QName("urn:cxf7232",
"EchoPort");
private static final String REQUEST = "<echo
xmlns=\"urn:cxf7232\">hello</echo>";
@@ -287,6 +291,38 @@ public class AsyncRetransmitTest {
}
}
+ @Test
+ public void testMaxRetransmitsConnectionRefused() throws Exception {
+ HTTPClientPolicy policy = new HTTPClientPolicy();
+ policy.setMaxRetransmits(1);
+ Dispatch<Source> dispatch = createDispatch("/echo", policy);
+ String address = "http://localhost:"; + REFUSED_PORT + "/echo";
+ dispatch.getRequestContext().put(Message.ENDPOINT_ADDRESS, address);
+ Throwable failure;
+ try {
+ Response<Source> response = dispatch.invokeAsync(new
StreamSource(new StringReader(REQUEST)));
+ try {
+ response.get(30, TimeUnit.SECONDS);
+ fail("A refused connection must fail the asynchronous
response");
+ return;
+ } catch (ExecutionException ex) {
+ failure = ex.getCause();
+ }
+ } catch (WebServiceException ex) {
+ // the connection may already be refused while the caller thread
sends the request
+ failure = ex;
+ } finally {
+ destroy(dispatch);
+ }
+ // reported as when the retransmits were processed by the caller thread
+ assertEquals("Could not send Message.", failure.getMessage());
+ Throwable cause = failure.getCause();
+ while (cause != null && !(cause instanceof ConnectException)) {
+ cause = cause.getCause();
+ }
+ assertNotNull("Expected the refused connection: " + failure, cause);
+ }
+
private void invokeAsync(String path, HTTPClientPolicy policy) throws
Exception {
Dispatch<Source> dispatch = createDispatch(path, policy);
try {
diff --git
a/systests/transports/src/test/java/org/apache/cxf/systest/http/AsyncRetransmitTest.java
b/systests/transports/src/test/java/org/apache/cxf/systest/http/AsyncRetransmitTest.java
index 8a2d00bce60..f1eaca5a48e 100644
---
a/systests/transports/src/test/java/org/apache/cxf/systest/http/AsyncRetransmitTest.java
+++
b/systests/transports/src/test/java/org/apache/cxf/systest/http/AsyncRetransmitTest.java
@@ -25,6 +25,7 @@ import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.io.StringReader;
+import java.net.ConnectException;
import java.net.InetSocketAddress;
import java.net.URI;
import java.nio.charset.StandardCharsets;
@@ -50,6 +51,7 @@ import com.sun.net.httpserver.HttpServer;
import jakarta.xml.ws.Dispatch;
import jakarta.xml.ws.Response;
import jakarta.xml.ws.Service;
+import jakarta.xml.ws.WebServiceException;
import jakarta.xml.ws.soap.SOAPBinding;
import org.apache.cxf.Bus;
import org.apache.cxf.BusFactory;
@@ -89,6 +91,8 @@ import static org.junit.Assert.fail;
@RunWith(Parameterized.class)
public class AsyncRetransmitTest {
public static final String PORT =
TestUtil.getPortNumber(AsyncRetransmitTest.class);
+ // nothing listens on it
+ public static final String REFUSED_PORT =
TestUtil.getPortNumber(AsyncRetransmitTest.class, 2);
private static final QName SERVICE = new QName("urn:cxf7232",
"EchoService");
private static final QName PORT_NAME = new QName("urn:cxf7232",
"EchoPort");
private static final String REQUEST = "<echo
xmlns=\"urn:cxf7232\">hello</echo>";
@@ -287,6 +291,38 @@ public class AsyncRetransmitTest {
}
}
+ @Test
+ public void testMaxRetransmitsConnectionRefused() throws Exception {
+ HTTPClientPolicy policy = new HTTPClientPolicy();
+ policy.setMaxRetransmits(1);
+ Dispatch<Source> dispatch = createDispatch("/echo", policy);
+ String address = "http://localhost:"; + REFUSED_PORT + "/echo";
+ dispatch.getRequestContext().put(Message.ENDPOINT_ADDRESS, address);
+ Throwable failure;
+ try {
+ Response<Source> response = dispatch.invokeAsync(new
StreamSource(new StringReader(REQUEST)));
+ try {
+ response.get(30, TimeUnit.SECONDS);
+ fail("A refused connection must fail the asynchronous
response");
+ return;
+ } catch (ExecutionException ex) {
+ failure = ex.getCause();
+ }
+ } catch (WebServiceException ex) {
+ // the connection may already be refused while the caller thread
sends the request
+ failure = ex;
+ } finally {
+ destroy(dispatch);
+ }
+ // reported as when the retransmits were processed by the caller thread
+ assertEquals("Could not send Message.", failure.getMessage());
+ Throwable cause = failure.getCause();
+ while (cause != null && !(cause instanceof ConnectException)) {
+ cause = cause.getCause();
+ }
+ assertNotNull("Expected the refused connection: " + failure, cause);
+ }
+
private void invokeAsync(String path, HTTPClientPolicy policy) throws
Exception {
Dispatch<Source> dispatch = createDispatch(path, policy);
try {
