CXF-9257: Complete client futures if the async callback throws

core/src/main/java/org/apache/cxf/endpoint/ClientImpl.java

@@ -517,7 +517,11 @@ public void onMessage(Message message) {
                                 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;
                             }
                         }

rt/frontend/jaxws/src/main/java/org/apache/cxf/jaxws/JaxwsClientCallback.java

@@ -86,41 +86,48 @@ public boolean isDone() {
     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) {

rt/frontend/jaxws/src/test/java/org/apache/cxf/jaxws/JaxwsClientCallbackTest.java

@@ -37,6 +37,7 @@
 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 void testTimeout() throws Exception {
         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);

rt/rs/client/src/main/java/org/apache/cxf/jaxrs/client/JaxrsClientCallback.java

@@ -59,7 +59,11 @@ public boolean cancel(boolean mayInterruptIfRunning) {
         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 void handleException(Map<String, Object> ctx, final Throwable ex) {
 
         // 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 T get() throws InterruptedException, ExecutionException {
                 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 T get(long timeout, TimeUnit unit) throws InterruptedException,
                 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;
             }

rt/rs/client/src/main/java/org/apache/cxf/jaxrs/client/WebClient.java

@@ -1033,12 +1033,21 @@ protected void doHandleAsyncResponse(Message message, Response r, JaxrsClientCal
                 }
             }
             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);
+                }
             }
         }
     }

rt/rs/client/src/test/java/org/apache/cxf/jaxrs/client/JaxrsClientCallbackTest.java

@@ -41,6 +41,7 @@
 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 void testTimeout() throws Exception {
         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);

systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/JAXRSAsyncClientTest.java

@@ -410,6 +410,28 @@ public void testGetBookAsync404Callback() throws Exception {
         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";