Do not throw from JobListenableFuture.isCancelled

This properly handles ExecutionException that can be thrown from getUninterruptibly.

Fixed Kotlin#2421.

Author: vadimsemenov

integration/kotlinx-coroutines-guava/src/ListenableFuture.kt

@@ -385,9 +385,24 @@ private class JobListenableFuture<T>(private val jobToCancel: Job): ListenableFu
         // this Future hasn't itself been successfully cancelled, the Future will return
         // isCancelled() == false. This is the only discovered way to reconcile the two different
         // cancellation contracts.
-        return auxFuture.isCancelled || (isDone && Uninterruptibles.getUninterruptibly(auxFuture) is Cancelled)
+        return auxFuture.isCancelled || auxFuture.completedWithCancellation
     }
 
+    /**
+     * Helper for [isCancelled] that takes into account that
+     * our auxiliary future can complete with [Cancelled] instance.
+     */
+    private val SettableFuture<*>.completedWithCancellation: Boolean
+        get() = isDone && try {
+            Uninterruptibles.getUninterruptibly(this) is Cancelled
+        } catch (e: CancellationException) {
+            true
+        } catch (t: Throwable) {
+            // In theory appart from CancellationException, getUninterruptibly can only
+            // throw ExecutionException, but to be safe we catch Throwable here.
+            false
+        }
+
     /**
      * Waits for [auxFuture] to complete by blocking, then uses its `result`
      * to get the `T` value `this` [ListenableFuture] is pointing to or throw a [CancellationException].

integration/kotlinx-coroutines-guava/test/ListenableFutureTest.kt

@@ -680,6 +680,33 @@ class ListenableFutureTest : TestBase() {
         finish(5)
     }
 
+    @Test
+    fun testFutureCompletedExceptionally() = runTest {
+        val testException = TestException()
+        // NonCancellable to not propagate error to this scope.
+        val future = future(context = NonCancellable) {
+            throw testException
+        }
+        yield()
+        assertTrue(future.isDone)
+        assertFalse(future.isCancelled)
+        val thrown = assertFailsWith<ExecutionException> { future.get() }
+        assertEquals(testException, thrown.cause)
+    }
+
+    @Test
+    fun testAsListenableFutureCompletedExceptionally() = runTest {
+        val testException = TestException()
+        val deferred = CompletableDeferred<String>().apply {
+            completeExceptionally(testException)
+        }
+        val asListenableFuture = deferred.asListenableFuture()
+        assertTrue(asListenableFuture.isDone)
+        assertFalse(asListenableFuture.isCancelled)
+        val thrown = assertFailsWith<ExecutionException> { asListenableFuture.get() }
+        assertEquals(testException, thrown.cause)
+    }
+
     private inline fun <reified T: Throwable> ListenableFuture<*>.checkFutureException() {
         val e = assertFailsWith<ExecutionException> { get() }
         val cause = e.cause!!