fix(rt): handle exceptions in background okhttp response body coroutine (#734)

Wrap okhttp response body handling to avoid fatal crash on Android due to unhandled exception

Co-authored-by: Ian Smith Botsford <[email protected]>

Author: aajtodd

.changes/156447f6-0f80-4fde-a38d-6b92260432b8.json

@@ -0,0 +1,8 @@
+{
+    "id": "156447f6-0f80-4fde-a38d-6b92260432b8",
+    "type": "bugfix",
+    "description": "Fix Android crash when OkHttp response body coroutine throws an exception",
+    "issues": [
+        "awslabs/aws-sdk-kotlin#753"
+    ]
+}

runtime/protocol/http-client-engines/http-client-engine-okhttp/jvm/src/aws/smithy/kotlin/runtime/http/engine/okhttp/OkHttpUtils.kt

@@ -80,17 +80,18 @@ internal fun OkHttpResponse.toSdkResponse(callContext: CoroutineContext): HttpRe
         val ch = SdkByteChannel(true)
         val writerContext = callContext + Dispatchers.IO + callContext.derivedName("response-body-writer")
         val job = GlobalScope.launch(writerContext) {
-            val buffer = ByteArray(DEFAULT_BUFFER_SIZE)
-            val source = body.source()
-
-            while (!source.exhausted()) {
-                val rc = source.read(buffer)
-                if (rc == -1) break
-                ch.writeFully(buffer, 0, rc)
+            val result = runCatching {
+                val buffer = ByteArray(DEFAULT_BUFFER_SIZE)
+                val source = body.source()
+                while (!source.exhausted()) {
+                    val rc = source.read(buffer)
+                    if (rc == -1) break
+                    ch.writeFully(buffer, 0, rc)
+                }
             }
 
             // immediately close when done to signal end of body stream
-            ch.close()
+            ch.close(result.exceptionOrNull())
         }
 
         job.invokeOnCompletion { cause ->

runtime/protocol/http-client-engines/http-client-engine-okhttp/jvm/test/aws/smithy/kotlin/runtime/http/engine/okhttp/OkHttpResponseTest.kt

@@ -111,13 +111,14 @@ class OkHttpResponseTest {
 
     @Test
     fun testSourceReadError(): Unit = runBlocking {
+        var exceptionBubbledUp: Throwable? = null
         val request = HttpRequest(HttpMethod.GET, Url.parse("https://aws.amazon.com"), Headers.Empty, HttpBody.Empty)
 
         val execContext = ExecutionContext()
 
         // replace default exception handler which will print out the stack trace by default.
         // We are expecting an exception so this message is misleading/confusing
-        val exHandler = CoroutineExceptionHandler { _, _ -> }
+        val exHandler = CoroutineExceptionHandler { _, t -> exceptionBubbledUp = t }
 
         // don't tie this to the current job or else it will tear down the test as well
         val callJob = Job()
@@ -173,5 +174,10 @@ class OkHttpResponseTest {
         callJob.complete()
         callJob.join()
         assertEquals(0, callContext.job.children.toList().size)
+
+        assertNull(
+            exceptionBubbledUp,
+            "Unexpected exception bubbled up. It should've be pushed to channel's close() method instead.",
+        )
     }
 }