Kotlin Flow timeout operator and Java ExecutorService integration: does timeout cancel upstream collector or only drop emissions?

Short answer

Yes — timeout() cancels the upstream collector; it does not merely drop emissions. But it does not shut down the underlying ExecutorService. Those are two separate lifecycles, and conflating them is what causes lingering Java threads.

What timeout actually does

The timeout() operator is implemented on top of withTimeoutOrNull: it runs the upstream collection inside a coroutine with a deadline. When the deadline expires, that collecting coroutine is cancelled via normal coroutine cancellation — a TimeoutCancellationException (a subtype of CancellationException) propagates upstream through the flow chain. Consequently:

• onCompletion runs with the cancellation cause, and finally blocks in upstream operators execute.
• Resources owned by the flow itself (channels in callbackFlow, handles closed in finally) are released.
• For a StateFlow, the cancelled collector simply stops receiving updates; the StateFlow and its other collectors are unaffected.

So the "only drops emissions" model is wrong — that describes filtering/backpressure operators like sample or conflate, not timeout. Timeout terminates collection.

What timeout does not do

Cancellation stops the coroutine, not the thread pool. A dispatcher created via Dispatchers.from(executor) or asCoroutineDispatcher() borrows the executor; the pool's threads stay alive until you shut the pool down. Additionally, cancellation is cooperative: it is delivered at suspension points. If upstream code runs a blocking Java call that ignores interruption, the coroutine is logically cancelled but the executor thread stays occupied until that call returns.

Guaranteeing executor cleanup

Own the executor lifecycle explicitly around the timed collection:

val executor = Executors.newFixedThreadPool(4)
try {
    executor.asCoroutineDispatcher().use { dispatcher ->
        withContext(dispatcher) {
            flow.timeout(5.seconds).collect { emit -> handle(emit) }
        }
    }
} finally {
    executor.shutdown()          // or shutdownNow() to interrupt running tasks
    executor.awaitTermination(2, TimeUnit.SECONDS)
}

Notes:

• shutdown() lets in-flight tasks finish; shutdownNow() interrupts them. Choose based on whether your blocking work responds to interruption.
• Never shut down a shared executor that other work depends on — Dispatchers.from grants no ownership.
• Make upstream work cooperative: use suspending calls or check ensureActive()/isActive in loops so cancellation is prompt.

Verify it yourself

1. Collect a test flow with timeout on the executor dispatcher and assert onCompletion fires with a TimeoutCancellationException cause.
2. After the timeout, check ThreadPoolExecutor.getActiveCount() or a thread dump to confirm no task lingers.
3. Log Thread.currentThread().isInterrupted inside blocking upstream work to see whether interruption reaches the thread.

These semantics are stable across kotlinx.coroutines versions, though the internal implementation of timeout() varies — check the KDoc for your specific version if you depend on internals.