Kotlin Coroutines

What you write:

suspend fun loadUser(id: String): User {
    val profile = fetchProfile(id)    // suspension point 1
    val orders = fetchOrders(id)      // suspension point 2
    return User(profile, orders)
}

What the compiler generates (simplified):

fun loadUser(id: String, cont: Continuation<User>): Any? {
    val sm = cont as? LoadUserStateMachine ?: LoadUserStateMachine(cont)
    
    when (sm.state) {
        0 -> {
            sm.state = 1
            val result = fetchProfile(id, sm)  // passes state machine as callback
            if (result == COROUTINE_SUSPENDED) return COROUTINE_SUSPENDED
            // If fetchProfile completed immediately (cached), fall through
        }
        1 -> {
            sm.profile = sm.result             // restore result from suspension
            sm.state = 2
            val result = fetchOrders(id, sm)
            if (result == COROUTINE_SUSPENDED) return COROUTINE_SUSPENDED
        }
        2 -> {
            sm.orders = sm.result
            return User(sm.profile, sm.orders)  // final result
        }
    }
}

The key insight: suspend doesn't make anything asynchronous by itself. It tells the compiler: "this function CAN pause." The compiler transforms it into a state machine that saves its state at each suspension point, returns COROUTINE_SUSPENDED, and can be resumed later via the Continuation callback.

Continuation is like a callback with state: "here's where I left off, here's my local variables, call me when the result is ready." But unlike callback hell, the code LOOKS sequential thanks to the compiler transformation.

C# comparison: C#'s async/await does exactly the same transformation — the compiler generates a state machine class implementing IAsyncStateMachine. Same concept, different syntax. Kotlin's suspend = C#'s async. Kotlin's direct call = C#'s await.

// SEQUENTIAL — direct calls:
suspend fun loadDashboard(): Dashboard {
    val user = fetchUser()         // waits ~200ms
    val orders = fetchOrders()     // waits ~300ms (starts AFTER user completes)
    return Dashboard(user, orders) // total: ~500ms
}

// CONCURRENT — async:
suspend fun loadDashboard(): Dashboard = coroutineScope {
    val user = async { fetchUser() }       // starts immediately
    val orders = async { fetchOrders() }   // starts immediately, in PARALLEL
    Dashboard(user.await(), orders.await()) // total: ~300ms (max of both)
}

The critical difference explained:

In Kotlin, calling a suspend function directly is like C#'s await — it suspends and waits for the result before continuing. There is NO implicit concurrency.

async { } is like C#'s Task.Run() or storing a Task<T> without immediately awaiting — it starts the work and gives you a handle (Deferred<T>) to get the result later.

// C# equivalent of sequential:
var user = await FetchUser();          // waits
var orders = await FetchOrders();      // waits (sequential)

// C# equivalent of concurrent:
var userTask = FetchUser();            // starts (no await yet)
var ordersTask = FetchOrders();        // starts in parallel
var user = await userTask;             // now wait for both
var orders = await ordersTask;

Common mistake — async without concurrency:

// POINTLESS async — immediately awaiting:
val user = async { fetchUser() }.await()  // same as direct call!
// async starts it, .await() immediately waits. No concurrency gained.
// Just write: val user = fetchUser()

Rule: use direct calls for sequential operations. Use async only when you need two or more things to run in parallel.

// Dispatchers.Default — CPU-intensive work:
withContext(Dispatchers.Default) {
    val hash = computeSHA256(largeData)     // CPU busy 100%
    val parsed = parseComplexJson(input)    // CPU busy 100%
}
// Thread count = CPU cores (e.g., 8 on 8-core machine)
// Adding more threads for CPU work = more context switching, no speedup

// Dispatchers.IO — blocking I/O:
withContext(Dispatchers.IO) {
    val result = db.query("SELECT ...")     // thread WAITS for DB
    val response = httpClient.get(url)      // thread WAITS for network
    val file = File("data.txt").readText()  // thread WAITS for disk
}
// Thread count = max(64, cores)
// Many threads OK because they're mostly WAITING, not computing

