Consistently unwrap exception in slow path of CompletionStage.asDeferred

* A dedicated test is added that checks consistency of unwrapping
  between slow and fast paths of both CompletionStage.await and
  CompletionStage.asDeferred implementations.
* In the fast-path for both of them implementation of
  CompletableFuture.get() does unwrapping of CompletionException, taking
  its cause when it is not null. To mimic this behavior, the slow path
  of both await and asDeferred shall perform similar unwrapping.

Fixes #1479

Author: elizarov

integration/kotlinx-coroutines-jdk8/src/future/Future.kt

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
  */
 
 package kotlinx.coroutines.future
@@ -124,9 +124,12 @@ public fun <T> CompletionStage<T>.asDeferred(): Deferred<T> {
     val result = CompletableDeferred<T>()
     whenComplete { value, exception ->
         if (exception == null) {
+            // the future has completed normally
             result.complete(value)
         } else {
-            result.completeExceptionally(exception)
+            // the future has completed with an exception, unwrap it consistently with fast path
+            // Note: In the fast-path the implementation of CompletableFuture.get() does unwrapping
+            result.completeExceptionally((exception as? CompletionException)?.cause ?: exception)
         }
     }
     if (this is Future<*>) result.cancelFutureOnCompletion(this)
@@ -171,7 +174,7 @@ private class ContinuationConsumer<T>(
     override fun accept(result: T?, exception: Throwable?) {
         val cont = this.cont ?: return // atomically read current value unless null
         if (exception == null) {
-            // the future has been completed normally
+            // the future has completed normally
             cont.resume(result as T)
         } else {
             // the future has completed with an exception, unwrap it to provide consistent view of .await() result and to propagate only original exception

integration/kotlinx-coroutines-jdk8/test/future/FutureTest.kt

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
  */
 
 package kotlinx.coroutines.future
@@ -264,11 +264,9 @@ class FutureTest : TestBase() {
         try {
             deferred.await()
             fail("deferred.await() should throw an exception")
-        } catch (e: CompletionException) {
+        } catch (e: TestException) {
             assertTrue(deferred.isCancelled)
-            val cause = e.cause?.cause!! // Stacktrace augmentation
-            assertTrue(cause is TestException)
-            assertEquals("something went wrong", cause.message)
+            assertEquals("something went wrong", e.message)
         }
     }
 
@@ -437,6 +435,45 @@ class FutureTest : TestBase() {
         }
     }
 
+    /**
+     * Tests that both [CompletionStage.await] and [CompletionStage.asDeferred] consistently unwrap
+     * [CompletionException] both in their slow and fast paths.
+     * See [issue #1479](https://github.com/Kotlin/kotlinx.coroutines/issues/1479).
+     */
+    @Test
+    fun testConsistentExceptionUnwrapping() = runTest {
+        expect(1)
+        // Check the fast path
+        val fFast = CompletableFuture.supplyAsync {
+            expect(2)
+            throw TestException()
+        }
+        fFast.checkFutureException<TestException>() // wait until it completes
+        // Fast path in await and asDeferred.await() shall produce TestException
+        expect(3)
+        val dFast = fFast.asDeferred()
+        assertFailsWith<TestException> { fFast.await() }
+        assertFailsWith<TestException> { dFast.await() }
+        // Same test, but future has not completed yet, check the slow path
+        expect(4)
+        val barrier = CyclicBarrier(2)
+        val fSlow = CompletableFuture.supplyAsync {
+            barrier.await()
+            expect(6)
+            throw TestException()
+        }
+        val dSlow = fSlow.asDeferred()
+        launch(start = CoroutineStart.UNDISPATCHED) {
+            expect(5)
+            // Slow path on await shall produce TestException, too
+            assertFailsWith<TestException> { fSlow.await() } // will suspend here
+            assertFailsWith<TestException> { dSlow.await() }
+            finish(7)
+        }
+        barrier.await()
+        fSlow.checkFutureException<TestException>() // now wait until it completes
+    }
+
     private inline fun <reified T: Throwable> CompletableFuture<*>.checkFutureException(vararg suppressed: KClass<out Throwable>) {
         val e = assertFailsWith<ExecutionException> { get() }
         val cause = e.cause!!