Kotlin Coroutineのerror handlingを理解する

Kotlin Fest 2019 での Kotlinコルーチンを理解しよう 2019 が素晴らしかったので、忘れないうちに復習することにします。
サンプルプログラムは前回記事にしたクリーンアーキテクチャのコードを利用しています。

ソフトウェアバージョン

software	version
OS	MacOS Mojave
Spring Boot	2.2.0.M5
Java	Corretto-11.0.4.11.1
Kotlin	1.3.50
Kotlinx.coroutine	1.3.0

Coroutine Scopeのerror handling

以下のケースでgetSampleDataPort.getSampleData2() で例外が発生するとアプリケーションが異常終了します。

class SampleDataService(
    private val getSampleDataPort: GetSampleDataPort
) : GetSampleDataUseCase {
    override suspend fun invoke() = coroutineScope {
        // try-catchしているので一見制御しているように見えるが...
        try {
            val sampleData1Deferred = async { getSampleDataPort.getSampleData() }
            val sampleData2Deferred = async { getSampleDataPort.getSampleData2() } // throw exception!

            val sampleData1 = sampleData1Deferred.await()
            val sampleData2 = sampleData2Deferred.await()

            GetSampleDataUseCase.Output(sampleData1)
        } catch (th: Throwable) {
            th.printStackTrace()
            GetSampleDataUseCase.Output(SampleData("error"))
        }
    }
}

kotlinx.coroutines.JobCancellationException: Parent job is Cancelling; job=ScopeCoroutine{Cancelling}@3508ef35
Caused by: com.example.demo.port.output.sample.GetSampleDataPort$SampleDataNotFoundException

親のcoroutine Scope内で例外が発生すると親スコープが終了してしまいます。
正しく例外を補足するためにはCoroutineの並行性を構造化します。

override suspend fun invoke() = coroutineScope {
    // 子スコープを用意してtryで囲う
    try {
        coroutineScope {
            val sampleData1Deferred = async { getSampleDataPort.getSampleData() }
            val sampleData2Deferred = async { getSampleDataPort.getSampleData2() } // throw exception!
    
            val sampleData1 = sampleData1Deferred.await()
            val sampleData2 = sampleData2Deferred.await()
            GetSampleDataUseCase.Output(sampleData1)
        }
    } catch (th: Throwable) {
        println("error occurred!")
        GetSampleDataUseCase.Output(SampleData("error"))
    }
}

error occurred!

Coroutine Scope内のasyncのerror handling

async処理それぞれに対して例外を補足したい場合、以下のように書くと正しく例外を補足できず、アプリケーションは異常終了します。

override suspend fun invoke() = coroutineScope {
    val sampleData1Deferred = async { getSampleDataPort.getSampleData() }
    val sampleData2Deferred = async { getSampleDataPort.getSampleData2() } // throw exception!

    val sampleData1 = try {
        sampleData1Deferred.await()
    } catch (th: Throwable) {
        SampleData("error")
    }

    val sampleData2 = try {
        sampleData2Deferred.await()
    } catch (th: Throwable) {
        SampleData("error:2")
    }
    
    GetSampleDataUseCase.Output(sampleData1)
}

com.example.demo.port.output.sample.GetSampleDataPort$SampleDataNotFoundException: null
    at com.example.demo.adapter.api.adapter.SampleDataApiAdapter.getSampleData2$suspendImpl(SampleDataApiAdapter.kt:13)
    Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException: 
Error has been observed at the following site(s):
    |_ checkpoint ⇢ Handler com.example.demo.adapter.web.controller.SampleApiController#getSample() [DispatcherHandler]
    |_ checkpoint ⇢ HTTP GET "/sample" [ExceptionHandlingWebHandler]
Stack trace:
        at com.example.demo.adapter.api.adapter.SampleDataApiAdapter.getSampleData2$suspendImpl(SampleDataApiAdapter.kt:13)
        at com.example.demo.adapter.api.adapter.SampleDataApiAdapter.getSampleData2(SampleDataApiAdapter.kt)
        at com.example.demo.service.sample.SampleDataService$invoke$2$sampleData2Deferred$1.invokeSuspend(SampleDataService.kt:16)
        at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:33)
        at kotlinx.coroutines.DispatchedTask.run(Dispatched.kt:241)
        at kotlinx.coroutines.scheduling.CoroutineScheduler.runSafely(CoroutineScheduler.kt:594)
        at kotlinx.coroutines.scheduling.CoroutineScheduler.access$runSafely(CoroutineScheduler.kt:60)
        at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.run(CoroutineScheduler.kt:740)

正しく処理するには以下のようにasync内でtryを記述します。

override suspend fun invoke() = coroutineScope {
    val sampleData1Deferred = async {
        try {
            getSampleDataPort.getSampleData()
        } catch (th: Throwable) {
            SampleData("error")
        }
    }

    val sampleData2Deferred = async {
        try {
            getSampleDataPort.getSampleData2() // throw exception!
        } catch (th: Throwable) {
            SampleData("error:2")
        }
    }

    val sampleData1 = sampleData1Deferred.await()
    val sampleData2 = sampleData2Deferred.await()
    println(sampleData1)  // -> SampleData(value=sample)
    println(sampleData2)  // -> SampleData(value=error:2)

    GetSampleDataUseCase.Output(sampleData1)
}

最後に

2018年のKotlin Festのスライドも最高でした。Kotlin Fest++