// Dispatchers.Main — Android UI thread:
withContext(Dispatchers.Main) {
    textView.text = "Loaded!"  // UI updates must be on Main thread
}
// Single thread. Only for Android/Swing/JavaFX UI frameworks.
// NOT available in backend (Spring Boot) — no Main dispatcher.

Why Dispatchers.IO exists in Kotlin but not in C#:

C#'s async I/O is truly non-blocking at the OS level (IOCP on Windows, epoll on Linux). When you await HttpClient.GetAsync(), no thread is blocked — the OS handles it.

Java/Kotlin: most I/O libraries (JDBC, file I/O, many HTTP clients) are BLOCKING. db.query() blocks the calling thread until the DB responds. Dispatchers.IO provides a large thread pool specifically for these blocking calls — so they don't exhaust the Default pool (which has few threads).

Dispatchers.Unconfined: starts on the caller's thread, resumes on whatever thread the suspension completed on. Useful for testing and very specific scenarios. Don't use in production — unpredictable thread behavior.

Creating custom dispatcher:

// Single-threaded dispatcher — serializes all operations:
val singleThread = newSingleThreadContext("my-thread")
withContext(singleThread) {
    // All code runs on one dedicated thread
    // No concurrent access — no synchronization needed
    // But: poor scaling, can't use more than one core
}
// Remember to close: singleThread.close()

val context = Dispatchers.IO + SupervisorJob() + CoroutineName("worker") + handler

// This is like a Map<Key, Element>:
// context[ContinuationInterceptor] = Dispatchers.IO    (WHERE to run)
// context[Job]                     = SupervisorJob()   (failure policy)
// context[CoroutineName]           = "worker"          (for debugging)
// context[CoroutineExceptionHandler] = handler         (error handling)

// Each element has a unique Key. + merges them:
val ctx1 = Dispatchers.IO + CoroutineName("a")
val ctx2 = ctx1 + Dispatchers.Default
// ctx2[ContinuationInterceptor] = Dispatchers.Default  (last wins!)
// ctx2[CoroutineName] = "a"                            (kept from ctx1)

Why + works: CoroutineContext defines operator fun plus(other: CoroutineContext). Same mechanism as our Money class with operator fun plus. Kotlin operator overloading makes the syntax clean.

Context inheritance: child coroutines inherit parent's context, but can override specific elements:

val scope = CoroutineScope(Dispatchers.IO + SupervisorJob())

scope.launch {
    // Inherits: IO dispatcher + SupervisorJob
    
    launch(Dispatchers.Default) {
        // Overrides: Default dispatcher
        // Inherits: SupervisorJob from parent
    }
    
    launch(CoroutineName("child-2")) {
        // Overrides: adds a name
        // Inherits: IO dispatcher + SupervisorJob
    }
}

Accessing context inside a coroutine:

launch {
    val name = coroutineContext[CoroutineName]?.name
    val job = coroutineContext[Job]
    val dispatcher = coroutineContext[ContinuationInterceptor]
    println("Running in: $name on $dispatcher")
}

// launch — exception goes to parent IMMEDIATELY:
coroutineScope {
    launch {
        throw RuntimeException("boom")
        // Exception propagates to coroutineScope RIGHT NOW
        // coroutineScope cancels other children
        // coroutineScope re-throws to caller
    }
    launch {
        delay(1000)  // CANCELLED by parent because sibling failed
    }
}
// Caller's try/catch catches "boom"

// async — exception stored, thrown at .await():
coroutineScope {
    val result = async {
        throw RuntimeException("boom")
        // Exception stored inside Deferred
    }
    println("still running!")     // THIS EXECUTES — exception not thrown yet
    result.await()                // NOW exception is thrown
}

BUT — important nuance with coroutineScope:

In coroutineScope, even async exceptions eventually propagate to the parent (structured concurrency). The "stored until await" behavior is more visible with supervisorScope:

