Throwing exception in channel sharing context with the consumer will also trigger uncaught exceptioon handler

[matejdro]

suspend fun produceData(): ReceiveChannel<String> {
    // I cannot use coroutineScope builder, since it makes this method wait for ever for channel to complete.
    // Use CorotuineScope(coroutineContext) instead

    return CoroutineScope(coroutineContext).produce(capacity = 0) {
        send("Something")

        throw IllegalStateException("Data production failed")
    }
}

fun main() = runBlocking {
    Thread.setDefaultUncaughtExceptionHandler { thread, throwable -> println("Got uncaught exception $throwable") }

    try {
        produceData().consumeEach {
            println("Received $it")
        }
    } catch (e: Exception) {
       println("Caught exception $e!")
    }
}

Running above code will both catch exception in the catch block of the main method AND throw it into uncaught exception handler.

I found two workarounds for this so far:

1. Remove CoroutineScope(coroutineContext) and just use GlobalScope instead, but that could potentially leave channel dangling if parent is cancelled before it starts consuming the channel.
2. Use cancel(Exception()) instead of throw Exception(), but that still causes problem if unexpected exception is thrown in the producer.

Is this a bug or is there a fundamental flaw somewhere in my code?

[elizarov]

This is the same problem as #763. All scoped builders cancel the parent scope on failure. This highlights the general complexity of exception handling in concurrent code. The solution is to use coroutineScope { ... } to delimit the blocks of concurrently executed code and put exception handling around it:

    try {
        coroutineScope { // delimit the scope of concurrency
            produceData().consumeEach {
                println("Received $it")
            }
        }
    } catch (e: Exception) {
        println("Caught exception $e!")
    }

Also, the better way to define produceData is by making it an extension on CoroutineScope, because it makes your intent to start a new coroutine from produceData function explicit:

suspend fun CoroutineScope.produceData(): ReceiveChannel<String> =
    produce(capacity = 0) {
        send("Something")
        throw IllegalStateException("Data production failed")
    }

You can read more about it in this blog post: https://medium.com/@elizarov/explicit-concurrency-67a8e8fd9b25

Does it help?

[matejdro]

Thanks for the info. CoroutineScope extension convenction actually sounds like a pretty good idea. However, one problem I see is how to do this when method is part of a class? For example:

class DataProducer {
    fun CoroutineScope.produceData(): ReceiveChannel<String> {
        return produce(capacity = 0) {
            send("Something")

            throw IllegalStateException("Data production failed")
        }
    }
}

Kotlin does not let me call DataProducer().produceData() even if I'm inside coroutine scope.

[elizarov]

@matejdro You'll have to use like this: with(DataProducer()) { produceData() }

[matejdro]

Thanks for the info. I think I can call this case closed.