Fan-Out With a Seatbelt: Six Ways flatMapMerge Bites Android Apps in Production

flatMapMerge looks like a one-line upgrade from "sequential" to "fast". It is also one of the few Flow operators where a harmless-looking change quietly alters ordering, error semantics, resource usage and lifecycle behaviour all at once.

This post walks through what the operator guarantees, when you should not reach for it at all, and six failure modes I would check for in any code review.

🧠 TL;DR (Too Long; Didn’t Read)

The official docs describe flatMapMerge as map(transform).flattenMerge(concurrency), and they explicitly discourage it in ordinary application flows. A suspending map is usually enough.

• transform is invoked sequentially. Only the returned inner flows are collected concurrently.
• Output order is the order in which items are emitted, not the order of the source.
• The operator is still marked @ExperimentalCoroutinesApi, so you need an opt-in and should expect the API surface to move.

1. What the operator actually does

@ExperimentalCoroutinesApi
fun <T, R> Flow<T>.flatMapMerge(
  concurrency: Int = DEFAULT_CONCURRENCY,
  transform: suspend (value: T) -> Flow<R>
): Flow<R>

Three details in the contract matter more than the signature suggests:

• transform runs one value at a time. If building the inner flow is itself slow (say it first reads a token from disk), that cost is serialised and it can hold up pulling new values from upstream.
• concurrency caps in-flight inner flows, not coroutines in general and not network calls. Once the cap is reached, further inner flows wait.
• Operator fusion. If you apply flowOn, buffer or produceIn after flatMapMerge, they are fused with its merging logic, so one properly configured channel is used for the merge. That sounds like a nicety, but it changes how you should reason about buffering (see failure mode 5).

2. Decide whether you need it at all

The reflex “I have a suspend call per item, so I need flatMapMerge" is usually wrong. If each input produces exactly one output, plain map with a suspend lambda is simpler, preserves order, and is far easier to debug:

// Boring, ordered, and good enough for most screens.
fun Flow<ArticleId>.toSummaries(repo: ArticleRepo): Flow<ArticleSummary> =
  map { id -> repo.loadSummary(id) }

flatMapMerge earns its keep only when both are true:

• each input expands into a stream of values (progress events, paged results, a socket), and
• you genuinely want several of those streams alive at once.

3. A realistic scenario: background photo backup

Our running example is a photo-backup feature. A queue of local media IDs needs uploading, each upload reports progress, and we want three uploads in parallel so we do not saturate a mobile connection.

sealed interface UploadEvent {
  val id: MediaId
  data class Progress(override val id: MediaId, val percent: Int) : UploadEvent
  data class Done(override val id: MediaId, val remoteUrl: String) : UploadEvent
  data class Failed(override val id: MediaId, val cause: Throwable) : UploadEvent
}

interface Uploader {
  /** Cold flow: starts the upload on collection, completes after Done. */
  fun upload(id: MediaId): Flow<UploadEvent>
}

private const val MAX_PARALLEL_UPLOADS = 3

@OptIn(ExperimentalCoroutinesApi::class)
fun BackupQueue.uploads(uploader: Uploader): Flow<UploadEvent> =
  pendingIds() // Flow<MediaId>
    .flatMapMerge(MAX_PARALLEL_UPLOADS) { id ->
      uploader.upload(id)
    }

This compiles, demos well on Wi-Fi, and has at least five latent problems. Let’s go through them.

4. Failure mode 1: you silently lose source order

Inner flows finish when they finish. A tiny thumbnail overtakes a 40 MB video, so the downstream sees completions in finish order. If a UI list or a protocol depends on source order, this is a correctness bug that only shows up under uneven latency, which means it passes every test on your fast office network.

Fixes, in order of preference

Carry an index and reassemble at the edge. Good when the consumer is a list that can be sorted:

data class Indexed<T>(val index: Int, val value: T)

@OptIn(ExperimentalCoroutinesApi::class)
fun <T, R> Flow<T>.mapParallelIndexed(
  concurrency: Int,
  block: suspend (T) -> R
): Flow<Indexed<R>> =
  withIndex()
    .flatMapMerge(concurrency) { (i, v) ->
      flow { emit(Indexed(i, block(v))) }
    }

For a finite collection where you need a final ordered List, skip Flow entirely. A semaphore-bounded async keeps order for free:

