Client: Wrap synchronous exceptions

In the past the Low Level REST Client was super careful not to wrap
any exceptions that it throws from synchronous calls so that callers can
catch the exceptions and work with them. The trouble with that is that
the exceptions are originally thrown on the async thread pool and then
transfered back into calling thread. That means that the stack trace of
the exception doesn't have the calling method which is *super* *ultra*
confusing.

This change always wraps exceptions transfered from the async thread
pool so that the stack trace of the thrown exception contains the
caller's stack. It tries to preserve the type of the throw exception but
this is quite a fiddly thing to get right. We have to catch every type
of exception that we want to preserve, wrap with the same type and
rethrow. I've preserved the types of all exceptions that we had tests
mentioning but no other exceptions. The other exceptions are either
wrapped in `IOException` or `RuntimeException`.

Closes elastic#28399

Author: nik9000

client/rest/src/main/java/org/elasticsearch/client/ResponseException.java

@@ -39,6 +39,16 @@ public ResponseException(Response response) throws IOException {
         this.response = response;
     }
 
+    /**
+     * Wrap a {@linkplain ResponseException} with another one with the current
+     * stack trace. This is used during synchronous calls so that the caller
+     * ends up in the stack trace of the exception thrown.
+     */
+    ResponseException(ResponseException e) throws IOException {
+        super(e.getMessage(), e);
+        this.response = e.getResponse();
+    }
+
     private static String buildMessage(Response response) throws IOException {
         String message = String.format(Locale.ROOT,
             "method [%s], host [%s], URI [%s], status line [%s]",

client/rest/src/main/java/org/elasticsearch/client/RestClient.java

@@ -38,6 +38,7 @@
 import org.apache.http.client.protocol.HttpClientContext;
 import org.apache.http.client.utils.URIBuilder;
 import org.apache.http.concurrent.FutureCallback;
+import org.apache.http.conn.ConnectTimeoutException;
 import org.apache.http.impl.auth.BasicScheme;
 import org.apache.http.impl.client.BasicAuthCache;
 import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
@@ -47,6 +48,7 @@
 
 import java.io.Closeable;
 import java.io.IOException;
+import java.net.SocketTimeoutException;
 import java.net.URI;
 import java.net.URISyntaxException;
 import java.util.ArrayList;
@@ -218,7 +220,8 @@ public Response performRequest(String method, String endpoint, Map<String, Strin
                                    HttpEntity entity, HttpAsyncResponseConsumerFactory httpAsyncResponseConsumerFactory,
                                    Header... headers) throws IOException {
         SyncResponseListener listener = new SyncResponseListener(maxRetryTimeoutMillis);
-        performRequestAsync(method, endpoint, params, entity, httpAsyncResponseConsumerFactory, listener, headers);
+        performRequestAsyncNoCatch(method, endpoint, params, entity, httpAsyncResponseConsumerFactory,
+            listener, headers);
         return listener.get();
     }
 
@@ -293,43 +296,50 @@ public void performRequestAsync(String method, String endpoint, Map<String, Stri
                                     HttpEntity entity, HttpAsyncResponseConsumerFactory httpAsyncResponseConsumerFactory,
                                     ResponseListener responseListener, Header... headers) {
         try {
-            Objects.requireNonNull(params, "params must not be null");
-            Map<String, String> requestParams = new HashMap<>(params);
-            //ignore is a special parameter supported by the clients, shouldn't be sent to es
-            String ignoreString = requestParams.remove("ignore");
-            Set<Integer> ignoreErrorCodes;
-            if (ignoreString == null) {
-                if (HttpHead.METHOD_NAME.equals(method)) {
-                    //404 never causes error if returned for a HEAD request
-                    ignoreErrorCodes = Collections.singleton(404);
-                } else {
-                    ignoreErrorCodes = Collections.emptySet();
-                }
+            performRequestAsyncNoCatch(method, endpoint, params, entity, httpAsyncResponseConsumerFactory,
+                responseListener, headers);
+        } catch (Exception e) {
+            responseListener.onFailure(e);
+        }
+    }
+
+    void performRequestAsyncNoCatch(String method, String endpoint, Map<String, String> params,
+                                    HttpEntity entity, HttpAsyncResponseConsumerFactory httpAsyncResponseConsumerFactory,
+                                    ResponseListener responseListener, Header... headers) {
+        Objects.requireNonNull(params, "params must not be null");
+        Map<String, String> requestParams = new HashMap<>(params);
+        //ignore is a special parameter supported by the clients, shouldn't be sent to es
+        String ignoreString = requestParams.remove("ignore");
+        Set<Integer> ignoreErrorCodes;
+        if (ignoreString == null) {
+            if (HttpHead.METHOD_NAME.equals(method)) {
+                //404 never causes error if returned for a HEAD request
+                ignoreErrorCodes = Collections.singleton(404);
             } else {
-                String[] ignoresArray = ignoreString.split(",");
-                ignoreErrorCodes = new HashSet<>();
-                if (HttpHead.METHOD_NAME.equals(method)) {
-                    //404 never causes error if returned for a HEAD request
-                    ignoreErrorCodes.add(404);
-                }
-                for (String ignoreCode : ignoresArray) {
-                    try {
-                        ignoreErrorCodes.add(Integer.valueOf(ignoreCode));
-                    } catch (NumberFormatException e) {
-                        throw new IllegalArgumentException("ignore value should be a number, found [" + ignoreString + "] instead", e);
-                    }
+                ignoreErrorCodes = Collections.emptySet();
+            }
+        } else {
+            String[] ignoresArray = ignoreString.split(",");
+            ignoreErrorCodes = new HashSet<>();
+            if (HttpHead.METHOD_NAME.equals(method)) {
+                //404 never causes error if returned for a HEAD request
+                ignoreErrorCodes.add(404);
+            }
+            for (String ignoreCode : ignoresArray) {
+                try {
+                    ignoreErrorCodes.add(Integer.valueOf(ignoreCode));
+                } catch (NumberFormatException e) {
+                    throw new IllegalArgumentException("ignore value should be a number, found [" + ignoreString + "] instead", e);
                 }
             }
-            URI uri = buildUri(pathPrefix, endpoint, requestParams);
-            HttpRequestBase request = createHttpRequest(method, uri, entity);
-            setHeaders(request, headers);
-            FailureTrackingResponseListener failureTrackingResponseListener = new FailureTrackingResponseListener(responseListener);
-            long startTime = System.nanoTime();
-            performRequestAsync(startTime, nextHost(), request, ignoreErrorCodes, httpAsyncResponseConsumerFactory,
-                    failureTrackingResponseListener);
-        } catch (Exception e) {
-            responseListener.onFailure(e);
         }
+        URI uri = buildUri(pathPrefix, endpoint, requestParams);
+        HttpRequestBase request = createHttpRequest(method, uri, entity);
+        setHeaders(request, headers);
+        FailureTrackingResponseListener failureTrackingResponseListener = new FailureTrackingResponseListener(responseListener);
+        long startTime = System.nanoTime();
+        performRequestAsync(startTime, nextHost(), request, ignoreErrorCodes, httpAsyncResponseConsumerFactory,
+                failureTrackingResponseListener);
     }
 
     private void performRequestAsync(final long startTime, final HostTuple<Iterator<HttpHost>> hostTuple, final HttpRequestBase request,
@@ -674,12 +684,30 @@ Response get() throws IOException {
                     e.addSuppressed(exception);
                     throw e;
                 }
-                //try and leave the exception untouched as much as possible but we don't want to just add throws Exception clause everywhere
+                /*
+                 * Wrap and rethrow whatever exception we received, copying the type
+                 * where possible so the synchronous API looks as much as possible
+                 * like the asynchronous API. We wrap the exception so that the caller's
+                 * signature shows up in any exception we throw.
+                 */
+                if (exception instanceof ResponseException) {
+                    throw new ResponseException((ResponseException) exception);
+                }
+                if (exception instanceof ConnectTimeoutException) {
+                    ConnectTimeoutException e = new ConnectTimeoutException(exception.getMessage());
+                    e.initCause(exception);
+                    throw e;
+                }
+                if (exception instanceof SocketTimeoutException) {
+                    SocketTimeoutException e = new SocketTimeoutException(exception.getMessage());
+                    e.initCause(exception);
+                    throw e;
+                }
                 if (exception instanceof IOException) {
-                    throw (IOException) exception;
+                    throw new IOException(exception.getMessage(), exception);
                 }
                 if (exception instanceof RuntimeException){
-                    throw (RuntimeException) exception;
+                    throw new RuntimeException(exception.getMessage(), exception);
                 }
                 throw new RuntimeException("error while performing request", exception);
             }

client/rest/src/test/java/org/elasticsearch/client/RestClientMultipleHostsTests.java

@@ -35,6 +35,7 @@
 import org.apache.http.message.BasicStatusLine;
 import org.apache.http.nio.protocol.HttpAsyncRequestProducer;
 import org.apache.http.nio.protocol.HttpAsyncResponseConsumer;
+import org.junit.After;
 import org.junit.Before;
 import org.mockito.invocation.InvocationOnMock;
 import org.mockito.stubbing.Answer;
@@ -44,6 +45,8 @@
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.Set;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 
 import static org.elasticsearch.client.RestClientTestUtil.randomErrorNoRetryStatusCode;
@@ -66,6 +69,7 @@
  */
 public class RestClientMultipleHostsTests extends RestClientTestCase {
 
+    private ExecutorService exec = Executors.newFixedThreadPool(1);
     private RestClient restClient;
     private HttpHost[] httpHosts;
     private HostsTrackingFailureListener failureListener;
@@ -79,23 +83,28 @@ public void createRestClient() throws IOException {
             @Override
             public Future<HttpResponse> answer(InvocationOnMock invocationOnMock) throws Throwable {
                 HttpAsyncRequestProducer requestProducer = (HttpAsyncRequestProducer) invocationOnMock.getArguments()[0];
-                HttpUriRequest request = (HttpUriRequest)requestProducer.generateRequest();
-                HttpHost httpHost = requestProducer.getTarget();
+                final HttpUriRequest request = (HttpUriRequest)requestProducer.generateRequest();
+                final HttpHost httpHost = requestProducer.getTarget();
                 HttpClientContext context = (HttpClientContext) invocationOnMock.getArguments()[2];
                 assertThat(context.getAuthCache().get(httpHost), instanceOf(BasicScheme.class));
-                FutureCallback<HttpResponse> futureCallback = (FutureCallback<HttpResponse>) invocationOnMock.getArguments()[3];
+                final FutureCallback<HttpResponse> futureCallback = (FutureCallback<HttpResponse>) invocationOnMock.getArguments()[3];
                 //return the desired status code or exception depending on the path
-                if (request.getURI().getPath().equals("/soe")) {
-                    futureCallback.failed(new SocketTimeoutException(httpHost.toString()));
-                } else if (request.getURI().getPath().equals("/coe")) {
-                    futureCallback.failed(new ConnectTimeoutException(httpHost.toString()));
-                } else if (request.getURI().getPath().equals("/ioe")) {
-                    futureCallback.failed(new IOException(httpHost.toString()));
-                } else {
-                    int statusCode = Integer.parseInt(request.getURI().getPath().substring(1));
-                    StatusLine statusLine = new BasicStatusLine(new ProtocolVersion("http", 1, 1), statusCode, "");
-                    futureCallback.completed(new BasicHttpResponse(statusLine));
-                }
+                exec.execute(new Runnable() {
+                    @Override
+                    public void run() {
+                        if (request.getURI().getPath().equals("/soe")) {
+                            futureCallback.failed(new SocketTimeoutException(httpHost.toString()));
+                        } else if (request.getURI().getPath().equals("/coe")) {
+                            futureCallback.failed(new ConnectTimeoutException(httpHost.toString()));
+                        } else if (request.getURI().getPath().equals("/ioe")) {
+                            futureCallback.failed(new IOException(httpHost.toString()));
+                        } else {
+                            int statusCode = Integer.parseInt(request.getURI().getPath().substring(1));
+                            StatusLine statusLine = new BasicStatusLine(new ProtocolVersion("http", 1, 1), statusCode, "");
+                            futureCallback.completed(new BasicHttpResponse(statusLine));
+                        }
+                    }
+                });
                 return null;
             }
         });
@@ -108,6 +117,14 @@ public Future<HttpResponse> answer(InvocationOnMock invocationOnMock) throws Thr
         restClient = new RestClient(httpClient, 10000, new Header[0], httpHosts, null, failureListener);
     }
 
+    /**
+     * Shutdown the executor so we don't leak threads into other test runs.
+     */
+    @After
+    public void shutdownExec() {
+        exec.shutdown();
+    }
+
     public void testRoundRobinOkStatusCodes() throws IOException {
         int numIters = RandomNumbers.randomIntBetween(getRandom(), 1, 5);
         for (int i = 0; i < numIters; i++) {
@@ -163,6 +180,11 @@ public void testRoundRobinRetryErrors() throws IOException {
             restClient.performRequest(randomHttpMethod(getRandom()), retryEndpoint);
             fail("request should have failed");
         } catch(ResponseException e) {
+            /*
+             * Unwrap the top level failure that was added so the stack trace contains
+             * the caller. It wraps the exception that contains the failed hosts.
+             */
+            e = (ResponseException) e.getCause();
             Set<HttpHost> hostsSet = new HashSet<>();
             Collections.addAll(hostsSet, httpHosts);
             //first request causes all the hosts to be blacklisted, the returned exception holds one suppressed exception each
@@ -183,6 +205,11 @@ public void testRoundRobinRetryErrors() throws IOException {
             } while(e != null);
             assertEquals("every host should have been used but some weren't: " + hostsSet, 0, hostsSet.size());
         } catch(IOException e) {
+            /*
+             * Unwrap the top level failure that was added so the stack trace contains
+             * the caller. It wraps the exception that contains the failed hosts.
+             */
+            e = (IOException) e.getCause();
             Set<HttpHost> hostsSet = new HashSet<>();
             Collections.addAll(hostsSet, httpHosts);
             //first request causes all the hosts to be blacklisted, the returned exception holds one suppressed exception each
@@ -221,6 +248,11 @@ public void testRoundRobinRetryErrors() throws IOException {
                     failureListener.assertCalled(response.getHost());
                     assertEquals(0, e.getSuppressed().length);
                 } catch(IOException e) {
+                    /*
+                     * Unwrap the top level failure that was added so the stack trace contains
+                     * the caller. It wraps the exception that contains the failed hosts.
+                     */
+                    e = (IOException) e.getCause();
                     HttpHost httpHost = HttpHost.create(e.getMessage());
                     assertTrue("host [" + httpHost + "] not found, most likely used multiple times", hostsSet.remove(httpHost));
                     //after the first request, all hosts are blacklisted, a single one gets resurrected each time
@@ -263,6 +295,11 @@ public void testRoundRobinRetryErrors() throws IOException {
                         assertThat(response.getHost(), equalTo(selectedHost));
                         failureListener.assertCalled(selectedHost);
                     } catch(IOException e) {
+                        /*
+                         * Unwrap the top level failure that was added so the stack trace contains
+                         * the caller. It wraps the exception that contains the failed hosts.
+                         */
+                        e = (IOException) e.getCause();
                         HttpHost httpHost = HttpHost.create(e.getMessage());
                         assertThat(httpHost, equalTo(selectedHost));
                         failureListener.assertCalled(selectedHost);