Kotlin Flow: Bugs That Look Correct at First

Flow has been a great API across all platforms, especially on Android, for dealing with synchronous updates. It looks very easy to use, thanks to its clean APIs, but it can be surprising along the way. Flow has much more to offer as we dig deeper or when we explore its APIs.

Flow bugs are surprisingly easy to introduce. The code compiles, the operators look good, everything works during dev testing but then an extra collector appears, and/or request gets cancelled, or a test never receives an update.

In this article, we will discuss 4 well-known bugs.

1. Accidentally executing an upstream operation multiple times.
2. Expecting catch to handle a downstream exception.
3. Losing intermediate values with conflate.
4. Waiting forever for a StateFlow that never starts collecting upstream.

For each of the bug, we will go through what, why and how (to fix it)

Let’s start!

1. Multiple (2 or more) collectors

What

Consider the following code:

fun observeUser(): Flow<User> = flow {
    emit(api.fetchUser())
}

Now let’s assume that this is being called by 2 different consumers:

launch {
    repository.observeUser().collect(::renderHeader)
}

launch {
    repository.observeUser().collect(::renderProfile)
}

This looks very reasonable, but the catch here is that observeUser() is cold, meaning each subscription will execute the upstream code independently.

TLDR: Creating a flow is not the same as executing it, and collecting a cold flow more than once can repeat its work.

Why

A cold flow does not start producing values merely because it was created. Its upstream work starts when a collector collects it. Every new collection starts that work again.

This behavior is often desirable. For example, two independent searches might need two independent requests. But when multiple consumers should share one upstream operation, separate collections can waste network requests or duplicate database subscriptions.

How

The idea is that consumers should share the same upstream. And that can be done using our beloved operator shareIn()

val sharedUser = repository.observeUser()
    .shareIn(
        scope = applicationScope,
        started = SharingStarted.WhileSubscribed(),
        replay = 1,
    )

launch {
    sharedUser.collect(::renderHeader)
}

launch {
    sharedUser.collect(::renderProfile)
}

Now the consumers share the upstream collection while the sharing strategy keeps it active. With WhileSubscribed(), collection starts when the first subscriber arrives and stops when the last subscriber leaves, subject to its configured timeout.

The replay = 1 setting also lets a new subscriber receive the latest emitted user.

Choose the scope and sharing strategy according to the desired lifecycle. Sharing is not automatically better for every cold flow.

2. Why doesn’t catch catch this exception?

What

Let’s consider this simple example:

flowOf(1000)
    .catch { emit(-1) }
    .onEach {
        if (it == 1) {
            error("Boom!")
        }
    }
    .collect()

One would expect that the catch operator to emit -1 when the exception occurs. But instead, it propagates to the collector.

Why

Flow operators are arranged sequentially. The catch operator handles exceptions from upstream of its position, but it does not catch exceptions thrown by downstream operators or the terminal collector.

In this example, onEach is downstream of catch. Therefore, its exception is outside the catch operator’s reach.

How

Move the potentially failing operation upstream of catch:

flowOf(1000)
    .onEach {
        if (it == 1) {
            error("Boom!")
        }
    }
    .catch { emit(-1) }
    .collect()

Now the exception thrown by onEach is upstream of catch, so the handler can emit -1. This distinction matters when deciding where to place error handling. A catch near the beginning of a pipeline may not handle failures introduced by transformations that appear later. Coroutine cancellation is part of structured concurrency and should generally propagate.

TLDR: catch handles upstream failures, not every exception that occurs anywhere in the pipeline.

3. Why did Flow skip a value?

What

Let’s assume there is this simple Flow

flow {
    emit(1)
    emit(2)
    emit(3)
}
.conflate()
.collect {
    delay(500)
    println(it)
}

You might expect the output to be:

1
2
3

But the collector can observe:

1
3

The exact scheduling depends on execution timing, but conflation allows intermediate values to be skipped when the collector is slower than the producer.

Why

conflate() allows the producer to continue without waiting for the collector to process each value. While the collector is busy, the pending value is replaced by the latest one. In effect, the collector gets the freshest available value rather than every intermediate value.

This is useful for streams where newer values supersede older ones, such as a rapidly changing location or UI state.

How

If every value must be processed, remove conflation:

flow {
    emit(1)
    emit(2)
    emit(3)
}.collect {
    delay(100)
    println(it)
}

If producer and collector should run asynchronously without intentionally dropping values, consider buffer() with an appropriate capacity and overflow strategy.

Remember that a finite buffer alone does not guarantee unlimited retention. If the producer outruns the consumer and the buffer fills, the producer may suspend or the chosen overflow strategy may discard values.

TLDR: conflate() prioritizes freshness over processing every value. Use it only when skipping intermediate values is acceptable. Good example would be showing updated stock price in stock market/trading app.

4. Why is my StateFlow still at its initial value?

What

Consider this code inside a coroutine test:

val state = flow {
    emit(42)
}.stateIn(
    scope = backgroundScope,
    started = SharingStarted.WhileSubscribed(),
    initialValue = 0,
)

runCurrent()

assertEquals(42, state.value)

The assertion fails: state.value is still 0.

Why

stateIn() creates a StateFlow, but the SharingStarted strategy determines when the upstream flow starts collecting.

With SharingStarted.WhileSubscribed(), the upstream does not start until there is a subscriber. Calling runCurrent() runs scheduled coroutines; it does not create a subscriber.

Therefore, the upstream flow has not emitted 42, and the state retains its initial value.

How

If the upstream should start immediately, use SharingStarted.Eagerly:

val state = flow {
    emit(42)
}.stateIn(
    scope = backgroundScope,
    started = SharingStarted.Eagerly,
    initialValue = 0,
)

runCurrent()

assertEquals(42, state.value)

Alternatively, keep WhileSubscribed() if that is the intended lifecycle, and start a collector in the test:

val collectionJob = backgroundScope.launch {
    state.collect {}
}

runCurrent()

assertEquals(42, state.value)

This second approach tests the actual subscription-driven behavior. It requires the appropriate coroutine imports and a test scope in which backgroundScope is available.

This issue is particularly relevant when testing Android ViewModels. A UI may normally collect a state flow, but a unit test that reads .value without collecting it may never trigger upstream work.

TLDR: stateIn() stores state, but it does not necessarily start collecting upstream immediately. Understand the SharingStarted strategy before asserting the state.

Conclusion

All of the examples in the section demonstrate that the Flow operators aren’t just syntax that allows for less typing. Every single operator is responsible for a specific behavior regarding execution, error handling, cancellation, buffering, and lifecycle.

If you choose to use an operator, you must ask yourself five questions:

1. Collection: If I add another collector, will it retrigger the upstream execution?
2. Errors: How much of the pipeline can this operator handle, and which part(s) of the pipeline are going to get an exception, and what part(s) will just continue on as if nothing happened?
3. Buffering: Is it acceptable to ignore the values in the buffer and just send the latest one?
4. Lifecycle: How should the flow share work among its collectors? When is the best time for a flow to begin and cease the collection of items?