# Why are downstream exceptions propagated upwards in Flow?

**URL:** <https://discuss.kotlinlang.org/t/why-are-downstream-exceptions-propagated-upwards-in-flow/19823>\
**Category:** Libraries\
**Created:** [November 6, 2020, 12:52pm UTC](https://discuss.kotlinlang.org/t/why-are-downstream-exceptions-propagated-upwards-in-flow/19823 "2020-11-06T12:52:59Z")\
**Posts on this page:** 5\
**Page:** 1

<div class="post-metadata">

**Author:** ![oakkitten](https://avatars.discourse-cdn.com/v4/letter/o/7cd45c/32.png) [@oakkitten](https://discuss.kotlinlang.org/u/oakkitten)\
**Post date:** [November 6, 2020, 12:52pm UTC](https://discuss.kotlinlang.org/t/why-are-downstream-exceptions-propagated-upwards-in-flow/19823/1 "2020-11-06T12:52:59Z")

</div>

I’m reading the [guide on Flow](https://kotlinlang.org/docs/reference/coroutines/flow.html#exception-transparency) and it says,

> Flows must be transparent to exceptions and it is a violation of the exception transparency to emit values in `the flow { ... }` builder from inside of a `try/catch` block. This guarantees that a collector throwing an exception can always catch it using `try/catch` as in the previous example.

This is a bit hard to understand, but if you actually do try use `try/catch` in emitter you can see the issue [(play.kotlinlang)](https://pl.kotl.in/6MZZ7s_FP):

```kotlin
import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

class FunnyException : Exception()

fun main() = runBlocking<Unit> {
    try {
        flow { 
            emit(0)
            try {
                emit(1)
            } catch (e: FunnyException) {
                emit(2)
            }
        }.onEach {
            if (it == 1) throw FunnyException()
        }.collect {
            println(it)
        }
    } catch (e: Throwable) {
        println("Caught $e")
    } 
}

```

This produces:

> ```plaintext
> 0
> Caught java.lang.IllegalStateException: Flow exception transparency is violated:
> Previous 'emit' call has thrown exception FunnyException, but then emission attempt of value '2' has been detected.
> Emissions from 'catch' blocks are prohibited in order to avoid unspecified behaviour, 'Flow.catch' operator can be used instead.
> For a more detailed explanation, please refer to Flow documentation.
> 
> ```

I guess in this particular piece of code it makes sense that the emitter shouldn’t try to emit after it receives an exception. What I don’t see is, why emitter receives `FunnyException` in the first place? What useful things can it do with this exception?

I expected that the exception thrown in `onEach` is propagated downwards instead, and `CancellationException` is thrown inside the `flow {}` block. Instead, that block can either produce the above error or it can completely suppress the exception (try commenting out `emit(2)`). I this case the program only prints “0”. This is not disallowed even though it seems Kotlin could detect this; what would be the purpose of such code? Doesn’t this also break the “exception transparency”?

P.S. Why is `CancellationException` an `Exception`? Wouldn’t it make sense to extend directly from `Throwable` so that it is not inadvertedly caught by `catch (e: Exception)`?

---

<div class="post-metadata">

**Author:** ![nickallendev](https://avatars.discourse-cdn.com/v4/letter/n/cdc98d/32.png) [@nickallendev](https://discuss.kotlinlang.org/u/nickallendev)\
**Post date:** [November 7, 2020, 11:00am UTC](https://discuss.kotlinlang.org/t/why-are-downstream-exceptions-propagated-upwards-in-flow/19823/2 "2020-11-07T11:00:02Z")

</div>

> [@oakkitten](#):
>
> This is a bit hard to understand

Basically this just means that if `emit` fails with an exception, then the whole collection should fail with that same exception.

> [@oakkitten](#):
>
> What I don’t see is, why emitter receives `FunnyException` in the first place?

Exceptions **always** propagate up to the caller. It’s the signal that the requested code failed. That exception is indicating that `emit` failed. This happens because `emit` eventually calls the `onEach` lambda that throws the exception.

> [@oakkitten](#):
>
> What useful things can it do with this exception?

The guide is saying that you aren’t allowed to do anything with the exception.

> [@oakkitten](#):
>
> I expected that the exception thrown in `onEach` is propagated downwards instead, and `CancellationException` is thrown

You can only throw an exception up to the caller. There’s not really another direction to throw it. If `emit` fails, the goal is to pass that exception (not some different exception) up to the code calling `collect`.

> [@oakkitten](#):
>
> This is not disallowed even though it seems Kotlin could detect this

`Flow` is just a regular Kotlin class defined in a Kotlin library. It can’t do anything special. The IDE could potentially detect this case and give a warning. You could file an [Issue](https://youtrack.jetbrains.com/issues).

> [@oakkitten](#):
>
> Doesn’t this also break the “exception transparency”?

Because `Flow` isn’t special and Kotlin has no way to require a lambda or method to be transparent to exceptions, it’s up to you to not break exception transparency.

> [@oakkitten](#):
>
> P.S. Why is `CancellationException` an `Exception` ? Wouldn’t it make sense to extend directly from `Throwable` so that it is not inadvertedly caught by `catch (e: Exception)` ?

`CancellationException` comes straight from Java, it’s not unique to coroutines.

---

<div class="post-metadata">

**Author:** ![oakkitten](https://avatars.discourse-cdn.com/v4/letter/o/7cd45c/32.png) [@oakkitten](https://discuss.kotlinlang.org/u/oakkitten)\
**Post date:** [November 7, 2020, 3:25pm UTC](https://discuss.kotlinlang.org/t/why-are-downstream-exceptions-propagated-upwards-in-flow/19823/3 "2020-11-07T15:25:11Z")

</div>

Aha, thanks, I think I’m getting it. Correct me if I’m wrong here:

So I was thinking about Flow in terms of Python’s asynchronous generators. In Python, you would do something like

```python
>>> async def emitter():
... yield 1
... yield 2
...
>>> async for x in emitter():
... print(x)
1
2

```

Kotlin’s Flow looks similar, but unlike `sequence {}`, _it does not involve iterators_

```kotlin
flow {
    emit(1)
    emit(2)
}.collect {
    println(it)
}

```

The key difference here is that in Python the steering wheel remains in the top level code, while in Kotlin it’s passed to the Flow block. In other words, in Python the part that is in charge is the `async for` loop; it requests stuff from `emitter` and then passes it to the body. In Kotlin, on the other hand, running `collect` will lauch the flow block that calls emit methods which are calling the downstream methods in a more or less direct way.

This allows Python to catch exception in a more direct way, you can simply do like

```python
try:
    await anext(emitter_obj)
except MyException:
    ...

```

But in Kotlin this code would’t be in charge. What you can do instead is wrap the whole `flow {}.collect {}` in a `try/catch`, which is what `.catch {}` is doing. It has to check that the exception doesn’t come from downstream so it’s a bit more complicated than that.

> CancellationException comes straight from Java, it’s not unique to coroutines.

Why not make a new thing?

---

<div class="post-metadata">

**Author:** ![nickallendev](https://avatars.discourse-cdn.com/v4/letter/n/cdc98d/32.png) [@nickallendev](https://discuss.kotlinlang.org/u/nickallendev)\
**Post date:** [November 8, 2020, 5:40am UTC](https://discuss.kotlinlang.org/t/why-are-downstream-exceptions-propagated-upwards-in-flow/19823/4 "2020-11-08T05:40:01Z")

</div>

Haven’t used coroutines in Python but that reminds me of use cases that rely on [Channels](https://kotlinlang.org/docs/reference/coroutines/channels.html).

With a `Channel`, you can receive just the next item at a time. Exceptions work more like you were expecting since the code producing values is just adding to the `Channel` instead of invoking the consumer code directly.

So why would anyone use `Flow` when there’s `Channel`?

`Flow` is more versatile (it waits to run until you collect), more efficient (direct calls are faster than communicating with a `Channel`), and simpler to use (nothing to close unlike a `Channel`).

If needed, any `Flow` can be consumed with a `Channel` by calling [produceIn](https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/produce-in.html). Or viceversa, with `asFlow`.

`Flow` is generally preferred for cases representing, manipulating, and combining streams of data.

Note that even with `Channel`, I think the semantics will be different that you are used to. `Channel` is an async queue, not an async iterator. The producer doesn’t wait for the value to be requested, it starts right away.

I’m not aware of an actual async iterator in Kotlin’s coroutine library though I can’t really think of a use case where it’d be preferable, so I’m not surprised.

> [@oakkitten](#):
>
> Why not make a new thing?

I imagine the Kotlin team saw the benefit of using the existing exception as better than any derived from creating some new second exception for the basically the same exact purpose.

---

<div class="post-metadata">

**Author:** ![oakkitten](https://avatars.discourse-cdn.com/v4/letter/o/7cd45c/32.png) [@oakkitten](https://discuss.kotlinlang.org/u/oakkitten)\
**Post date:** [November 12, 2020, 9:40pm UTC](https://discuss.kotlinlang.org/t/why-are-downstream-exceptions-propagated-upwards-in-flow/19823/5 "2020-11-12T21:40:07Z")

</div>

`Channel`s are still not quite the same, as the producer still has control, doesn’t it? If the consumer of the channel takes 3 elements, the producer will go ahead and make the fourth one. It will stop after, but that’s not quite what we want with lazy iterators…

…So for the sake of learning this whole business I went ahead and tried implementing an asynchronous sequence. My idea was to make it behave exactly like `sequence {}`:

```kotlin
runBlocking {
    val seq = asyncSequence {
        yield(1)
        delay(100)
        yield(2)
    }

    for (i in seq) println(i)
}

```

(This is just a proof of concept made for fun. In all likelihood not thread-safe. This has one cool feature: if the generator is hot, the `finally` block after `yield` is always executed!)

While it works, `asyncSequence` is rather useless. It produces an “iterator” that’s [suitable for the for loop](https://kotlinlang.org/docs/reference/control-flow.html#for-loops), but it doesn’t implement the actual `Iterator` interface because of the suspended functions, so the for loop is about the only place where it can be used. But I’m still proud of this.

the code behind `asyncSequence` (native):

```kotlin
package asynciterator

import kotlinx.coroutines.*
import kotlin.coroutines.intrinsics.createCoroutineUnintercepted
import kotlin.coroutines.*
import kotlin.experimental.ExperimentalTypeInference
import kotlin.system.getTimeMillis

interface AsyncCoroutineScope<T> : CoroutineScope {
    suspend fun yield(value: T)
}

interface AsyncIterator<T> {
    operator fun iterator() = this
    suspend operator fun hasNext(): Boolean
    suspend operator fun next(): T
}

private sealed class Status<T>
private class NotReady<T> : Status<T>()
private class Ready<T>(val value: T) : Status<T>()
private class Done<T> : Status<T>()
private class Failed<T>(val exception: Throwable) : Status<T>()

class AsyncIteratorImpl<T>(
    override val coroutineContext: CoroutineContext
) : AsyncIterator<T>, AsyncCoroutineScope<T>, Continuation<Unit> {
    private lateinit var mainContinuation: Continuation<Status<T>>
    internal lateinit var sequenceContinuation: Continuation<Unit>

    private var status: Status<T> = NotReady()

    override operator fun iterator() = this

    private suspend fun ensureNext() {
        if (status is NotReady) {
            status = suspendCoroutine {
                mainContinuation = it
                sequenceContinuation.resume(Unit)
            }
            status.let { if (it is Failed) throw it.exception }
        }
    }

    private fun throwBadStatus(): Nothing {
        when (val status = status) {
            is Done -> throw NoSuchElementException("Asynchronous iterator is exhausted")
            is Failed -> throw IllegalStateException("Asynchronous sequence block previously threw an exception", status.exception)
            else -> throw IllegalStateException("This ain't possible")
        }
    }

    override suspend operator fun hasNext(): Boolean {
        ensureNext()
        if (status is Done) return false
        if (status is Ready) return true
        throwBadStatus()
    }

    override suspend operator fun next(): T {
        ensureNext()
        status.let { if (it is Ready) return it.value.also { status = NotReady() } }
        throwBadStatus()
    }

    override suspend fun yield(value: T) = suspendCancellableCoroutine<Unit> {
        sequenceContinuation = it
        mainContinuation.resume(Ready(value))
    }

    // the following is completion continuation stuff. resumeWith here will be called when the coroutine
    // either run to the very end or throws an exception; this includes CancellationException
    override val context = coroutineContext

    override fun resumeWith(result: Result<Unit>) {
        val exception = result.exceptionOrNull()
        if (exception is CancellationException) return
        mainContinuation.resume(if (exception != null) Failed(exception) else Done())
    }
}

@OptIn(ExperimentalTypeInference::class)
fun <T> CoroutineScope.asyncSequence(@BuilderInference block: suspend AsyncCoroutineScope<T>.() -> Unit): AsyncIterator<T> {
    return AsyncIteratorImpl<T>(coroutineContext).apply {
        sequenceContinuation = block.createCoroutineUnintercepted(this, this)
    }
}

```

a few tests:

```kotlin
@file:Suppress("ControlFlowWithEmptyBody")

package asynciterator

import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.delay
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import kotlin.math.absoluteValue
import kotlin.system.measureTimeMillis
import kotlin.test.Test
import kotlin.test.assertFailsWith
import kotlin.test.assertTrue

fun CoroutineScope.makeSequence(delayDuration: Long = 0) = asyncSequence {
    for (x in 1..5) {
        delay(delayDuration)
        yield(x)
    }
}

@Suppress("DIVISION_BY_ZERO")
fun CoroutineScope.makeFailingSequence() = asyncSequence {
    for (x in makeSequence()) yield(x)
    1 / 0
}

@Test fun sum() = runBlocking {
    var sum = 0
    for (x in makeSequence()) sum += x
    assertTrue { sum == 15 }
}

@Test fun concurrency() {
    val single = measureTimeMillis {
        runBlocking {
            launch { for (x in makeSequence(100)) {} }
        }
    }

    val double = measureTimeMillis {
        runBlocking {
            launch { for (x in makeSequence(100)) {} }
            launch { for (x in makeSequence(100)) {} }
        }
    }

    assertTrue { (double - single).absoluteValue / double < 0.1 }
}

@Test fun throws() = runBlocking<Unit> {
    assertFailsWith(ArithmeticException::class) {
        for (x in makeFailingSequence()) {}
    }
}

@Test fun `throws if next() is called again`() = runBlocking<Unit> {
    val seq = makeFailingSequence()

    assertFailsWith(ArithmeticException::class) {
        for (x in seq) {}
    }

    val e = assertFailsWith(IllegalStateException::class) {
        for (x in seq) {}
    }

    assertTrue { e.cause is ArithmeticException }
}

@Test fun `doesn't automatically proceed after yield()`() = runBlocking<Unit> {
    for (x in makeFailingSequence()) {
        if (x == 5) break
    }
}

@Test fun `runs finalizer`() {
    var finalizerRun = false

    runBlocking {
        val seq = asyncSequence {
            try {
                for (x in 1..5) yield(x)
            } finally {
                finalizerRun = true
            }
        }

        for (x in seq) {
            if (x == 3) break
        }
    }

    assertTrue { finalizerRun }
}

```