[Backport 2.x] Recreate or wrap exceptions thrown by async transport implementations to preserve caller stack traces

CHANGELOG.md

@@ -14,6 +14,7 @@ Inspired from [Keep a Changelog](https://keepachangelog.com/en/1.0.0/)
 
 ### Fixed
 - ApacheHttpClient5Transport requires Apache Commons Logging dependency ([#1003](https://github.com/opensearch-project/opensearch-java/pull/1003))
+- Preserve caller information in stack traces when synchronous callers use asynchronous transports ([#656](https://github.com/opensearch-project/opensearch-java/pull/656))
 
 ### Security
 

java-client/src/main/java/org/opensearch/client/transport/aws/AwsSdk2Transport.java

@@ -14,6 +14,8 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.UnsupportedEncodingException;
+import java.net.ConnectException;
+import java.net.SocketTimeoutException;
 import java.net.URI;
 import java.net.URISyntaxException;
 import java.net.URLEncoder;
@@ -30,12 +32,14 @@
 import javax.annotation.CheckForNull;
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
+import javax.net.ssl.SSLHandshakeException;
 import org.opensearch.client.json.JsonpDeserializer;
 import org.opensearch.client.json.JsonpMapper;
 import org.opensearch.client.json.jackson.JacksonJsonpMapper;
 import org.opensearch.client.opensearch._types.ErrorCause;
 import org.opensearch.client.opensearch._types.ErrorResponse;
 import org.opensearch.client.opensearch._types.OpenSearchException;
+import org.opensearch.client.opensearch.generic.OpenSearchClientException;
 import org.opensearch.client.transport.Endpoint;
 import org.opensearch.client.transport.GenericEndpoint;
 import org.opensearch.client.transport.JsonEndpoint;
@@ -199,17 +203,14 @@ public <RequestT, ResponseT, ErrorT> ResponseT performRequest(
             try {
                 return executeAsync((SdkAsyncHttpClient) httpClient, clientReq, requestBody, endpoint, options).get();
             } catch (ExecutionException e) {
-                Throwable cause = e.getCause();
-                if (cause != null) {
-                    if (cause instanceof IOException) {
-                        throw (IOException) cause;
-                    }
-                    if (cause instanceof RuntimeException) {
-                        throw (RuntimeException) cause;
-                    }
-                    throw new RuntimeException(cause);
+                Exception cause = extractAndWrapCause(e);
+                if (cause instanceof IOException) {
+                    throw (IOException) cause;
+                }
+                if (cause instanceof RuntimeException) {
+                    throw (RuntimeException) cause;
                 }
-                throw new RuntimeException(e);
+                throw new IllegalStateException("unexpected exception type: must be either RuntimeException or IOException", cause);
             } catch (InterruptedException e) {
                 throw new IOException("HttpRequest was interrupted", e);
             }
@@ -615,4 +616,56 @@ private static <T> Optional<T> or(Optional<T> opt, Supplier<? extends Optional<?
             return Objects.requireNonNull(r);
         }
     }
+
+    /**
+     * Wrap the exception so the caller's signature shows up in the stack trace, taking care to copy the original type and message
+     * where possible so async and sync code don't have to check different exceptions.
+     */
+    private static Exception extractAndWrapCause(Exception exception) {
+        if (exception instanceof InterruptedException) {
+            throw new RuntimeException("thread waiting for the response was interrupted", exception);
+        }
+        if (exception instanceof ExecutionException) {
+            ExecutionException executionException = (ExecutionException) exception;
+            Throwable t = executionException.getCause() == null ? executionException : executionException.getCause();
+            if (t instanceof Error) {
+                throw (Error) t;
+            }
+            exception = (Exception) t;
+        }
+        if (exception instanceof SocketTimeoutException) {
+            SocketTimeoutException e = new SocketTimeoutException(exception.getMessage());
+            e.initCause(exception);
+            return e;
+        }
+        if (exception instanceof SSLHandshakeException) {
+            SSLHandshakeException e = new SSLHandshakeException(
+                exception.getMessage() + "\nSee https://opensearch.org/docs/latest/clients/java/ for troubleshooting."
+            );
+            e.initCause(exception);
+            return e;
+        }
+        if (exception instanceof ConnectException) {
+            ConnectException e = new ConnectException(exception.getMessage());
+            e.initCause(exception);
+            return e;
+        }
+        if (exception instanceof IOException) {
+            return new IOException(exception.getMessage(), exception);
+        }
+        if (exception instanceof OpenSearchException) {
+            final OpenSearchException e = new OpenSearchException(((OpenSearchException) exception).response());
+            e.initCause(exception);
+            return e;
+        }
+        if (exception instanceof OpenSearchClientException) {
+            final OpenSearchClientException e = new OpenSearchClientException(((OpenSearchClientException) exception).response());
+            e.initCause(exception);
+            return e;
+        }
+        if (exception instanceof RuntimeException) {
+            return new RuntimeException(exception.getMessage(), exception);
+        }
+        return new RuntimeException("error while performing request", exception);
+    }
 }

java-client/src/main/java/org/opensearch/client/transport/httpclient5/ApacheHttpClient5Transport.java

@@ -15,6 +15,8 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
+import java.net.ConnectException;
+import java.net.SocketTimeoutException;
 import java.net.URI;
 import java.net.URISyntaxException;
 import java.util.AbstractMap;
@@ -36,13 +38,16 @@
 import java.util.concurrent.CompletionException;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
 import java.util.zip.GZIPOutputStream;
 import javax.annotation.Nullable;
+import javax.net.ssl.SSLHandshakeException;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.apache.hc.client5.http.ConnectTimeoutException;
 import org.apache.hc.client5.http.auth.AuthCache;
 import org.apache.hc.client5.http.auth.AuthScheme;
 import org.apache.hc.client5.http.auth.AuthScope;
@@ -57,6 +62,7 @@
 import org.apache.hc.client5.http.protocol.HttpClientContext;
 import org.apache.hc.core5.concurrent.FutureCallback;
 import org.apache.hc.core5.http.ClassicHttpResponse;
+import org.apache.hc.core5.http.ConnectionClosedException;
 import org.apache.hc.core5.http.ContentType;
 import org.apache.hc.core5.http.Header;
 import org.apache.hc.core5.http.HttpEntity;
@@ -77,6 +83,8 @@
 import org.opensearch.client.json.JsonpDeserializer;
 import org.opensearch.client.json.JsonpMapper;
 import org.opensearch.client.json.NdJsonpSerializable;
+import org.opensearch.client.opensearch._types.OpenSearchException;
+import org.opensearch.client.opensearch.generic.OpenSearchClientException;
 import org.opensearch.client.transport.Endpoint;
 import org.opensearch.client.transport.GenericEndpoint;
 import org.opensearch.client.transport.GenericSerializable;
@@ -145,15 +153,16 @@ public <RequestT, ResponseT, ErrorT> ResponseT performRequest(
         TransportOptions options
     ) throws IOException {
         try {
-            return performRequestAsync(request, endpoint, options).join();
-        } catch (final CompletionException ex) {
-            if (ex.getCause() instanceof RuntimeException) {
-                throw (RuntimeException) ex.getCause();
-            } else if (ex.getCause() instanceof IOException) {
-                throw (IOException) ex.getCause();
-            } else {
-                throw new IOException(ex.getCause());
+            return performRequestAsync(request, endpoint, options).get();
+        } catch (final Exception ex) {
+            Exception cause = extractAndWrapCause(ex);
+            if (cause instanceof IOException) {
+                throw (IOException) cause;
+            }
+            if (cause instanceof RuntimeException) {
+                throw (RuntimeException) cause;
             }
+            throw new IllegalStateException("unexpected exception type: must be either RuntimeException or IOException", cause);
         }
     }
 
@@ -1083,4 +1092,77 @@ public InputStream asInput() {
             return new ByteArrayInputStream(this.buf, 0, this.count);
         }
     }
+
+    /**
+     * Wrap the exception so the caller's signature shows up in the stack trace, taking care to copy the original type and message
+     * where possible so async and sync code don't have to check different exceptions.
+     */
+    private static Exception extractAndWrapCause(Exception exception) {
+        if (exception instanceof InterruptedException) {
+            throw new RuntimeException("thread waiting for the response was interrupted", exception);
+        }
+        if (exception instanceof ExecutionException) {
+            ExecutionException executionException = (ExecutionException) exception;
+            Throwable t = executionException.getCause() == null ? executionException : executionException.getCause();
+            if (t instanceof Error) {
+                throw (Error) t;
+            }
+            exception = (Exception) t;
+        }
+        if (exception instanceof ConnectTimeoutException) {
+            ConnectTimeoutException e = new ConnectTimeoutException(exception.getMessage());
+            e.initCause(exception);
+            return e;
+        }
+        if (exception instanceof SocketTimeoutException) {
+            SocketTimeoutException e = new SocketTimeoutException(exception.getMessage());
+            e.initCause(exception);
+            return e;
+        }
+        if (exception instanceof ConnectionClosedException) {
+            ConnectionClosedException e = new ConnectionClosedException(exception.getMessage());
+            e.initCause(exception);
+            return e;
+        }
+        if (exception instanceof SSLHandshakeException) {
+            SSLHandshakeException e = new SSLHandshakeException(
+                exception.getMessage() + "\nSee https://opensearch.org/docs/latest/clients/java/ for troubleshooting."
+            );
+            e.initCause(exception);
+            return e;
+        }
+        if (exception instanceof ConnectException) {
+            ConnectException e = new ConnectException(exception.getMessage());
+            e.initCause(exception);
+            return e;
+        }
+        if (exception instanceof ResponseException) {
+            try {
+                ResponseException e = new ResponseException(((ResponseException) exception).getResponse());
+                e.initCause(exception);
+                return e;
+            } catch (final IOException ex) {
+                // We are not able to reconstruct the response, throw IOException instead
+                return new IOException(exception.getMessage(), exception);
+            }
+        }
+        if (exception instanceof IOException) {
+            return new IOException(exception.getMessage(), exception);
+        }
+        if (exception instanceof OpenSearchException) {
+            final OpenSearchException e = new OpenSearchException(((OpenSearchException) exception).response());
+            e.initCause(exception);
+            return e;
+        }
+        if (exception instanceof OpenSearchClientException) {
+            final OpenSearchClientException e = new OpenSearchClientException(((OpenSearchClientException) exception).response());
+            e.initCause(exception);
+            return e;
+        }
+        if (exception instanceof RuntimeException) {
+            return new RuntimeException(exception.getMessage(), exception);
+        }
+        return new RuntimeException("error while performing request", exception);
+    }
+
 }

java-client/src/test/java/org/opensearch/client/opensearch/integTest/aws/AwsSdk2AsyncStacktraceIT.java

@@ -0,0 +1,32 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.client.opensearch.integTest.aws;
+
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+
+import java.util.List;
+import org.apache.logging.log4j.core.util.Throwables;
+import org.junit.Test;
+import org.opensearch.client.opensearch.OpenSearchClient;
+
+// It would be nice to extend AbstractAsyncStracktraceIT.
+public class AwsSdk2AsyncStacktraceIT extends AwsSdk2TransportTestCase {
+    @Test
+    public void testFailureFromClientPreservesStacktraceOfCaller() throws Exception {
+        final OpenSearchClient client = getClient(false, null, null);
+        Exception thrown = assertThrows(Exception.class, () -> client.indices().get(g -> g.index("nonexisting-index")));
+
+        List<String> stacktraceElements = Throwables.toStringList(thrown);
+        boolean someElementContainsCallerMethodName = stacktraceElements.stream()
+            .anyMatch(it -> it.contains("testFailureFromClientPreservesStacktraceOfCaller"));
+
+        assertTrue(someElementContainsCallerMethodName);
+    }
+}

java-client/src/test/java11/org/opensearch/client/opensearch/integTest/AbstractAsyncStracktraceIT.java

@@ -0,0 +1,25 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.client.opensearch.integTest;
+
+import org.apache.logging.log4j.core.util.Throwables;
+import org.junit.Test;
+
+public abstract class AbstractAsyncStracktraceIT extends OpenSearchJavaClientTestCase {
+    @Test
+    public void testFailureFromClientPreservesStacktraceOfCaller() throws Exception {
+        var thrown = assertThrows(Exception.class, () -> javaClient().indices().get(g -> g.index("nonexisting-index")));
+
+        var stacktraceElements = Throwables.toStringList(thrown);
+        var someElementContainsCallerMethodName = stacktraceElements.stream()
+            .anyMatch(it -> it.contains("testFailureFromClientPreservesStacktraceOfCaller"));
+
+        assertTrue(someElementContainsCallerMethodName);
+    }
+}

java-client/src/test/java11/org/opensearch/client/opensearch/integTest/httpclient5/AsyncStacktraceIT.java

@@ -0,0 +1,13 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.client.opensearch.integTest.httpclient5;
+
+import org.opensearch.client.opensearch.integTest.AbstractAsyncStracktraceIT;
+
+public class AsyncStacktraceIT extends AbstractAsyncStracktraceIT implements HttpClient5TransportSupport {}