// supervisorScope — async exception truly deferred:
supervisorScope {
    val result = async {
        throw RuntimeException("boom")
    }
    // Other children NOT cancelled — supervisor doesn't propagate
    try {
        result.await()            // exception thrown HERE
    } catch (e: Exception) {
        println("Caught: ${e.message}")  // handle it
    }
}

Practical implication: use launch for fire-and-forget where failure should cancel the whole operation. Use async when you want to handle each failure independently:

supervisorScope {
    val results = urls.map { url ->
        async {
            try { fetch(url) }
            catch (e: Exception) { null }  // failed URL returns null
        }
    }
    val successful = results.awaitAll().filterNotNull()
    // Some URLs failed — that's OK, we got the rest
}

// main() — must bridge from non-suspend to suspend:
fun main() {                      // non-suspend entry point
    runBlocking {                  // BLOCKS main thread
        val result = fetchUser()   // can call suspend functions here
        println(result)
    }
    // main thread blocked until runBlocking completes
}

// Test — same bridge:
@Test
fun `should transfer money`() = runBlocking {
    val result = service.transfer(from, to, amount)
    result shouldBe TransferResult.Success(...)
}
// JUnit test method is non-suspend. runBlocking bridges.

Why runBlocking inside a coroutine is dangerous:

// BAD — runBlocking inside coroutine:
suspend fun processAll(items: List<Item>) {
    items.forEach { item ->
        runBlocking {              // BLOCKS the thread!
            processItem(item)
        }
    }
}
// Dispatchers.Default has 8 threads.
// 8 coroutines call processAll simultaneously.
// All 8 threads blocked in runBlocking.
// No threads left for processItem to resume on.
// DEADLOCK.

// GOOD — just call directly:
suspend fun processAll(items: List<Item>) {
    items.forEach { item ->
        processItem(item)          // direct call — suspends, doesn't block
    }
}

For Kotest: BehaviorSpec and StringSpec handle suspend automatically — no runBlocking needed inside tests. But for setup code or JUnit integration, runBlocking is acceptable.

Spring Boot: controllers with suspend fun are automatically called within a coroutine context by Spring WebFlux. No runBlocking needed.

// coroutineScope — structured, waits, fail-fast:
suspend fun handleRequest() = coroutineScope {
    val a = async { fetchA() }
    val b = async { fetchB() }
    Result(a.await(), b.await())
}
// handleRequest() doesn't return until BOTH complete.
// If fetchA fails, fetchB is cancelled.
// Exception propagates to caller.
// Use for: request handling, parallel decomposition.

// supervisorScope — structured, waits, independent failures:
suspend fun sendNotifications() = supervisorScope {
    launch { sendEmail() }     // fails
    launch { sendPush() }      // continues!
    launch { sendSMS() }       // continues!
}
// sendNotifications() waits for all three.
// If email fails, push and SMS are NOT cancelled.
// Use for: independent subtasks where partial success is OK.

// Standalone CoroutineScope — NOT structured:
class KafkaConsumer {
    private val scope = CoroutineScope(Dispatchers.IO + SupervisorJob())
    
    fun start() {
        scope.launch { consumeMessages() }  // runs in background
        // start() returns IMMEDIATELY
        // consumeMessages() continues running independently
    }
    
    fun stop() {
        scope.cancel()  // manually cancel when shutting down
    }
}
// Use for: background tasks (Kafka consumer, scheduled jobs)
// whose lifecycle is tied to a component, not a request.

When to use each:

coroutineScope: 90% of cases. Parallel subtasks within a request. Structured, safe, exceptions propagate.

supervisorScope: independent subtasks. Notifications, analytics, non-critical side effects.

Standalone CoroutineScope: background processes, Kafka consumers, scheduled jobs. Needs manual .cancel() on shutdown.

// WRONG — try/catch outside launch:
try {
    scope.launch { throw RuntimeException("boom") }
} catch (e: Exception) {
    // NEVER REACHED! launch returns Job immediately.
    // Exception happens later, inside the coroutine.
}

