This is an automated email from the ASF dual-hosted git repository. reta pushed a commit to branch 4.1.x-fixes in repository https://gitbox.apache.org/repos/asf/cxf.git
commit 9c6072ea097b4e68035c57620b053e35ae6991ba Author: Markus Heiden <[email protected]> AuthorDate: Sun Oct 11 20:00:00 2026 +0200 CXF-9257: Complete client futures if the async callback throws (#3565) * Complete client futures if the async callback throws JaxrsClientCallback and JaxwsClientCallback invoke the user's callback before completing their future. If the callback throws, the exception escapes and the future is never completed, so threads blocked in Future.get() hang forever. In the JAX-WS case ClientImpl additionally catches the exception and invokes the same callback a second time. Guard every callback invocation: - If the callback throws while handling a successful response, the future completes exceptionally with that exception, like the JAX-WS reference implementation does. - If the callback throws while handling a failure, cancellation or interruption, the original exception is kept and the callback's exception is attached as a suppressed exception. Co-Authored-By: Claude Fable 5.1 <[email protected]> Claude-Session: https://claude.ai/code/session_0149ipFQyLFL9z5WdmhCRg3r * Follow the callback contract instead of guarding handleResponse CXF callers invoke ClientCallback.handleException() if handleResponse() throws, so the callbacks don't need to guard the success path. Revert those guards and fix the call sites that don't follow the contract: ClientImpl's fault observer wrapper and WebClient's ClientAsyncResponseInterceptor. The failure paths stay guarded in the callbacks, because callers can't complete the future if handleException() throws. Don't add a callback exception as suppressed to itself, which would throw IllegalArgumentException if the callback rethrows the exception it was given. Co-Authored-By: Claude Opus 5.5 <[email protected]> Claude-Session: https://claude.ai/code/session_01Ka6VjhpAcDrSERoiLP2Cxx --------- Co-authored-by: Claude Fable 5.1 <[email protected]> (cherry picked from commit cca9715fb00577b31a74a85f2fabbc8feff08bb5) --- .../java/org/apache/cxf/endpoint/ClientImpl.java | 6 +- .../org/apache/cxf/jaxws/JaxwsClientCallback.java | 69 ++++++++++++---------- .../apache/cxf/jaxws/JaxwsClientCallbackTest.java | 53 +++++++++++++++++ .../cxf/jaxrs/client/JaxrsClientCallback.java | 31 ++++++++-- .../org/apache/cxf/jaxrs/client/WebClient.java | 15 ++++- .../cxf/jaxrs/client/JaxrsClientCallbackTest.java | 69 ++++++++++++++++++++++ .../cxf/systest/jaxrs/JAXRSAsyncClientTest.java | 22 +++++++ 7 files changed, 226 insertions(+), 39 deletions(-) diff --git a/core/src/main/java/org/apache/cxf/endpoint/ClientImpl.java b/core/src/main/java/org/apache/cxf/endpoint/ClientImpl.java index 17fdc79b400..35803e1c4f2 100644 --- a/core/src/main/java/org/apache/cxf/endpoint/ClientImpl.java +++ b/core/src/main/java/org/apache/cxf/endpoint/ClientImpl.java @@ -517,7 +517,11 @@ public class ClientImpl Map<String, Object> ctx = responseContext.get(Thread.currentThread()); List<Object> resList = CastUtils.cast(inMsg.getContent(List.class)); Object[] result = resList == null ? null : resList.toArray(); - callback.handleResponse(ctx, result); + try { + callback.handleResponse(ctx, result); + } catch (Throwable t) { + callback.handleException(ctx, t); + } return; } } diff --git a/rt/frontend/jaxws/src/main/java/org/apache/cxf/jaxws/JaxwsClientCallback.java b/rt/frontend/jaxws/src/main/java/org/apache/cxf/jaxws/JaxwsClientCallback.java index c73da2d9f3f..bdae53f0214 100644 --- a/rt/frontend/jaxws/src/main/java/org/apache/cxf/jaxws/JaxwsClientCallback.java +++ b/rt/frontend/jaxws/src/main/java/org/apache/cxf/jaxws/JaxwsClientCallback.java @@ -86,41 +86,48 @@ class JaxwsClientCallback<T> extends ClientCallback { public void handleException(Map<String, Object> ctx, final Throwable ex) { context = ctx; + // The handler has to be called *before* future completes if (handler != null) { - handler.handleResponse(new Response<T>() { - protected boolean cancelled; - - public Map<String, Object> getContext() { - return context; - } - - public boolean cancel(boolean mayInterruptIfRunning) { - cancelled = true; - return true; - } - - public T get() throws InterruptedException, ExecutionException { - throw new ExecutionException(ex); + try { + handler.handleResponse(new Response<T>() { + protected boolean cancelled; + + public Map<String, Object> getContext() { + return context; + } + + public boolean cancel(boolean mayInterruptIfRunning) { + cancelled = true; + return true; + } + + public T get() throws InterruptedException, ExecutionException { + throw new ExecutionException(ex); + } + + public T get(long timeout, TimeUnit unit) + throws InterruptedException, ExecutionException, TimeoutException { + + throw new ExecutionException(ex); + } + + public boolean isCancelled() { + return cancelled; + } + + public boolean isDone() { + return true; + } + + }); + } catch (Throwable t) { + // The future has to complete even if the handler fails, otherwise waiting threads hang forever + if (t != ex) { + ex.addSuppressed(t); } - - public T get(long timeout, TimeUnit unit) - throws InterruptedException, ExecutionException, TimeoutException { - - throw new ExecutionException(ex); - } - - public boolean isCancelled() { - return cancelled; - } - - public boolean isDone() { - return true; - } - - }); + } } - // The handler has to be called *before* future completes delegate.completeExceptionally(mapThrowable(ex)); synchronized (this) { diff --git a/rt/frontend/jaxws/src/test/java/org/apache/cxf/jaxws/JaxwsClientCallbackTest.java b/rt/frontend/jaxws/src/test/java/org/apache/cxf/jaxws/JaxwsClientCallbackTest.java index 73380b40ceb..8bbfbe9c4b3 100644 --- a/rt/frontend/jaxws/src/test/java/org/apache/cxf/jaxws/JaxwsClientCallbackTest.java +++ b/rt/frontend/jaxws/src/test/java/org/apache/cxf/jaxws/JaxwsClientCallbackTest.java @@ -37,6 +37,7 @@ import org.junit.Before; import org.junit.Test; import static org.hamcrest.CoreMatchers.equalTo; +import static org.hamcrest.CoreMatchers.sameInstance; import static org.hamcrest.MatcherAssert.assertThat; import static org.junit.Assert.assertThrows; @@ -151,6 +152,58 @@ public class JaxwsClientCallbackTest { callback.get(10, TimeUnit.MILLISECONDS); } + @Test + public void testHandleResponseCallbackThrows() throws Exception { + final RuntimeException handlerException = new RuntimeException("handler failed"); + useThrowingHandler(handlerException); + + final CyclicBarrier barrier = new CyclicBarrier(2); + Object[] result = new Object[0]; + // Callers invoke handleException() if handleResponse() throws + schedule(barrier, () -> { + try { + callback.handleResponse(ctx, result); + } catch (Throwable t) { + callback.handleException(ctx, t); + } + }); + barrier.await(5, TimeUnit.SECONDS); + + ExecutionException ex = assertThrows(ExecutionException.class, () -> callback.get(5, TimeUnit.SECONDS)); + assertThat(ex.getCause(), sameInstance(handlerException)); + assertThat(handlerException.getSuppressed().length, equalTo(0)); + assertThrows(ExecutionException.class, () -> callback.get()); + assertThat(callback.isCancelled(), equalTo(false)); + assertThat(callback.isDone(), equalTo(true)); + } + + @Test + public void testHandleExceptionCallbackThrows() throws Exception { + final RuntimeException handlerException = new RuntimeException("handler failed"); + useThrowingHandler(handlerException); + + final RuntimeException cause = new RuntimeException("invocation failed"); + final CyclicBarrier barrier = new CyclicBarrier(2); + schedule(barrier, () -> callback.handleException(ctx, cause)); + barrier.await(5, TimeUnit.SECONDS); + + ExecutionException ex = assertThrows(ExecutionException.class, () -> callback.get(5, TimeUnit.SECONDS)); + assertThat(ex.getCause(), sameInstance(cause)); + assertThat(cause.getSuppressed(), equalTo(new Throwable[] {handlerException})); + assertThat(callback.isCancelled(), equalTo(false)); + assertThat(callback.isDone(), equalTo(true)); + } + + private void useThrowingHandler(final RuntimeException handlerException) { + handler = new AsyncHandler<String>() { + @Override + public void handleResponse(Response<String> res) { + throw handlerException; + } + }; + callback = new JaxwsClientCallback<String>(handler, null); + } + private void schedule(CyclicBarrier barrier, Runnable runnable) { executor.schedule(() -> { barrier.await(5, TimeUnit.SECONDS); diff --git a/rt/rs/client/src/main/java/org/apache/cxf/jaxrs/client/JaxrsClientCallback.java b/rt/rs/client/src/main/java/org/apache/cxf/jaxrs/client/JaxrsClientCallback.java index 0532270a450..b1a590e299b 100644 --- a/rt/rs/client/src/main/java/org/apache/cxf/jaxrs/client/JaxrsClientCallback.java +++ b/rt/rs/client/src/main/java/org/apache/cxf/jaxrs/client/JaxrsClientCallback.java @@ -59,7 +59,11 @@ public class JaxrsClientCallback<T> extends ClientCallback { if (!started) { // The handler has to be called *before* future completes if (handler != null) { - handler.failed(new CancellationException()); + try { + handler.failed(new CancellationException()); + } catch (Throwable t) { + // The future has to be cancelled even if the handler fails + } } delegate.cancel(mayInterruptIfRunning); @@ -98,7 +102,14 @@ public class JaxrsClientCallback<T> extends ClientCallback { // The handler has to be called *before* future completes if (handler != null) { - handler.failed(ex); + try { + handler.failed(ex); + } catch (Throwable t) { + // The future has to complete even if the handler fails, otherwise waiting threads hang forever + if (t != ex) { + ex.addSuppressed(t); + } + } } delegate.completeExceptionally(ex); @@ -131,7 +142,13 @@ public class JaxrsClientCallback<T> extends ClientCallback { return getObject(callback.get()[0]); } catch (InterruptedException ex) { if (callback.handler != null) { - callback.handler.failed(ex); + try { + callback.handler.failed(ex); + } catch (Throwable t) { + if (t != ex) { + ex.addSuppressed(t); + } + } } throw ex; } @@ -142,7 +159,13 @@ public class JaxrsClientCallback<T> extends ClientCallback { return getObject(callback.get(timeout, unit)[0]); } catch (InterruptedException ex) { if (callback.handler != null) { - callback.handler.failed(ex); + try { + callback.handler.failed(ex); + } catch (Throwable t) { + if (t != ex) { + ex.addSuppressed(t); + } + } } throw ex; } diff --git a/rt/rs/client/src/main/java/org/apache/cxf/jaxrs/client/WebClient.java b/rt/rs/client/src/main/java/org/apache/cxf/jaxrs/client/WebClient.java index 74f79e11681..ef57df668a5 100644 --- a/rt/rs/client/src/main/java/org/apache/cxf/jaxrs/client/WebClient.java +++ b/rt/rs/client/src/main/java/org/apache/cxf/jaxrs/client/WebClient.java @@ -1033,12 +1033,21 @@ public class WebClient extends AbstractClient { } } if (cb.getResponseClass() == null || Response.class.equals(cb.getResponseClass())) { - cb.handleResponse(message, new Object[] {r}); + try { + cb.handleResponse(message, new Object[] {r}); + } catch (Throwable t) { + cb.handleException(message, t); + } } else if (r.getStatus() >= 300) { cb.handleException(message, convertToWebApplicationException(r)); } else { - cb.handleResponse(message, new Object[] {r.getEntity()}); - closeAsyncResponseIfPossible(r, message, cb); + try { + cb.handleResponse(message, new Object[] {r.getEntity()}); + } catch (Throwable t) { + cb.handleException(message, t); + } finally { + closeAsyncResponseIfPossible(r, message, cb); + } } } } diff --git a/rt/rs/client/src/test/java/org/apache/cxf/jaxrs/client/JaxrsClientCallbackTest.java b/rt/rs/client/src/test/java/org/apache/cxf/jaxrs/client/JaxrsClientCallbackTest.java index 4653414c22f..94cfe6094d6 100644 --- a/rt/rs/client/src/test/java/org/apache/cxf/jaxrs/client/JaxrsClientCallbackTest.java +++ b/rt/rs/client/src/test/java/org/apache/cxf/jaxrs/client/JaxrsClientCallbackTest.java @@ -41,6 +41,7 @@ import org.junit.Test; import static org.hamcrest.CoreMatchers.equalTo; import static org.hamcrest.CoreMatchers.instanceOf; import static org.hamcrest.CoreMatchers.nullValue; +import static org.hamcrest.CoreMatchers.sameInstance; import static org.hamcrest.MatcherAssert.assertThat; import static org.junit.Assert.assertThrows; @@ -182,6 +183,74 @@ public class JaxrsClientCallbackTest { future.get(10, TimeUnit.MILLISECONDS); } + @Test + public void testHandleResponseCallbackThrows() throws Exception { + final RuntimeException handlerException = new RuntimeException("completed failed"); + useThrowingHandler(handlerException); + + final CyclicBarrier barrier = new CyclicBarrier(2); + Object[] result = new String[] {"results"}; + // Callers invoke handleException() if handleResponse() throws + schedule(barrier, () -> { + try { + callback.handleResponse(ctx, result); + } catch (Throwable t) { + callback.handleException(ctx, t); + } + }); + barrier.await(5, TimeUnit.SECONDS); + + ExecutionException ex = assertThrows(ExecutionException.class, () -> future.get(5, TimeUnit.SECONDS)); + assertThat(ex.getCause(), sameInstance(handlerException)); + assertThat(handlerException.getSuppressed().length, equalTo(0)); + assertThrows(ExecutionException.class, () -> future.get()); + assertThat(future.isCancelled(), equalTo(false)); + assertThat(future.isDone(), equalTo(true)); + } + + @Test + public void testHandleExceptionCallbackThrows() throws Exception { + final RuntimeException handlerException = new RuntimeException("failed failed"); + useThrowingHandler(handlerException); + + final RuntimeException cause = new RuntimeException("invocation failed"); + final CyclicBarrier barrier = new CyclicBarrier(2); + schedule(barrier, () -> callback.handleException(ctx, cause)); + barrier.await(5, TimeUnit.SECONDS); + + ExecutionException ex = assertThrows(ExecutionException.class, () -> future.get(5, TimeUnit.SECONDS)); + assertThat(ex.getCause(), sameInstance(cause)); + assertThat(cause.getSuppressed(), equalTo(new Throwable[] {handlerException})); + assertThat(future.isCancelled(), equalTo(false)); + assertThat(future.isDone(), equalTo(true)); + } + + @Test + public void testHandleCancellationCallbackThrows() throws Exception { + useThrowingHandler(new RuntimeException("failed failed")); + + assertThat(future.cancel(true), equalTo(true)); + assertThrows(InterruptedException.class, () -> future.get(10, TimeUnit.MILLISECONDS)); + assertThat(future.isCancelled(), equalTo(true)); + assertThat(future.isDone(), equalTo(true)); + } + + private void useThrowingHandler(final RuntimeException handlerException) { + handler = new InvocationCallback<String>() { + @Override + public void failed(Throwable throwable) { + throw handlerException; + } + + @Override + public void completed(String response) { + throw handlerException; + } + }; + callback = new JaxrsClientCallback<String>(handler, String.class, null); + future = (JaxrsResponseFuture<String>)callback.createFuture(); + } + private void schedule(CyclicBarrier barrier, Runnable runnable) { executor.schedule(() -> { barrier.await(5, TimeUnit.SECONDS); diff --git a/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/JAXRSAsyncClientTest.java b/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/JAXRSAsyncClientTest.java index f56a9ee2b3c..c45661e0c56 100644 --- a/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/JAXRSAsyncClientTest.java +++ b/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/JAXRSAsyncClientTest.java @@ -410,6 +410,28 @@ public class JAXRSAsyncClientTest extends AbstractBusClientServerTestBase { wc.close(); } + @Test + public void testGetBookAsyncCallbackThrows() throws Exception { + String address = "http://localhost:"; + PORT + "/bookstore/books/123"; + WebClient wc = createWebClient(address); + final RuntimeException handlerException = new RuntimeException("completed failed"); + InvocationCallback<Book> callback = new InvocationCallback<Book>() { + public void completed(Book response) { + throw handlerException; + } + + public void failed(Throwable error) { + } + }; + try { + wc.async().get(callback).get(10, TimeUnit.SECONDS); + fail("Exception expected"); + } catch (ExecutionException ex) { + assertTrue(ex.getCause() == handlerException); + } + wc.close(); + } + @Test public void testClientResponseFilter() throws Exception { final String address = "http://localhost:"; + PORT + "/bookstore/books/wildcard";
