Preserve the caller's stacktrace for blocking API (#2420)

Motivation:

Users of blocking API do not expect `ExecutionException`, like users
of `Future.get()`. However, `ExecutionException` plays an important role
to capture the caller's stacktrace, which helps to understand which path
made a blocking call. Currently, we unwrap `ExecutionException` and lose
this information.

Modifications:

- Instead of dropping the `ExecutionException` wrapper, move it as a
suppressed exception for the original one;
- Test new behavior;

Result:

Users can understand what path requested blocking invocation when an
async operation fails with an exception.

servicetalk-concurrent-api-internal/src/main/java/io/servicetalk/concurrent/api/internal/BlockingUtils.java

@@ -21,7 +21,7 @@
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
 
-import static io.servicetalk.utils.internal.PlatformDependent.throwException;
+import static io.servicetalk.utils.internal.ThrowableUtils.throwException;
 
 /**
  * Common utility functions to unwrap {@link ExecutionException} from async operations.
@@ -50,7 +50,7 @@ public static <T> T futureGetCancelOnInterrupt(Future<T> future) throws Exceptio
             future.cancel(false);
             throw e;
         } catch (ExecutionException e) {
-            return throwException(executionExceptionCause(e));
+            return throwException(unwrapExecutionException(e));
         }
     }
 
@@ -69,7 +69,7 @@ public static <T> T blockingInvocation(Single<T> source) throws Exception {
         try {
             return source.toFuture().get();
         } catch (final ExecutionException e) {
-            return throwException(executionExceptionCause(e));
+            return throwException(unwrapExecutionException(e));
         }
     }
 
@@ -86,11 +86,31 @@ public static void blockingInvocation(Completable source) throws Exception {
         try {
             source.toFuture().get();
         } catch (final ExecutionException e) {
-            throwException(executionExceptionCause(e));
+            throwException(unwrapExecutionException(e));
         }
     }
 
-    private static Throwable executionExceptionCause(ExecutionException original) {
-        return (original.getCause() != null) ? original.getCause() : original;
+    private static Throwable unwrapExecutionException(ExecutionException ee) {
+        final Throwable cause = ee.getCause();
+        if (cause == null) {
+            return ee;
+        }
+        cause.addSuppressed(new SuppressedExecutionException(ee));
+        return cause;
+    }
+
+    private static final class SuppressedExecutionException extends ExecutionException {
+        private static final long serialVersionUID = 2001711259056665126L;
+
+        SuppressedExecutionException(final ExecutionException original) {
+            super("Blocking operation terminated with an error");
+            setStackTrace(original.getStackTrace());
+        }
+
+        @Override
+        public synchronized Throwable fillInStackTrace() {
+            // We inherit the stack-trace from the original ExecutionException
+            return this;
+        }
     }
 }

servicetalk-concurrent-api-internal/src/test/java/io/servicetalk/concurrent/api/internal/BlockingUtilsTest.java

@@ -0,0 +1,61 @@
+/*
+ * Copyright © 2022 Apple Inc. and the ServiceTalk project authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.servicetalk.concurrent.api.internal;
+
+import io.servicetalk.concurrent.api.Completable;
+import io.servicetalk.concurrent.api.Single;
+import io.servicetalk.concurrent.internal.DeliberateException;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.concurrent.ExecutionException;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.arrayWithSize;
+import static org.hamcrest.Matchers.emptyArray;
+import static org.hamcrest.Matchers.instanceOf;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.not;
+import static org.hamcrest.Matchers.sameInstance;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+class BlockingUtilsTest {
+
+    @Test
+    void testSingle() {
+        test(t -> BlockingUtils.blockingInvocation(Single.failed(t)));
+    }
+
+    @Test
+    void testCompletable() {
+        test(t -> BlockingUtils.blockingInvocation(Completable.failed(t)));
+    }
+
+    private static void test(BlockingTask blockingTask) {
+        DeliberateException expected = new DeliberateException();
+        DeliberateException e = assertThrows(DeliberateException.class, () -> blockingTask.run(expected));
+        assertThat(e, is(sameInstance(expected)));
+        assertThat(e.getSuppressed(), is(arrayWithSize(1)));
+        Throwable ee = e.getSuppressed()[0];
+        assertThat(ee, is(instanceOf(ExecutionException.class)));
+        assertThat(ee.getStackTrace(), is(not(emptyArray())));
+    }
+
+    @FunctionalInterface
+    private interface BlockingTask {
+        void run(Throwable t) throws Exception;
+    }
+}