Canceling flatMapMerge not canceling the source?

[akarnokd]

In the following code, the source Flow is not aborted as I'd expect:

@Test
fun flatMapCancel() = runBlocking {
    flow {
        try {
            println("-> 1")
            emit(1)
            println("-> 2")
            emit(2)
            println("-> 3")
            emit(3)
            println("-> 4")
            emit(4)
            println("-> 5")
        } catch (ex: Throwable) {
            ex.printStackTrace()
        }
    }
    .flatMapMerge(2) { v -> flow { emit(v) } }
    .take(2)
    .collect { println (it) }
}

prints:

-> 1
-> 2
-> 3
1
2
-> 4
-> 5

To check, setting concurrency to 1, thus using flatMapConcat underneath does abort.

Is this an expected property of flatMapMerge?

[qwwdfsad]

Is this an expected property of flatMapMerge?

Yes and no. In coroutines, cancellation is asynchronous (not in your case, but in general) and cooperative: either it is checked during suspension by our primitives or manually with isActive or ensureActive primitives. But probably we can do better and check cancellation on some code paths.

[qwwdfsad]

Returned to this issue again, turns out some implementation detail is leaking into upstream as well.
Now the next emit after successful collection will throw CE.
Will be fixed in 1.3.0.