// Option 1: CoroutineExceptionHandler
val handler = CoroutineExceptionHandler { _, e ->
    logger.error("Background task failed", e)
}
val scope = CoroutineScope(Dispatchers.IO + SupervisorJob() + handler)
scope.launch { throw RuntimeException("boom") }
// handler catches it. Logged. Not lost.

// Option 2: try/catch INSIDE launch
scope.launch {
    try {
        riskyOperation()
    } catch (e: CancellationException) {
        throw e  // ALWAYS re-throw CancellationException!
    } catch (e: Exception) {
        logger.error("Failed", e)
    }
}

// Option 3: launchSafe wrapper (production pattern)
fun CoroutineScope.launchSafe(name: String, block: suspend () -> Unit) =
    launch {
        try { block() }
        catch (e: CancellationException) { throw e }
        catch (e: Exception) { logger.error("[$name] failed", e) }
    }

scope.launchSafe("kafka-consumer") { consumeMessages() }

Why SupervisorJob is critical for standalone scopes:

// With Job() — one child failure kills the scope:
val scope = CoroutineScope(Dispatchers.IO + Job())
scope.launch { task1() }
scope.launch { throw Exception() }  // fails
scope.launch { task3() }            // NEVER STARTS — scope is dead!

// With SupervisorJob() — children are independent:
val scope = CoroutineScope(Dispatchers.IO + SupervisorJob())
scope.launch { task1() }            // runs
scope.launch { throw Exception() }  // fails, others unaffected
scope.launch { task3() }            // runs!

// FLOW — cold, declarative data stream:
fun priceUpdates(): Flow<Price> = flow {
    while (true) {
        emit(fetchLatestPrice())
        delay(1000)
    }
}
// Nothing runs until someone collects:
priceUpdates()
    .filter { it.amount > 100 }
    .map { it.format() }
    .collect { display(it) }      // NOW it runs

// Two collectors = two independent streams
// Each collector gets its own execution

// CHANNEL — hot, imperative communication pipe:
val payments = Channel<Payment>(capacity = 100)

// Producer coroutine:
launch {
    for (p in incomingPayments) {
        payments.send(p)          // suspends if buffer full
    }
}

// Consumer coroutine:
launch {
    for (p in payments) {         // suspends if buffer empty
        process(p)
    }
}
// One payment goes to one consumer only
// If no consumer is listening, send() suspends (backpressure)

Key differences:

Flow is cold: nothing happens until .collect(). Each collector starts a fresh execution. Like a recipe — reading it doesn't cook the food.

Channel is hot: data is sent whether or not someone is receiving. If a producer sends to a channel and no consumer is listening, the data waits in the buffer (or send() suspends). Like a conveyor belt — it moves regardless of who's watching.

Choose Flow for: data transformation pipelines, reactive UI updates, anything that looks like list.filter.map but asynchronous. Choose Channel for: producer-consumer communication between two coroutines, fan-out (one producer, multiple consumers), actor pattern.

SharedFlow / StateFlow — hot Flows:

// StateFlow — always has a current value, like LiveData:
val balance = MutableStateFlow(Money(0, "EUR"))
balance.value = newBalance           // update
balance.collect { updateUI(it) }     // immediately gets current + future

// SharedFlow — hot event stream, multiple collectors share:
val events = MutableSharedFlow<Event>()
events.emit(event)                   // all collectors receive it

// runTest — virtual time, delays execute instantly:
@Test
fun `should retry 3 times`() = runTest {
    val service = RetryService(maxRetries = 3, delayMs = 60_000)
    
    val result = service.fetchWithRetry()  // internally: delay(60_000) x 3
    // In real time: would take 3 minutes
    // In runTest: INSTANT — virtual time skips all delays
    
    result shouldBe expected
}

