Invalid flow variant error when using flatMapMerge

[ansman]

Might be related to #1421

Running this code will throw an exception:

runBlocking {
    makeParentRequest().collect {
        println("Got stuff")
    }
}

fun makeStuffRequests() =
    flow { repeat(100) { emit(it)}}
        .flatMapMerge {
            flow {
                delay(10)
                emit(Unit)
            }
        }

fun makeParentRequest() = flow {
    // Crash when emitting here due to concurrent emits
    makeStuffRequests().collect { emit(it) }
}

The exception in question:

Exception in thread "main" java.lang.IllegalStateException: Flow invariant is violated: emission from another coroutine is detected (child of StandaloneCoroutine{Active}@606d8acf, expected child of BlockingCoroutine{Active}@782830e). FlowCollector is not thread-safe and concurrent emissions are prohibited. To mitigate this restriction please use 'channelFlow' builder instead of 'flow'
	at kotlinx.coroutines.flow.internal.SafeCollector$checkContext$result$1.invoke(SafeCollector.kt:79)
	at kotlinx.coroutines.flow.internal.SafeCollector$checkContext$result$1.invoke(SafeCollector.kt:12)
	at kotlin.coroutines.CoroutineContext$Element$DefaultImpls.fold(CoroutineContext.kt:70)
	at kotlinx.coroutines.Job$DefaultImpls.fold(Job.kt)
	at kotlinx.coroutines.JobSupport.fold(JobSupport.kt:27)
	at kotlin.coroutines.CombinedContext.fold(CoroutineContextImpl.kt:57)
	at kotlinx.coroutines.flow.internal.SafeCollector.checkContext(SafeCollector.kt:37)
	at kotlinx.coroutines.flow.internal.SafeCollector.emit(SafeCollector.kt:30)
	at org.jetbrains.kotlin.idea.scratch.generated.ScratchFileRunnerGenerated$ScratchFileRunnerGenerated$makeParentRequest$1$invokeSuspend$$inlined$collect$1.emit(Collect.kt:136)
	at kotlinx.coroutines.flow.internal.SerializingCollector.emit(Concurrent.kt:57)
	at kotlinx.coroutines.flow.internal.SafeCollector.emit(SafeCollector.kt:33)
	at org.jetbrains.kotlin.idea.scratch.generated.ScratchFileRunnerGenerated$ScratchFileRunnerGenerated$makeStuffRequests$3$1.invokeSuspend(tmp.kt:26)
	at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:33)
	at kotlinx.coroutines.ResumeModeKt.resumeMode(ResumeMode.kt:67)
	at kotlinx.coroutines.DispatchedKt.resume(Dispatched.kt:319)
	at kotlinx.coroutines.DispatchedKt.dispatch(Dispatched.kt:301)
	at kotlinx.coroutines.CancellableContinuationImpl.dispatchResume(CancellableContinuationImpl.kt:250)
	at kotlinx.coroutines.CancellableContinuationImpl.resumeImpl(CancellableContinuationImpl.kt:260)
	at kotlinx.coroutines.CancellableContinuationImpl.resumeUndispatched(CancellableContinuationImpl.kt:332)
	at kotlinx.coroutines.EventLoopImplBase$DelayedResumeTask.run(EventLoop.common.kt:483)
	at kotlinx.coroutines.EventLoopImplBase.processNextEvent(EventLoop.common.kt:270)
	at kotlinx.coroutines.BlockingCoroutine.joinBlocking(Builders.kt:79)
	at kotlinx.coroutines.BuildersKt__BuildersKt.runBlocking(Builders.kt:54)
	at kotlinx.coroutines.BuildersKt.runBlocking(Unknown Source)
	at kotlinx.coroutines.BuildersKt__BuildersKt.runBlocking$default(Builders.kt:36)
	at kotlinx.coroutines.BuildersKt.runBlocking$default(Unknown Source)
	at org.jetbrains.kotlin.idea.scratch.generated.ScratchFileRunnerGenerated$ScratchFileRunnerGenerated.generated_get_instance_res0(tmp.kt:15)
	at org.jetbrains.kotlin.idea.scratch.generated.ScratchFileRunnerGenerated.main(tmp.kt:38)

Switching to use flatMapConcat makes it work again.

[elizarov]

Thanks for the report. This is indeed a problem within flatMapMerge implementation and its interaction with context preservation.