flatMapLatest + retryWhen lead to emission loss when collected inside TestScope or inside runBlocking scope

[AlexWoodblock]

Kotlin version: 1.6.10
Coroutines version:
Platform: JVM

val list = flow {
  emit(1)
}
.flatMapLatest {
  flow {
    emit("First item for $it")
    throw RuntimeException()
  }
}
.retryWhen { _, _ ->
  emit("Error!")
  true
 }
.take(6)
.toList()

println(list)

Running this code prints [Error!, Error!, Error!, Error!, Error!, Error!] inside runTest or runBlocking, despite having emit("First item for $it") before the exception throwing.

However, collecting the flow in another dispatcher (for example, Dispatchers.Default, or Dispatchers.IO) will make it work as expected, printing [First item for 1, Error!, First item for 1, Error!, First item for 1, Error!].

[dkhalanskyjb]

This code:

runBlocking {
    repeat(10) {
        val list = withContext(Dispatchers.Default) {
            flow {
                emit(1)
            }
                .flatMapLatest {
                    flow {
                        emit("First item for $it")
                        throw RuntimeException()
                    }
                }
                .retryWhen { _, _ ->
                    emit("Error!")
                    true
                }
                .take(6)
                .toList()
        }
        println(list)
    }
}

Produced the following for me:

[First item for 1, Error!, First item for 1, Error!, Error!, Error!]
[Error!, Error!, First item for 1, Error!, Error!, Error!]
[Error!, Error!, Error!, Error!, Error!, Error!]
[Error!, Error!, Error!, Error!, Error!, Error!]
[Error!, Error!, Error!, Error!, Error!, Error!]
Exception in thread "main" java.lang.RuntimeException
	at repro3118.Reproduce3118Kt$main$1$1$list$1$2$1.invokeSuspend(Reproduce3118.kt:16)
	at repro3118.Reproduce3118Kt$main$1$1$list$1$2$1.invoke(Reproduce3118.kt)
	at repro3118.Reproduce3118Kt$main$1$1$list$1$2$1.invoke(Reproduce3118.kt)
	at kotlinx.coroutines.flow.SafeFlow.collectSafely(Builders.kt:61)
	at kotlinx.coroutines.flow.AbstractFlow.collect(Flow.kt:227)
        ...
	Suppressed: kotlinx.coroutines.flow.internal.AbortFlowException: Flow was aborted, no more elements needed

So, I didn't get the result you expected even once. If you get your result consistently, do you maybe mean some other code?

In any case, to answer your questions, Dispatchers.Default and Dispatchers.IO are multithreaded. Collections and emissions happen in different coroutines, which communicate via channels. Because of that, emissions can happen in one thread while collections simultaneously happen in another.

Both runBlocking and runTest, on the other hand, are single-threaded, and only one coroutine may execute at a time. That order is highly dependent on the implementation details. In this case, that order is emit(1), emit("First …"), throw, emit("Error!"), take, true, six times in a row. You can see that take does not get a chance to fire after emit("First …"), it is immediately conflated by emit("Error!").

I'm not sure if this behavior is a bug, of either retryWhen or flatMapLatest. @qwwdfsad, could you comment on this? Also, see the stack trace above.

As a workaround, if you insert a yield between emit("First ...") and throw, then the expected result always happens with runBlocking and runTest, because this would force the emission coroutine to yield so the collection coroutine has a chance to run on the single available thread.

[AlexWoodblock]

@dkhalanskyjb

If you get your result consistently, do you maybe mean some other code?

I consistently get it between different runs of the code.

Running the code I initially attached still produces this:

[Error!, Error!, Error!, Error!, Error!, Error!]

However, when I run the code you provided, I get the following output consistently (on MacOS 12.1 for Macbook Air with M1):

