Avoid synchronous checked exceptions in 'suspend fun's

Because of a race condition caused by thread preemption, this can sometimes happen. Explicitly yield to ensure the exception is delivered asynchronously to the continuation.

(cherry picked from commit 60c92e3)

retrofit/src/main/java/retrofit2/HttpServiceMethod.java

@@ -186,9 +186,22 @@ static final class SuspendForBody<ResponseT> extends HttpServiceMethod<ResponseT
 
       //noinspection unchecked Checked by reflection inside RequestFactory.
       Continuation<ResponseT> continuation = (Continuation<ResponseT>) args[args.length - 1];
-      return isNullable
-          ? KotlinExtensions.awaitNullable(call, continuation)
-          : KotlinExtensions.await(call, continuation);
+
+      // Calls to OkHttp Call.enqueue() like those inside await and awaitNullable can sometimes
+      // invoke the supplied callback with an exception before the invoking stack frame can return.
+      // Coroutines will intercept the subsequent invocation of the Continuation and throw the
+      // exception synchronously. A Java Proxy cannot throw checked exceptions without them being
+      // declared on the interface method. To avoid the synchronous checked exception being wrapped
+      // in an UndeclaredThrowableException, it is intercepted and supplied to a helper which will
+      // force suspension to occur so that it can be instead delivered to the continuation to
+      // bypass this restriction.
+      try {
+        return isNullable
+            ? KotlinExtensions.awaitNullable(call, continuation)
+            : KotlinExtensions.await(call, continuation);
+      } catch (Exception e) {
+        return KotlinExtensions.yieldAndThrow(e, continuation);
+      }
     }
   }
 }

retrofit/src/main/java/retrofit2/KotlinExtensions.kt

@@ -19,6 +19,7 @@
 package retrofit2
 
 import kotlinx.coroutines.suspendCancellableCoroutine
+import kotlinx.coroutines.yield
 import kotlin.coroutines.resume
 import kotlin.coroutines.resumeWithException
 
@@ -95,3 +96,8 @@ suspend fun <T : Any> Call<T>.awaitResponse(): Response<T> {
     })
   }
 }
+
+internal suspend fun Exception.yieldAndThrow(): Nothing {
+  yield()
+  throw this
+}

retrofit/src/test/java/retrofit2/KotlinSuspendTest.kt

@@ -273,4 +273,26 @@ class KotlinSuspendTest {
     val body = runBlocking { example.body() }
     assertThat(body).isEqualTo("HiHiHiHiHi")
   }
+
+  @Test fun checkedExceptionsAreNotSynchronouslyThrown() = runBlocking {
+    val retrofit = Retrofit.Builder()
+        .baseUrl("https://unresolved-host.com/")
+        .addConverterFactory(ToStringConverterFactory())
+        .build()
+    val example = retrofit.create(Service::class.java)
+
+    server.shutdown()
+
+    // The problematic behavior of the UnknownHostException being synchronously thrown is
+    // probabilistic based on thread preemption. Running a thousand times will almost always
+    // trigger it, so we run an order of magnitude more to be safe.
+    repeat(10000) {
+      try {
+        example.body()
+        fail()
+      } catch (_: IOException) {
+        // We expect IOException, the bad behavior will wrap this in UndeclaredThrowableException.
+      }
+    }
+  }
 }