// advanceTimeBy — simulate passage of time:
@Test
fun `should expire cache after 5 minutes`() = runTest {
    cache.put("key", "value")
    
    advanceTimeBy(4 * 60 * 1000)  // advance 4 minutes
    cache.get("key") shouldBe "value"  // still cached
    
    advanceTimeBy(2 * 60 * 1000)  // advance 2 more minutes (total 6)
    cache.get("key") shouldBe null     // expired!
}

// advanceUntilIdle — run all pending coroutines:
@Test
fun `should process all items`() = runTest {
    val processor = BatchProcessor()
    processor.submitAll(items)     // launches coroutines internally
    
    advanceUntilIdle()             // run all pending coroutines to completion
    
    processor.processedCount shouldBe items.size
}

Testing with MockK for suspend functions:

val repo = mockk<AccountRepository>()
coEvery { repo.findById(any()) } returns account  // coEvery for suspend
coVerify { repo.findById(AccountId("A1")) }        // coVerify for suspend

// Regular every/verify for non-suspend functions
// coEvery/coVerify for suspend functions

Kotest + coroutines: BehaviorSpec and StringSpec bodies are already suspend — no runBlocking or runTest wrapper needed for simple tests. Use runTest explicitly when testing virtual time or TestDispatcher features.

Common testing mistake:

// BAD — runBlocking ignores leaked coroutines:
runBlocking {
    launch { infiniteLoop() }  // leaked! But test passes.
}

// GOOD — runTest catches leaks:
runTest {
    launch { infiniteLoop() }  // TEST FAILS: "Unfinished coroutines"
}

// withTimeout — throws on timeout:
suspend fun fetchWithTimeout(): User {
    return withTimeout(5000) {         // 5 second limit
        httpClient.get("/user/123")    // if this takes > 5s...
    }                                  // TimeoutCancellationException thrown!
}
// Caller must handle the exception:
try {
    val user = fetchWithTimeout()
} catch (e: TimeoutCancellationException) {
    showError("Request timed out")
}

// withTimeoutOrNull — returns null on timeout:
suspend fun fetchOrNull(): User? {
    return withTimeoutOrNull(5000) {   // 5 second limit
        httpClient.get("/user/123")
    }                                  // returns null if timeout, no exception
}
val user = fetchOrNull() ?: defaultUser

How it works under the hood: withTimeout starts a coroutine and schedules a cancellation after the specified delay. If the block completes before the timeout, the scheduled cancellation is cancelled. If not, CancellationException (specifically TimeoutCancellationException) is thrown at the next suspension point.

Important: TimeoutCancellationException extends CancellationException. In coroutineScope, it's treated as cancellation — siblings are cancelled. But in supervisorScope or when caught explicitly, it can be handled as a timeout without cancelling siblings.

Real-world pattern — circuit breaker with timeout:

suspend fun callExternalApi(): Response {
    return withTimeoutOrNull(3000) {
        externalService.call()
    } ?: Response.fallback("Service unavailable")
}

// Problem — tight CPU loop can't be cancelled:
suspend fun processLargeDataset(data: List<Item>) {
    for (item in data) {
        heavyComputation(item)    // no suspension point!
    }
    // If scope.cancel() is called during this loop,
    // the coroutine keeps running until the loop finishes.
    // CancellationException is only thrown at SUSPENSION POINTS.
}

// Fix — check manually with isActive or ensureActive():

// Option 1: isActive (check as boolean):
suspend fun processLargeDataset(data: List<Item>) {
    for (item in data) {
        if (!isActive) break       // exit loop if cancelled
        heavyComputation(item)
    }
}

// Option 2: ensureActive() (throws if cancelled):
suspend fun processLargeDataset(data: List<Item>) {
    for (item in data) {
        ensureActive()             // throws CancellationException if cancelled
        heavyComputation(item)
    }
}

// Option 3: yield() (suspend + check cancellation):
suspend fun processLargeDataset(data: List<Item>) {
    for (item in data) {
        yield()                    // suspend briefly, check cancellation
        heavyComputation(item)     // also gives other coroutines a chance to run
    }
}

