Kotlin Flow: Bugs That Look Correct at First

Kotlin Flow is a powerful API for handling asynchronous streams of data, especially in Android development. Its clean APIs make it easy to get started, but things can get surprising as we dig deeper into its operators and execution model.

Flow bugs are surprisingly easy to introduce. The code compiles, the operators look right, and everything works during development. Then an extra collector appears, a request gets cancelled, or a test never receives an update.

In this article, we’ll explore four common 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 whose upstream never starts collecting.

For each bug, we’ll look at what happens, why it happens, and how to fix it.

Let’s start!

1. Why Is My Upstream Operation Executing Multiple Times?

What

Consider the following code:

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

Now imagine that two different consumers call this function:

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

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

This looks perfectly reasonable. However, observeUser() returns a cold Flow, meaning each collection independently executes the upstream code.

Creating a Flow is not the same as executing it. Collecting a cold Flow more than once can repeat its work.

Why

A cold Flow doesn’t start producing values merely because it was created. Its upstream work starts when a collector collects it, and every new collection starts that work again.

This behavior is often desirable. For example, two independent searches might need two independent network requests.

However, when multiple consumers should share the same upstream operation, separate collections can waste network requests or duplicate database subscriptions.

How

The idea is to let consumers share the same upstream collection. We can do that with shareIn():

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

launch {
    sharedUser.collect(::renderHeader)
}

launch {
    sharedUser.collect(::renderProfile)
}

Now both consumers can share the upstream collection while the sharing strategy controls when it runs.

With SharingStarted.WhileSubscribed(), collection starts when the first subscriber arrives and stops when the last subscriber leaves, subject to any configured stop timeout.

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

Choose the scope and sharing strategy according to the desired lifecycle. Sharing isn’t automatically better for every cold Flow.

TL;DR: Multiple collectors of a cold Flow can repeat upstream work. Use shareIn() when multiple consumers should share one upstream collection.

2. Why Doesn’t catch Catch This Exception?

What

Consider this simple example:

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

You might expect the catch operator to emit -1 when the exception occurs. Instead, the exception propagates to the collector.

Why

Flow operators are arranged sequentially. The catch operator handles exceptions thrown by upstream operators, but it doesn’t catch exceptions thrown by downstream operators or the terminal collector.

In this example, onEach appears downstream of catch. Its exception is therefore outside the catch operator’s reach.

How

Move the potentially failing operation before catch:

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

Now the exception thrown by onEach occurs 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 won’t handle failures introduced by operators that appear later.

Also remember that coroutine cancellation is part of structured concurrency and should generally propagate rather than be swallowed as an ordinary error.

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

3. Why Did Flow Skip a Value?

What

Imagine we have 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 may observe:

1
3

The exact result depends on scheduling and 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 every value.

While the collector is busy, the pending value can be replaced by a newer 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 rapidly changing locations or UI state.

How

If every value must be processed, remove conflation:

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

Without conflation, this sequential Flow processes every emitted value.

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

Remember that a finite buffer doesn’t guarantee unlimited retention. If the producer outruns the consumer and the buffer fills, the producer may suspend or values may be discarded, depending on the chosen overflow strategy.

A real-world example: Imagine displaying a live stock price in a trading app. If the price changes rapidly, displaying the latest available price may be more useful than rendering every intermediate update. Conflation can be appropriate for that display—but not for processing every trade or recording every transaction.

TL;DR: conflate() prioritizes freshness over processing every value. Use it only when skipping intermediate values is acceptable.

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 its upstream Flow starts collecting.

With SharingStarted.WhileSubscribed(), the upstream doesn’t start until there is a subscriber.

Calling runCurrent() executes coroutines that are scheduled to run on the test scheduler. It doesn’t create a subscriber.

Therefore, the upstream Flow hasn’t emitted 42, and the state retains its initial value of 0.

How

There are two ways to address this, depending on the behavior you want to test.

Option 1: Start upstream collection immediately

If the upstream should start as soon as the sharing coroutine runs, use SharingStarted.Eagerly:

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

runCurrent()

assertEquals(42, state.value)

Option 2: Keep WhileSubscribed() and start a collector

If subscription-driven behavior is intentional, start a collector in the test:

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

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

runCurrent()

assertEquals(42, state.value)

The second approach tests the actual subscription-driven behavior. The test uses backgroundScope, which is available in the appropriate coroutine test scope, and requires the relevant coroutine-test imports.

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

TL;DR: stateIn() stores state, but it doesn’t necessarily start collecting upstream immediately. Understand the SharingStarted strategy before asserting the state.

Conclusion

These examples demonstrate that Flow operators aren’t just syntax that reduces the amount of code we write. Every operator has specific behavior involving execution, error handling, cancellation, buffering, and lifecycle.

Whenever you choose a Flow operator, ask yourself five questions:

1. Collection: If I add another collector, will it retrigger upstream execution, or will the work be shared?
2. Errors: Which parts of the pipeline can this operator handle, and where will exceptions propagate?
3. Buffering: Is it acceptable to skip intermediate values, or must every value be processed?
4. Lifecycle: How should the Flow share work among collectors? When should collection start and stop?
5. Testing: Does my test actually trigger the collection behavior required for the assertion to pass?

Understanding these details makes Flow code more predictable and helps prevent bugs that are difficult to diagnose from the code alone.

The next time a Flow behaves unexpectedly, look beyond the syntax. Ask yourself what is being collected, when it starts, how values travel through the pipeline, and where errors can escape.