[First item for 1, Error!, Error!, Error!, Error!, Error!]
[Error!, First item for 1, Error!, Error!, First item for 1, Error!]
Exception in thread "main" java.lang.RuntimeException
	at test.MainKt$main$1$1$list$1$2$1.invokeSuspend(Main.kt:20)
	at test.MainKt$main$1$1$list$1$2$1.invoke(Main.kt)
	at test.MainKt$main$1$1$list$1$2$1.invoke(Main.kt)
	at kotlinx.coroutines.flow.SafeFlow.collectSafely(Builders.kt:61)
	at kotlinx.coroutines.flow.AbstractFlow.collect(Flow.kt:227)
	at kotlinx.coroutines.flow.FlowKt__CollectKt.emitAll(Collect.kt:109)
	at kotlinx.coroutines.flow.FlowKt.emitAll(Unknown Source)
	at test.MainKt$main$1$1$list$1$invokeSuspend$$inlined$flatMapLatest$1.invokeSuspend(Merge.kt:216)
	at test.MainKt$main$1$1$list$1$invokeSuspend$$inlined$flatMapLatest$1.invoke(Merge.kt)
	at test.MainKt$main$1$1$list$1$invokeSuspend$$inlined$flatMapLatest$1.invoke(Merge.kt)
	at kotlinx.coroutines.flow.internal.ChannelFlowTransformLatest$flowCollect$3$1$2.invokeSuspend(Merge.kt:34)
	at kotlinx.coroutines.flow.internal.ChannelFlowTransformLatest$flowCollect$3$1$2.invoke(Merge.kt)
	at kotlinx.coroutines.flow.internal.ChannelFlowTransformLatest$flowCollect$3$1$2.invoke(Merge.kt)
	at kotlinx.coroutines.intrinsics.UndispatchedKt.startCoroutineUndispatched(Undispatched.kt:55)
	at kotlinx.coroutines.CoroutineStart.invoke(CoroutineStart.kt:112)
	at kotlinx.coroutines.AbstractCoroutine.start(AbstractCoroutine.kt:126)
	at kotlinx.coroutines.BuildersKt__Builders_commonKt.launch(Builders.common.kt:56)
	at kotlinx.coroutines.BuildersKt.launch(Unknown Source)
	at kotlinx.coroutines.BuildersKt__Builders_commonKt.launch$default(Builders.common.kt:47)
	at kotlinx.coroutines.BuildersKt.launch$default(Unknown Source)
	at kotlinx.coroutines.flow.internal.ChannelFlowTransformLatest$flowCollect$3$1.emit(Merge.kt:33)
	at kotlinx.coroutines.flow.internal.SafeCollectorKt$emitFun$1.invoke(SafeCollector.kt:15)
	at kotlinx.coroutines.flow.internal.SafeCollectorKt$emitFun$1.invoke(SafeCollector.kt:15)
	at kotlinx.coroutines.flow.internal.SafeCollector.emit(SafeCollector.kt:77)
	at kotlinx.coroutines.flow.internal.SafeCollector.emit(SafeCollector.kt:59)
	at test.MainKt$main$1$1$list$1$1.invokeSuspend(Main.kt:15)
	at test.MainKt$main$1$1$list$1$1.invoke(Main.kt)
	at test.MainKt$main$1$1$list$1$1.invoke(Main.kt)
	at kotlinx.coroutines.flow.SafeFlow.collectSafely(Builders.kt:61)
	at kotlinx.coroutines.flow.AbstractFlow.collect(Flow.kt:227)
	at kotlinx.coroutines.flow.internal.ChannelFlowTransformLatest$flowCollect$3.invokeSuspend(Merge.kt:27)
	at kotlinx.coroutines.flow.internal.ChannelFlowTransformLatest$flowCollect$3.invoke(Merge.kt)
	at kotlinx.coroutines.flow.internal.ChannelFlowTransformLatest$flowCollect$3.invoke(Merge.kt)
	at kotlinx.coroutines.intrinsics.UndispatchedKt.startUndispatchedOrReturn(Undispatched.kt:89)
	at kotlinx.coroutines.CoroutineScopeKt.coroutineScope(CoroutineScope.kt:264)
	at kotlinx.coroutines.flow.internal.ChannelFlowTransformLatest.flowCollect(Merge.kt:25)
	at kotlinx.coroutines.flow.internal.ChannelFlowOperator.collectTo$suspendImpl(ChannelFlow.kt:157)
	at kotlinx.coroutines.flow.internal.ChannelFlowOperator.collectTo(ChannelFlow.kt)
	at kotlinx.coroutines.flow.internal.ChannelFlow$collectToFun$1.invokeSuspend(ChannelFlow.kt:60)
	at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:33)
	at kotlinx.coroutines.DispatchedTask.run(DispatchedTask.kt:106)
	at kotlinx.coroutines.scheduling.CoroutineScheduler.runSafely(CoroutineScheduler.kt:571)
	at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.executeTask(CoroutineScheduler.kt:750)
	at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.runWorker(CoroutineScheduler.kt:678)
	at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.run(CoroutineScheduler.kt:665)
	Suppressed: kotlinx.coroutines.flow.internal.AbortFlowException: Flow was aborted, no more elements needed

so between different repeats of the same code block it's not really consistent with your results.

The yield() workaround works, as expected, but it's a little confusing for me - coming from RxJava, call to onNext won't return until the emission is processed, if there are no observeOn operators added. Is the same true for emit - does it suspend until the emission is processed, or it only suspends until emission is scheduled, even for a single thread?

[dkhalanskyjb]

when I run the code you provided

My code is just your code, but run in Dispatchers.Default and collected ten times, right?

The point was exactly that the code is not deterministic. You said initially that, when being run in Dispatchers.Default, the code produces the following:

[First item for 1, Error!, First item for 1, Error!, First item for 1, Error!]

However, this result is not a given and is dependent on a particular thread switching order.

so between different repeats of the same code block it's not really consistent with your results.

Yeah, because each run would give different results, and almost none of them are what you expected.

The code can be rigged to almost always produce the results you want by doing, say, delay(20) between emit("First ...") and throw, but once in a while, another result would happen.

Is the same true for emit

Nope, it only suspends until it's enqueued, the delivery is not guaranteed. Even without errors being thrown, an internal buffer may even overflow, in which case some values will be lost.