This is an automated email from the ASF dual-hosted git repository. reta pushed a commit to branch 3.6.x-fixes in repository https://gitbox.apache.org/repos/asf/cxf.git
commit 4da41767d5eb32956e7071184bab8bbea0fef1fb 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 72097949a49..521985c66b0 100644 --- a/core/src/main/java/org/apache/cxf/endpoint/ClientImpl.java +++ b/core/src/main/java/org/apache/cxf/endpoint/ClientImpl.java @@ -514,7 +514,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 8f1ed9c22b1..7a4722aef88 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 @@ -87,41 +87,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 1e867038c47..ab753bf8dbb 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 @@ -38,6 +38,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; @@ -152,6 +153,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 5c123cd4a13..c40aaea61c0 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 @@ -60,7 +60,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); @@ -99,7 +103,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); @@ -132,7 +143,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; } @@ -143,7 +160,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 d650e18bf6b..d19183bd529 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 @@ -1034,12 +1034,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 e7618b10b05..90a226cb082 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 @@ -42,6 +42,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; @@ -183,6 +184,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 fd238515fe5..68877ff323c 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 @@ -411,6 +411,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";