When to use each:

ensureActive(): most common. Lightweight check, throws if cancelled. Use in tight loops.

isActive: when you need cleanup before stopping. if (!isActive) { cleanup(); break }

yield(): when you want to be "nice" to other coroutines — gives them CPU time. Also checks cancellation. Slightly more overhead than ensureActive().

C# equivalent: cancellationToken.ThrowIfCancellationRequested() = ensureActive(). cancellationToken.IsCancellationRequested = isActive.

// Thread.sleep — BLOCKS OS thread:
launch(Dispatchers.Default) {    // 8 threads in pool
    Thread.sleep(1000)           // blocks 1 of 8 threads for 1 second
}
// 1 thread wasted. 7 remaining for all other coroutines.
// 8 coroutines all do Thread.sleep → all 8 threads blocked → deadlock!

// delay — SUSPENDS coroutine:
launch(Dispatchers.Default) {
    delay(1000)                  // coroutine suspended, thread FREED
}
// Thread immediately available for other coroutines.
// 10,000 coroutines can all delay(1000) on 8 threads — no problem.

What happens internally:

delay(1000): schedules a timer. Saves coroutine state (Continuation). Frees the thread. After 1000ms, the timer fires. Continuation is resumed — possibly on a different thread from the same dispatcher.

Thread.sleep(1000): OS puts the thread to sleep. Thread exists in memory (1MB stack) but does nothing. After 1000ms, OS wakes the thread. No other coroutine can use this thread during the sleep.

The rules:

In suspend function:   use delay()         (suspends coroutine)
In regular code:       use Thread.sleep()   (blocks thread)
In tests:              use runTest + delay() (virtual time, instant)

class Application {
    private val scope = CoroutineScope(Dispatchers.IO + SupervisorJob())

    fun start() {
        scope.launch { kafkaConsumer.consume() }
        scope.launch { scheduledCleanup() }
        
        Runtime.getRuntime().addShutdownHook(Thread {
            scope.cancel()  // Step 2: cancel all coroutines
            // Each coroutine:
            //   - Gets CancellationException at next suspension point
            //   - Runs finally { } blocks for cleanup
            //   - Completes with "cancelled" status
        })
    }
}

// Inside a coroutine with cleanup:
scope.launch {
    try {
        while (isActive) {
            val msg = kafka.poll()
            processMessage(msg)      // if cancelled during processing...
        }
    } finally {
        // Runs even on cancellation!
        kafka.commitOffsets()         // save progress
        kafka.close()                 // close connection
        logger.info("Consumer stopped gracefully")
    }
}

The timeline:

T=0s    K8s decides to replace pod (rolling update)
T=0s    K8s sends SIGTERM to JVM process
T=0s    JVM runs shutdown hooks
T=0s    Shutdown hook: scope.cancel()
T=0s    All coroutines get CancellationException
T=0.1s  Coroutines run finally blocks (cleanup, commit, close)
T=0.5s  All coroutines completed. JVM exits cleanly.
        
