This is an automated email from the ASF dual-hosted git repository.
ffang pushed a commit to branch 4.1.x-fixes
in repository https://gitbox.apache.org/repos/asf/cxf.git
The following commit(s) were added to refs/heads/4.1.x-fixes by this push:
new 9fb69ad4dd6 CXF-9272: Report the failure of an asynchronous call whose
connection… (#3701)
9fb69ad4dd6 is described below
commit 9fb69ad4dd63dba84266f05ac85a731d7ded6d9b
Author: Freeman(Yue) Fang <[email protected]>
AuthorDate: Wed Nov 11 08:32:41 2026 -0500
CXF-9272: Report the failure of an asynchronous call whose connection…
(#3701)
* CXF-9272: Report the failure of an asynchronous call whose connection is
refused
When the request of an asynchronous call failed, for example because the
connection was refused, the failure callback of HttpClientHTTPConduit
recorded the exception on the out message and closed the stream. The caller
closing the stream afterwards then found the exception and did not schedule
the response handling, as it takes it as reported already, so neither the
response nor the fault observer was ever called. This happened whenever the
connection failed before the caller closed the stream, for example while it
was still writing a large body, or when sending a small one from close().
The failure callback now only releases a writer waiting for the connection
or blocked on the body, by closing the body stream. The caller's close()
schedules the response handling, which gets the failure from the future and
reports it to the fault observer, once. As the callback no longer closes
the stream, it cannot send a request without a body again either.
* CXF-9272: Check that the failure reported is the refused connection
The test now also checks that the failure reported, to the fault observer
or to the caller by the stream, is caused by the ConnectException of the
refused connection.
(cherry picked from commit 2d1bcdb5ece9d2d789d1e82bccadb705468413eb)
---
rt/transports/http/pom.xml | 5 +
.../cxf/transport/http/HttpClientHTTPConduit.java | 29 ++-
.../HttpClientHTTPConduitAsyncFailureTest.java | 216 +++++++++++++++++++++
3 files changed, 235 insertions(+), 15 deletions(-)
diff --git a/rt/transports/http/pom.xml b/rt/transports/http/pom.xml
index 5df9977cb57..26377d8f659 100644
--- a/rt/transports/http/pom.xml
+++ b/rt/transports/http/pom.xml
@@ -68,6 +68,11 @@
<version>${cxf.mockito.version}</version>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.cxf</groupId>
+ <artifactId>cxf-testutils</artifactId>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.apache.cxf</groupId>
<artifactId>cxf-core</artifactId>
diff --git
a/rt/transports/http/src/main/java/org/apache/cxf/transport/http/HttpClientHTTPConduit.java
b/rt/transports/http/src/main/java/org/apache/cxf/transport/http/HttpClientHTTPConduit.java
index 472f83dd9c6..341a7e9868c 100644
---
a/rt/transports/http/src/main/java/org/apache/cxf/transport/http/HttpClientHTTPConduit.java
+++
b/rt/transports/http/src/main/java/org/apache/cxf/transport/http/HttpClientHTTPConduit.java
@@ -1157,22 +1157,21 @@ public class HttpClientHTTPConduit extends
URLConnectionHTTPConduit {
future = cl.sendAsync(request, handler);
}
future.exceptionally(ex -> {
- // Record the failure on the message so the close() invoked
below
- // does not treat this as an un-sent request and resend it: the
- // shared WrappedOutputStream#close() already skips
- // handleHeadersTrustCaching() once outMessage carries an
Exception
- // (CXF-9250).
- outMessage.setContent(Exception.class,
- ex instanceof Exception ? (Exception) ex : new
IOException(ex));
- if (pout != null) {
- synchronized (pout) {
- pout.notifyAll();
+ // Only release a writer waiting for the connection or blocked
on the body. The
+ // caller still closes the stream, and the failure is reported
from there, by the
+ // response handling getting it from the future. Closing the
whole stream here
+ // would send a request without a body again, and recording
the failure on the
+ // message would keep the caller's close() from reporting it.
+ final OutputStream out = pout;
+ if (out != null) {
+ synchronized (out) {
+ out.notifyAll();
+ }
+ try {
+ out.close();
+ } catch (IOException e) {
+ ex.addSuppressed(e);
}
- }
- try {
- close();
- } catch (IOException e) {
- ex.addSuppressed(e);
}
return null;
});
diff --git
a/rt/transports/http/src/test/java/org/apache/cxf/transport/http/HttpClientHTTPConduitAsyncFailureTest.java
b/rt/transports/http/src/test/java/org/apache/cxf/transport/http/HttpClientHTTPConduitAsyncFailureTest.java
new file mode 100644
index 00000000000..74125489cdf
--- /dev/null
+++
b/rt/transports/http/src/test/java/org/apache/cxf/transport/http/HttpClientHTTPConduitAsyncFailureTest.java
@@ -0,0 +1,216 @@
+/**
+ * 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.cxf.transport.http;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.ConnectException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeMap;
+import java.util.TreeSet;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.cxf.Bus;
+import org.apache.cxf.bus.extension.ExtensionManagerBus;
+import org.apache.cxf.message.Exchange;
+import org.apache.cxf.message.ExchangeImpl;
+import org.apache.cxf.message.Message;
+import org.apache.cxf.message.MessageImpl;
+import org.apache.cxf.phase.PhaseInterceptorChain;
+import org.apache.cxf.service.model.EndpointInfo;
+import org.apache.cxf.testutil.common.TestUtil;
+import org.apache.cxf.transports.http.configuration.HTTPClientPolicy;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * An asynchronous call to an address that refuses the connection must report
the failure to the
+ * fault observer, also when the connection fails before the caller closes the
request stream.
+ */
+public class HttpClientHTTPConduitAsyncFailureTest {
+ private static final String PORT =
TestUtil.getPortNumber(HttpClientHTTPConduitAsyncFailureTest.class);
+
+ private Bus bus;
+ private final List<HttpClientHTTPConduit> conduits = new ArrayList<>();
+
+ @Before
+ public void setUp() {
+ bus = new ExtensionManagerBus();
+ }
+
+ @After
+ public void tearDown() {
+ conduits.forEach(HttpClientHTTPConduit::close);
+ bus.shutdown(true);
+ }
+
+ @Test
+ public void testConnectionRefusedBeforeCloseWithChunking() throws
Exception {
+ assertRefusedBeforeCloseReportedOnce(true);
+ }
+
+ @Test
+ public void testConnectionRefusedBeforeCloseWithoutChunking() throws
Exception {
+ assertRefusedBeforeCloseReportedOnce(false);
+ }
+
+ @Test
+ public void testConnectionRefusedWhileSendingOnClose() throws Exception {
+ HttpClientHTTPConduit conduit = newConduit(refusedAddress(), true);
+ Message message = newAsyncMessage();
+ // a body under the chunking threshold is sent when the stream is
closed: the connection is
+ // refused while the caller is still sending it, here before the trust
decision is made
+ message.put(MessageTrustDecider.class, new MessageTrustDecider() {
+ @Override
+ public void establishTrust(String conduitName, URLConnectionInfo
connectionInfo, Message m) {
+ awaitRefused(m);
+ }
+ });
+ conduit.prepare(message);
+ OutputStream os = message.getContent(OutputStream.class);
+
+ assertReportedOnce(message, () -> {
+ os.write(new byte[8]);
+ os.close();
+ });
+ }
+
+ private void assertRefusedBeforeCloseReportedOnce(boolean allowChunking)
throws Exception {
+ HttpClientHTTPConduit conduit = newConduit(refusedAddress(),
allowChunking);
+ Message message = newAsyncMessage();
+ conduit.prepare(message);
+ OutputStream os = message.getContent(OutputStream.class);
+
+ assertReportedOnce(message, () -> {
+ // more than the chunking threshold, so that the request is sent
before the stream is closed
+ os.write(new byte[1024]);
+ awaitRefused(message);
+ os.close();
+ });
+ }
+
+ /**
+ * The refused connection must be reported once: either to the caller, by
an exception from the
+ * stream, as the out chain then reports it, or else to the fault observer
+ */
+ private static void assertReportedOnce(Message message, IORunnable send)
throws Exception {
+ AtomicInteger faults = new AtomicInteger();
+ AtomicReference<Throwable> reported = new AtomicReference<>();
+ CountDownLatch faultReported = new CountDownLatch(1);
+
((PhaseInterceptorChain)message.getInterceptorChain()).setFaultObserver(m -> {
+ reported.compareAndSet(null, m.getContent(Exception.class));
+ faults.incrementAndGet();
+ faultReported.countDown();
+ });
+ int failures = 0;
+ try {
+ send.run();
+ } catch (IOException | RuntimeException ex) {
+ reported.compareAndSet(null, ex);
+ failures++;
+ }
+ if (failures == 0) {
+ assertTrue("the failure was not reported", faultReported.await(10,
TimeUnit.SECONDS));
+ }
+ Thread.sleep(500);
+ assertEquals("the failure was reported more than once", 1, failures +
faults.get());
+ assertTrue("the failure reported is not the refused connection: " +
reported.get(),
+ hasCause(reported.get(), ConnectException.class));
+ }
+
+ private static boolean hasCause(Throwable t, Class<? extends Throwable>
cls) {
+ for (Throwable c = t; c != null; c = c.getCause()) {
+ if (cls.isInstance(c)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ /** Waits until the connection of the request has been refused, and its
callbacks have run */
+ private static void awaitRefused(Message message) {
+ HttpClientHTTPConduit.HttpClientWrappedOutputStream os =
+
(HttpClientHTTPConduit.HttpClientWrappedOutputStream)message.getContent(OutputStream.class);
+ try {
+ long deadline = System.currentTimeMillis() + 10_000;
+ while (!os.future.isDone() && System.currentTimeMillis() <
deadline) {
+ Thread.sleep(10);
+ }
+ assertTrue("the connection was not refused",
os.future.isCompletedExceptionally());
+ Thread.sleep(200);
+ } catch (InterruptedException ex) {
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ private interface IORunnable {
+ void run() throws IOException;
+ }
+
+ private HttpClientHTTPConduit newConduit(String address, boolean
allowChunking) throws Exception {
+ EndpointInfo ei = new EndpointInfo();
+ ei.setAddress(address);
+ HttpClientHTTPConduit conduit = new HttpClientHTTPConduit(bus, ei,
null);
+ HTTPClientPolicy policy = new HTTPClientPolicy();
+ policy.setConnectionTimeout(3000);
+ policy.setReceiveTimeout(3000);
+ policy.setAllowChunking(allowChunking);
+ policy.setChunkingThreshold(16);
+ conduit.setClient(policy);
+ conduit.finalizeConfig();
+ conduits.add(conduit);
+ return conduit;
+ }
+
+ private Message newAsyncMessage() {
+ Message message = new MessageImpl();
+ Map<String, List<String>> headers = new
TreeMap<>(String.CASE_INSENSITIVE_ORDER);
+ headers.put("content-type", new ArrayList<>(List.of("text/xml")));
+ message.put(Message.PROTOCOL_HEADERS, headers);
+ message.put(Message.HTTP_REQUEST_METHOD, "POST");
+
+ Exchange exchange = new ExchangeImpl();
+ exchange.put(Bus.class, bus);
+ exchange.setSynchronous(false);
+ exchange.setOutMessage(message);
+ message.setExchange(exchange);
+
+ PhaseInterceptorChain chain = new PhaseInterceptorChain(new
TreeSet<>());
+ message.setInterceptorChain(chain);
+ // as the out chain does before it closes the stream, so that the
failure can unwind it
+ chain.doIntercept(message);
+ return message;
+ }
+
+ /** An address nothing listens on, so that the connection is refused */
+ private static String refusedAddress() {
+ return "http://localhost:"; + PORT + "/echo";
+ }
+}