import kotlinx.coroutines.async
import kotlinx.coroutines.awaitAll
import kotlinx.coroutines.coroutineScope
import kotlinx.coroutines.sync.Semaphore
import kotlinx.coroutines.sync.withPermit

suspend fun <T, R> Iterable<T>.parallelMap(
  limit: Int,
  block: suspend (T) -> R
): List<R> = coroutineScope {
  val gate = Semaphore(limit)
  map { item ->
    async { gate.withPermit { block(item) } }
  }.awaitAll()
}

5. Failure mode 2: the default concurrency you never chose

Leave concurrency out and you inherit DEFAULT_CONCURRENCY. At the time of writing that is 16 and can be changed through a JVM system property; check the constant's documentation for the version you ship, because the point is that you did not choose it.

One practical note: the constant is a separate public API from the operator and, as far as I can tell, carries a preview opt-in (@FlowPreview) rather than @ExperimentalCoroutinesApi. Don't reference it from your own code to size anything; define your own constant instead, as shown below.

Sixteen parallel inner flows can mean:

• sixteen bitmaps being decoded at once on a low-memory device,
• sixteen requests competing for an HTTP client that only runs a handful per host (the rest simply queue inside the client, so your “parallelism” is an illusion),
• sixteen cursors against a local database connection pool.

Pick the number from the scarcest resource, name it, and keep it next to the code that owns that resource:

// Derived from the HTTP client's per-host limit, not from optimism.
private val UPLOAD_PARALLELISM =
  httpClient.dispatcher.maxRequestsPerHost.coerceAtMost(3)

6. Failure mode 3: slot starvation from flows that never finish

This is the one that burns teams who combine flatMapMerge with Room, DataStore or any other observable source. An observing flow never completes, so it never releases its slot.

// Looks fine. Is not.
userIds.flatMapMerge(concurrency = 3) { id ->
  dao.observeProfile(id)
}

With six IDs, the first three observeProfile flows occupy every slot forever. IDs 4 to 6 are never collected, nothing throws, nothing logs, and QA reports "some avatars never load, but only for users with many contacts". Raising concurrency to a bigger number only moves the cliff.

The structural fix is to stop modelling "one observer per ID":

@Query("SELECT * FROM profile WHERE id IN (:ids)")
fun observeProfiles(ids: List<Long>): Flow<List<Profile>>

// One observer for the whole set. React to id changes with flatMapLatest.
@OptIn(ExperimentalCoroutinesApi::class)
fun visibleProfiles(ids: Flow<List<Long>>): Flow<List<Profile>> =
  ids.flatMapLatest { dao.observeProfiles(it) }

Rule of thumb: flatMapMerge is for inner flows that complete. If the inner flow is a subscription, you want combine, a set-based query, or flatMapLatest.

7. Failure mode 4: one inner failure cancels everyone

An exception in any inner flow propagates to the merged flow, which cancels the other in-flight inner flows. One corrupt file out of fifty and your whole backup run stops.

Model failure as data and contain it inside transform:

@OptIn(ExperimentalCoroutinesApi::class)
fun BackupQueue.uploads(uploader: Uploader): Flow<UploadEvent> =
  pendingIds()
    .flatMapMerge(MAX_PARALLEL_UPLOADS) { id ->
      uploader.upload(id)
        .catch { cause -> emit(UploadEvent.Failed(id, cause)) }
    }

Two cautions:

• Prefer the catch operator over a hand-written try/catch around emit. If you do write a try/catch in a flow { } builder, never swallow CancellationException; rethrow it, or cancellation stops working.
• catch only sees upstream exceptions. A throw from the collector's own lambda is not intercepted, which is correct, but be aware of it when debugging.

8. Failure mode 5: misreading fusion and buffering

Because buffer and flowOn placed after flatMapMerge are fused with it, the buffer you write is the one the merge itself writes into. Practically:

uploads
  .flatMapMerge(3) { uploader.upload(it) }
  .buffer(capacity = 256) // configures the merge channel, not a separate stage
  .collect { render(it) }

A slow collector does not immediately slow the inner flows. They keep producing until that channel fills, which can mean hundreds of progress events queued in memory. For progress-style data, conflate() semantics or a bounded BufferOverflow.DROP_OLDEST buffer is usually what you actually want.

