Do not throw from JobListenableFuture.isCancelled

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

@@ -347,6 +347,17 @@ private class JobListenableFuture<T>(private val jobToCancel: Job): ListenableFu
      */
     private val auxFuture = SettableFuture.create<Any>()
 
+    /**
+     * `true` if [auxFuture.get][ListenableFuture.get] throws [ExecutionException].
+     *
+     * Note: this is eventually consistent with the state of [auxFuture].
+     *
+     * Unfortunately, there's no API to figure out if [ListenableFuture] throws [ExecutionException]
+     * apart from calling [ListenableFuture.get] on it. To avoid unnecessary [ExecutionException] allocation
+     * we use this field.
+     */
+    private var auxFutureIsFailed: Boolean = false
+
     /**
      * When the attached coroutine [isCompleted][Job.isCompleted] successfully
      * its outcome should be passed to this method.
@@ -366,7 +377,8 @@ private class JobListenableFuture<T>(private val jobToCancel: Job): ListenableFu
     // CancellationException is wrapped into `Cancelled` to preserve original cause and message.
     // All the other exceptions are delegated to SettableFuture.setException.
     fun completeExceptionallyOrCancel(t: Throwable): Boolean =
-        if (t is CancellationException) auxFuture.set(Cancelled(t)) else auxFuture.setException(t)
+        if (t is CancellationException) auxFuture.set(Cancelled(t))
+        else auxFuture.setException(t).also { if (it) auxFutureIsFailed = true }
 
     /**
      * Returns cancellation _in the sense of [Future]_. This is _not_ equivalent to
@@ -385,7 +397,16 @@ 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 || isDone && !auxFutureIsFailed && try {
+            Uninterruptibles.getUninterruptibly(auxFuture) is Cancelled
+        } catch (e: CancellationException) {
+            // `auxFuture` got cancelled right after `auxFuture.isCancelled` returned false.
+            true
+        } catch (e: ExecutionException) {
+            // `auxFutureIsFailed` hasn't been updated yet.
+            auxFutureIsFailed = true
+            false
+        }
     }
 
     /**

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

@@ -680,6 +680,50 @@ 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)
+    }
+
+    @Test
+    fun stressTestJobListenableFutureIsCancelledDoesNotThrow() = runTest {
+        repeat(1000) {
+            val deferred = CompletableDeferred<String>()
+            val asListenableFuture = deferred.asListenableFuture()
+            // We heed two threads to test a race condition.
+            withContext(Dispatchers.Default) {
+                val cancellationJob = launch {
+                    asListenableFuture.cancel(false)
+                }
+                while (!cancellationJob.isCompleted) {
+                    asListenableFuture.isCancelled // Shouldn't throw.
+                }
+            }
+        }
+    }
+
     private inline fun <reified T: Throwable> ListenableFuture<*>.checkFutureException() {
         val e = assertFailsWith<ExecutionException> { get() }
         val cause = e.cause!!