Client: Wrap synchronous exceptions (#28919)

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 transferred 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 #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

@@ -20,6 +20,7 @@
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.apache.http.ConnectionClosedException;
 import org.apache.http.Header;
 import org.apache.http.HttpEntity;
 import org.apache.http.HttpHost;
@@ -38,6 +39,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 +49,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;
@@ -201,6 +204,14 @@ public Response performRequest(String method, String endpoint, Map<String, Strin
      * they previously failed (the more failures, the later they will be retried). In case of failures all of the alive nodes (or dead
      * nodes that deserve a retry) are retried until one responds or none of them does, in which case an {@link IOException} will be thrown.
      *
+     * This method works by performing an asynchronous call and waiting
+     * for the result. If the asynchronous call throws an exception we wrap
+     * it and rethrow it so that the stack trace attached to the exception
+     * contains the call site. While we attempt to preserve the original
+     * exception this isn't always possible and likely haven't covered all of
+     * the cases. You can get the original exception from
+     * {@link Exception#getCause()}.
+     *
      * @param method the http method
      * @param endpoint the path of the request (without host and port)
      * @param params the query_string parameters
@@ -218,7 +229,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 +305,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 +693,35 @@ 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 ConnectionClosedException) {
+                    ConnectionClosedException e = new ConnectionClosedException(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);
             }