flowOn(Dispatchers.IO) placed after the operator is fused with the merge, so it changes the context of everything upstream of it: the source flow, the transform calls, and the collection of every inner flow. A flowOn placed before the operator only affects the source flow; transform and the inner flows then run in whatever context the merge itself runs in (normally the collector's). Mixing the two up leads to "why is transform running on Main?" mysteries, so put the flowOn after the operator when you want the whole fan-out off the main thread, and add an extra flowOn earlier only if the source needs a different context.

State your intent explicitly, and then test it (section 10).

9. Failure mode 6: lifecycle and restart semantics

Flows are cold. Every new collection restarts the whole pipeline, including uploads that were half done. Collect from a lifecycle-aware scope, and decide deliberately what a restart means:

class BackupViewModel(queue: BackupQueue, uploader: Uploader) : ViewModel() {
  val state: StateFlow<BackupUiState> =
    queue.uploads(uploader)
      .scan(BackupUiState()) { acc, event -> acc.reduce(event) }
      .stateIn(
        scope = viewModelScope,
        started = SharingStarted.WhileSubscribed(stopTimeoutMillis = 5_000),
        initialValue = BackupUiState()
      )
}

// In the UI layer
viewLifecycleOwner.lifecycleScope.launch {
  viewLifecycleOwner.repeatOnLifecycle(Lifecycle.State.STARTED) {
    viewModel.state.collect { render(it) }
  }
}

Two real-world traps:

• Rotation is fine, backgrounding is not. WhileSubscribed(5_000) bridges configuration changes, but once the app has been in the background longer than the timeout, upstream is cancelled. Uploads that need to survive that belong in WorkManager (or a foreground service), with the Flow used only to observe their progress.
• Restart must be idempotent. If cancellation can interrupt an upload mid-flight, the next collection will begin it again. Use stable upload keys or server-side dedup so a retry does not create a duplicate.

10. Testing the contract, not the happy path

The most valuable test is the one that proves your concurrency cap actually holds. Virtual time makes it fast and deterministic:

@OptIn(ExperimentalCoroutinesApi::class)
@Test
fun `never exceeds the configured concurrency`() = runTest {
  val inFlight = AtomicInteger(0)
  val peak = AtomicInteger(0)
  val results = (1..20).asFlow()
    .flatMapMerge(concurrency = 4) { id ->
      flow {
        val now = inFlight.incrementAndGet()
        peak.accumulateAndGet(now) { a, b -> maxOf(a, b) }
        try {
          delay(100) // virtual time inside runTest
          emit(id)
        } finally {
          inFlight.decrementAndGet()
        }
      }
    }
    .toList()
  assertEquals(4, peak.get())
  assertEquals((1..20).toSet(), results.toSet()) // compare as sets: order is not guaranteed
}

The counters are atomic on purpose. Under runTest's default scheduler everything runs on one thread, so plain vars would happen to work, but the test would turn flaky the moment someone adds a real dispatcher inside transform.

Add the companion tests that map to the failure modes above: one inner flow throws and the others still complete (isolation), a never-ending inner flow does not starve the rest (use set-based observation), and a deliberately uneven delay produces out-of-order output (so nobody “fixes” your reindexing later).

11. Review checklist

• Does each input really expand to a stream? If it is one value, use map.
• Is concurrency an explicit, named constant tied to a real resource limit?
• Does every inner flow complete? If any is a subscription, redesign.
• Is order irrelevant, or is there an index and a reassembly step?
• Is failure converted to data inside transform?
• Is the buffer / flowOn placement intentional, given that both fuse with the merge?
• What happens when collection restarts, and is that safe to repeat?
• Is there a test that proves the concurrency cap?

🔚 Closing thought

flatMapMerge is not dangerous because it is complicated; it is dangerous because its defaults are plausible. The unbounded-feeling number, the lost order, the missing error boundary: each one is invisible until production traffic exposes it. Treat it as a power tool: use it deliberately, bound it explicitly, and prefer the plain map whenever the problem allows.

Reference: Official API documentation for flatMapMerge in kotlinx.coroutines. Behaviour described for the default concurrency value and for catch /cancellation semantics reflects my understanding of the library at the time of writing; confirm against the version you ship.

📘 Master Your Next Technical Interview

Since Java is the foundation of Android development, mastering DSA is essential. I highly recommend “Mastering Data Structures & Algorithms in Java”. It’s a focused roadmap covering 100+ coding challenges to help you ace your technical rounds.

• E-book: Available on Google Play
• Kindle Edition: Available on Amazon
• Topmate: Available on Topmate

Also available in Paperback & Hardcover.

Originally published on M