Unexpected crash with AbortFlowException

[sellmair]

After upgrading from version 1.2.2 to 1.3.0-RC2 we encountered this kind of crash at runtime:

kotlinx.coroutines.flow.internal.AbortFlowException: Flow was aborted, no more elements needed
	(Coroutine boundary)
	at kotlinx.coroutines.channels.AbstractSendChannel.offer(AbstractChannel.kt:165)
	at kotlinx.coroutines.channels.RendezvousChannelStressTest$testOfferByThreadStressTest$1$1.invoke(RendezvousChannelStressTest.kt:21)
	at kotlinx.coroutines.channels.RendezvousChannelStressTest$testOfferByThreadStressTest$1$1.invoke(RendezvousChannelStressTest.kt:10)
	at kotlin.concurrent.ThreadsKt$thread$thread$1.run(Thread.kt:30)
Caused by: kotlinx.coroutines.flow.internal.AbortFlowException: Flow was aborted, no more elements needed
	at kotlinx.coroutines.flow.FlowKt__ReduceKt$first$$inlined$collect$2.emit(Collect.kt:138)
	at kotlinx.coroutines.flow.FlowKt__ChannelsKt.emitAll(Channels.kt:56)
	at kotlinx.coroutines.flow.FlowKt.emitAll(Unknown Source)
	at kotlinx.coroutines.flow.ConsumeAsFlow.collect(Channels.kt:124)

Sample code to reproduce: (see #1455 )

        val channel = Channel<Long>(Channel.RENDEZVOUS)
        val valueReceived = AtomicBoolean(false)
        try {
            thread {
                try {
                    var i = 0L
                    while (!valueReceived.get()) {
                        i++
                        channel.offer(i)
                    }
                } catch (e: ClosedSendChannelException){

                }
            }

            @Suppress("DEPRECATION")
            channel.consumeAsFlow().first { true }
        } finally {
            valueReceived.set(true)
        }

Potentially related: #1433

[sellmair]

After looking further into this issue, I recognized that the exception is actually thrown by the channel.offer call. Our actual stack trace inside the project just showed the call to channel.first (probably, because of the stack trace recovery not being active).
The only difference now is, that version 1.2.2 threw ClosedSendChannelException whereas version 1.3.0-RC2 just "re-throws" what the underlying exception was (e.g. "AbortFlowException").

I also noticed that the documentation does not specify which exception is thrown if the channel "isClosedForSend". Therefore, I think this issue can be closed because it is working as expected.

Maybe some kind of boundary for the exception thrown here would make sense here (like the old ClosedSendChannelException)?

Side Note:
I tried

        System.setProperty(DEBUG_PROPERTY_NAME, DEBUG_PROPERTY_VALUE_ON)
        System.setProperty("kotlinx.coroutines.stacktrace.recovery", "true")

To enable stack trace recovery on android, without any success. Am I missing something?

[qwwdfsad]

Thanks for the report!

It indeed works as expected: first operator cancels the channel and then racy offer re-throws this exception. This is a pain point when integrating with regular threads that will hopefully be fixed in #974

To enable stack trace recovery on android, without any success. Am I missing something?

It should be enough, but only if you set these properties before touching any coroutines-related code. Another approach is to tune R8 rules:

-assumenosideeffects class kotlinx.coroutines.DebugKt {
    boolean getDEBUG() return true;
    boolean getRECOVER_STACK_TRACES() return true;
}