T=30s   (If JVM didn't exit) K8s sends SIGKILL — forced termination
        No cleanup. Data may be lost.

Spring Boot: @PreDestroy is called during shutdown — put scope.cancel() there. Spring handles the shutdown hook internally.

// Mutex — coroutine-friendly:
val mutex = Mutex()

suspend fun debit(amount: Money) = mutex.withLock {
    // Only one coroutine enters at a time
    // Others: SUSPENDED — thread freed for other work
    check(balance >= amount)
    balance -= amount
}
// mutex.withLock auto-releases on exit, even on exception (like try-finally)

// synchronized — thread-blocking:
fun debitBlocking(amount: Money) {
    synchronized(lock) {
        // Only one thread enters at a time
        // Others: BLOCKED — thread does nothing, 1MB stack wasted per thread
        check(balance >= amount)
        balance -= amount
    }
}

Important difference from C#:

C#: lock(obj) — every object can be a lock (built-in monitor).

Kotlin: Mutex() — you create it explicitly as a class field. Objects don't have built-in monitors for coroutine locking.

Kotlin DOES have synchronized(obj) (inherited from JVM), but it blocks threads — wrong tool for coroutines.

class Account(val id: AccountId, initialBalance: Money) {
    var balance = initialBalance
        private set
    val mutex = Mutex()  // explicit field, not built into the object
    
    suspend fun debit(amount: Money) = mutex.withLock {
        require(balance >= amount) { "Insufficient funds" }
        balance -= amount
    }
}

// StateFlow — reactive state holder:
class AccountViewModel {
    private val _balance = MutableStateFlow(Money(0, "EUR"))
    val balance: StateFlow<Money> = _balance   // expose as read-only

    suspend fun updateBalance(new: Money) {
        _balance.value = new                    // emit new value
    }
}

// Collector:
viewModel.balance.collect { money ->
    println("Balance: ${money.amount}")
}
// Immediately receives current value (0 EUR)
// Then receives every update

// StateFlow properties:
// - Always has a value (.value property)
// - New collector immediately gets current value
// - distinctUntilChanged built-in (same value not re-emitted)
// - Conflated: if value changes 3 times before collector processes, it only sees latest

// SharedFlow — event stream:
class EventBus {
    private val _events = MutableSharedFlow<Event>()
    val events: SharedFlow<Event> = _events

    suspend fun emit(event: Event) {
        _events.emit(event)
    }
}

// Multiple collectors — all receive the same event:
launch { eventBus.events.collect { handleA(it) } }
launch { eventBus.events.collect { handleB(it) } }
eventBus.emit(TransferCompleted(...))
// Both handleA and handleB receive TransferCompleted

// SharedFlow properties:
// - No initial value (unlike StateFlow)
// - replay = 0: new collectors don't see past events
// - replay = N: new collectors see last N events
// - NOT distinctUntilChanged (same event emitted multiple times)

When to use each:

StateFlow: observable state — balance, user profile, loading status. Always has a current value. Latest value matters, intermediate values can be skipped.

SharedFlow: events — transfer completed, notification received, error occurred. No "current" state. Each event must be processed. Replay for late subscribers.

Regular Flow: one-shot data streams — database query results, file reading, API responses. Cold, per-collector execution.

// awaitAll — wait for results from async:
val results: List<User> = coroutineScope {
    val deferreds: List<Deferred<User>> = userIds.map { id ->
        async { fetchUser(id) }
    }
    deferreds.awaitAll()  // returns List<User> when all complete
}
// Use when you NEED the return values

// joinAll — wait for completion of launch:
coroutineScope {
    val jobs: List<Job> = payments.map { payment ->
        launch { processPayment(payment) }
    }
    jobs.joinAll()  // returns Unit when all complete
}
// Use when you just need to wait for side effects to finish

Error behavior: both throw if any child fails (assuming coroutineScope). awaitAll() throws the first exception. joinAll() also throws if a job failed.

C# equivalent: awaitAll() = Task.WhenAll(tasks) (returns results). joinAll() = Task.WhenAll(tasks) with void tasks (just waits).

// BAD — GlobalScope:
fun handleRequest(req: Request) {
    GlobalScope.launch {
        processInBackground(req)
    }
    // Problems:
    // 1. Who waits for this? Nobody. If server shuts down, coroutine may be mid-operation.
    // 2. Who handles errors? Nobody. Exception goes to stderr.
    // 3. Who cancels this? Nobody. Even after response is sent, it keeps running.
    // 4. Memory leak potential — coroutine holds references indefinitely.
}

// GOOD — structured scope:
suspend fun handleRequest(req: Request) = coroutineScope {
    launch {
        processInBackground(req)
    }
    // coroutineScope waits for launch to complete.
    // If request is cancelled, launch is cancelled.
    // If launch fails, handleRequest fails.
    // No orphans.
}

When GlobalScope is TECHNICALLY acceptable:

Application-level singletons that live for the entire app lifecycle: logging, metrics publishing. But even then, a custom standalone CoroutineScope with proper error handling and shutdown is better. GlobalScope is essentially CoroutineScope(EmptyCoroutineContext) with no Job, no handler, no dispatcher specified.

IntelliJ inspection: using GlobalScope shows a warning: "Structured concurrency is preferred." The Kotlin team discourages its use in kotlinx.coroutines documentation.

C# async/await                    Kotlin coroutines
---------------------------------------------------------
async Task<T> Method()            suspend fun method(): T
await task                         direct call (implicit await)
Task.Run(() => ...)                launch { }
Task<T>                           Deferred<T>
Task.WhenAll(t1, t2)               awaitAll() / joinAll()
CancellationToken                  Automatic (structured concurrency)
ConfigureAwait(false)              withContext(Dispatchers.IO)
IAsyncEnumerable<T>               Flow<T>
lock(obj)                          Mutex.withLock { }
SemaphoreSlim                      Semaphore
Channel<T>                         Channel<T>

Kotlin advantages:

Structured concurrency: parent cancels children automatically. In C#, you must thread CancellationToken through every method manually. Forget once = task runs forever.

No ConfigureAwait: C# developers must write .ConfigureAwait(false) on every await in library code to avoid deadlocks with SynchronizationContext. Kotlin has explicit Dispatchers — no ambient context to worry about.

suspend in type system: the compiler enforces that suspend functions are only called from suspend contexts. C# allows calling async methods without awaiting — a common mistake that silently drops the result.

C# advantages:

True async I/O: C#'s HttpClient.GetAsync() uses OS-level non-blocking I/O (IOCP). No thread blocked at any point. Kotlin/JVM: most I/O (JDBC) is blocking — needs Dispatchers.IO thread pool. Only Ktor and R2DBC are truly non-blocking.

Simpler mental model: await is explicit — you see every suspension point. In Kotlin, calling fetchUser() looks like a normal function call but actually suspends. Readability trade-off.

Mature ecosystem: async/await since C# 5 (2012). Kotlin coroutines stable since 2018. C# has 10 more years of async library maturation.

The complete C# to Kotlin mapping:

// C#:                              // Kotlin:

async Task<User> Fetch()           suspend fun fetch(): User
await Fetch()                       fetch()  // implicit await!
Task.Run(() => Work())              launch { work() }
var t = FetchAsync()                val d = async { fetch() }
var u = await t                     val u = d.await()
Task.WhenAll(t1, t2)                listOf(d1,d2).awaitAll()
CancellationToken token             automatic! (structured concurrency)
token.ThrowIfCancellationReq()      ensureActive()
token.IsCancellationRequested       isActive
ConfigureAwait(false)               withContext(Dispatchers.IO)
Task.Delay(1000)                    delay(1000)
lock(obj) { }                       mutex.withLock { }
SemaphoreSlim(10, 10)               Semaphore(10)
Channel<T>                          Channel<T>
IAsyncEnumerable<T>                 Flow<T>
await foreach (var x in stream)     flow.collect { x -> }

What C# DOESN'T have (Kotlin's unique advantage):

1. Structured concurrency: coroutineScope { } automatically waits for all children AND cancels them if one fails. In C#: manual WhenAll + manual CancellationTokenSource + manual linking. Forget any piece = orphaned tasks or resource leaks.

2. SupervisorScope: independent child failures without cascading cancellation. C# has no built-in equivalent — you implement it manually with try/catch per task.

3. suspend in type system: suspend fun tells the compiler "this function can pause." Calling it from non-suspend code = compile error. C# allows calling async methods without await — only a warning, not an error.

For the interview: "I have extensive C# async/await experience which maps directly to Kotlin coroutines. Same state machine compilation, same concept. The key improvements Kotlin adds: structured concurrency eliminates manual CancellationToken threading, no ConfigureAwait issues, and suspend is enforced by the compiler. My 25 years of async programming in C# transfers directly."