Build Concurrent Data Workflows with Kotlin Coroutines

Table of Contents

1. Challenge

   Step 1: Model the data sources

   Many Kotlin services fetch data from multiple independent sources (APIs, microservices, database shards), and the sequential approach silently taxes every user who waits for each call to finish before the next one starts. If five warehouse feeds each take 300 ms, sequential calls cost 1.5 seconds. A coroutine-based fan-out finishes in roughly 300 ms.

   Coroutines let you run those calls concurrently without manually managing threads or writing callback chains. They also give you built-in tools for the failure cases that make concurrent code hard in practice: per-call timeouts, structured error isolation, and clean cancellation semantics.

   In this lab, you'll put those patterns to work in a supply chain analytics CLI that fetches inventory data from several simulated warehouse feeds in parallel, isolates slow or failing feeds using timeouts and supervisor scope, and writes an aggregated report to the console and an output file.

   The application directory contains a pre-configured Gradle project with source and test files already created, each with package declarations and stub bodies ready for your implementations. In this first step, you'll define the data types that represent a warehouse and its inventory result, populate the default dataset the pipeline will use, and implement the suspend function that simulates each warehouse feed.

   Feeling stuck? Check out the matching solution/stepN/ folder for the step you're on. Give it a try on your own first: the solution is your safety net, not your starting point.

   info> This lab experience was developed by the Pluralsight team using an internally developed AI tool. All sections were verified by human experts for accuracy prior to publications. However, content may still contain errors or inaccuracies, and we recommend independent verification.
   
   To report a problem or provide feedback, click here. Feedback may be used to improve accuracy in accordance with our Privacy Policy.

   Modeling a domain entity with a data class

   Kotlin's data class is the standard choice for plain data carriers. When you declare a class with the data modifier, the compiler automatically generates equals, hashCode, toString, and copy implementations based on the constructor properties. The copy function is particularly useful for concurrency testing because it lets you create a modified version of a value without mutating the original.

   The Warehouse type needs three fields: an id string to uniquely identify each source, a delayMs long to control how long the simulated feed takes to respond, and an itemCount integer as the inventory value it reports.

   warning> The lab environment on the right may take up to a minute to finish loading. If you don't see the terminal or project files yet, please wait a moment before continuing. ### Carrying the source ID in the result type

   When a pipeline fans out to multiple sources simultaneously, each result needs to carry enough information to identify where it came from. Returning a raw count isn't enough: you need to know which warehouse produced it so you can report on individual feed health and correlate results back to their source.

   InventorySnapshot includes a warehouseId field (a copy of the originating warehouse's id) alongside the itemCount. Because it carries related fields together and you'll want equality and copy semantics when writing tests, it belongs in a data class as well. ### Using varied delays to make concurrency observable

   The real value of running warehouse feeds in parallel is that the total wall-clock time approaches the slowest single feed, not the sum of all feeds. To make that distinction visible in tests, your dataset needs delays spread far enough apart that sequential and concurrent execution produce measurably different elapsed times.

   If the longest single delay is well under the sum of all delays, a test can assert that the pipeline finished near the longest delay rather than the sum, which proves the feeds ran concurrently. A list where the sum of delays exceeds 1500 ms but no single delay exceeds 800 ms gives you that clear boundary. ### Simulating latency with a suspend function

   The delay function from kotlinx.coroutines is the idiomatic way to introduce a wait in a coroutine. Unlike Thread.sleep, it suspends the coroutine without blocking the underlying thread: execution is paused, the thread is freed to run other coroutines, and the coroutine resumes when the delay elapses.

   In tests, delay is intercepted by TestCoroutineScheduler. When you advance the scheduler, virtual time moves forward instantly rather than waiting for real wall-clock time. This makes timing-sensitive tests fast and deterministic.

2. Challenge

   Step 2: Launch concurrent fetches

   The data model and simulated feed are in place, giving you a fetchInventory function that correctly models variable latency and returns a typed snapshot. Now, you'll build the pipeline layer that turns individual suspend calls into a concurrent fan-out. By the end of this step, runPipeline will launch every warehouse feed at the same time and collect all results, finishing in roughly the time of the slowest single feed rather than the sequential sum.

   Launching feeds concurrently with coroutineScope and async

   coroutineScope creates a child scope that suspends the calling coroutine until every coroutine launched inside it completes. If any child throws an unhandled exception, the scope cancels the remaining children and rethrows. async starts a coroutine and returns a Deferred<T> rather than a result. Because nothing awaits the deferred immediately, all the launched coroutines run at the same time.

   awaitAll() collects all deferred values. It's preferred over map { it.await() } because it cancels any remaining deferreds if one fails, rather than waiting for them one at a time.

   The typical shape of a concurrent fan-out looks like this:

   return coroutineScope {
       items.map { item ->
           async { process(item) }
       }.awaitAll()
   }
   ``` ### Making the dispatcher injectable for tests
   
   `CoroutineDispatcher` controls which thread pool a coroutine runs on. Accepting one as a parameter with a sensible default keeps the production behavior correct while letting tests pass a `StandardTestDispatcher` to gain precise control over virtual time and execution order.
   
   `withContext(dispatcher)` switches the coroutine to the given dispatcher for the duration of its block, then switches back automatically when it returns. Wrapping the body of `runPipeline` in `withContext(dispatcher)` means the parameter is not just accepted but actually used, making the dispatcher swap observable in tests.

3. Challenge

   Step 3: Add resilience with timeouts and supervisor scope

   The concurrent pipeline fetches all feeds in parallel, but a feed that hangs indefinitely will block the entire pipeline and an uncaught exception propagates across sibling coroutines. In this step, you'll introduce a typed result hierarchy to represent feed outcomes, enforce per-feed time limits, and switch to supervisorScope so a failing feed cannot cancel its neighbors. By the end of this step, runPipeline returns one typed result per warehouse regardless of whether each feed succeeded, timed out, or threw an exception.

   Typed results with sealed interfaces

   A sealed interface restricts all implementations to the same compilation unit. The compiler uses this to enforce exhaustive when expressions: if you add a third case later, every when that does not handle it becomes a compile error at the call site. This makes sealed types the standard Kotlin pattern for representing a closed set of outcomes.

   FeedResult.Success wraps a completed snapshot; FeedResult.Unavailable carries the originating warehouse ID and a human-readable reason so callers can report on what went wrong. Making them both data class implementations means you get equality and structured access for free. ### Preserving cooperative cancellation when catching exceptions

   Kotlin's structured concurrency uses CancellationException as the signal that a coroutine should stop. If a broad catch (e: Exception) block swallows it, the coroutine silently ignores the cancellation request and the parent scope can no longer clean up reliably.

   The rule is: always rethrow CancellationException before handling any other exception type. Because TimeoutCancellationException is a subtype of CancellationException, the catch ordering matters: a catch (e: CancellationException) clause placed first would also catch timeouts, preventing them from being handled separately. ### Catch clause ordering with TimeoutCancellationException

   TimeoutCancellationException is a subtype of CancellationException. Kotlin evaluates catch clauses top to bottom, and the first matching clause wins. If catch (e: CancellationException) appears before a TimeoutCancellationException clause, timeouts are rethrown instead of converted to Unavailable results.

   The correct ordering is: TimeoutCancellationException first, then CancellationException (rethrow), then Exception. ### Isolating failures with supervisorScope

   In a regular coroutineScope, an unhandled exception in any child cancels all siblings and rethrows in the parent. supervisorScope breaks that link: each child has an independent failure policy, so one cancelled or failed coroutine does not affect the rest.

   This is the right choice when child jobs represent independent units of work, as each warehouse feed does here. The API is a drop-in replacement for coroutineScope; the only change is the name.

4. Challenge

   Step 4: Aggregate results and write the report

   The pipeline now returns a typed list where each entry is either a Success or Unavailable, giving you a complete picture of the run even when some feeds fail. In this step you'll implement the aggregation function that computes inventory totals from the mixed-result list, the report writer that formats and persists the output, and the main function that wires everything together. The step ends with the pipeline running end-to-end from the CLI.

   Aggregating a mixed-result list in a single pass

   When you need to compute multiple metrics from a single collection (total items from successes, count of successes, count of unavailables), iterating the list once and accumulating all three values together is more efficient than filtering separately for each metric. Using a when expression to branch on is FeedResult.Success and is FeedResult.Unavailable gives you an exhaustive match because FeedResult is sealed. ### Choosing the right dispatcher for blocking I/O

   Dispatchers.Default is sized for CPU-bound work and uses a limited thread pool; blocking it with I/O operations starves coroutines waiting for CPU time. Dispatchers.IO uses a separate, expandable pool designed for blocking I/O calls such as file writes.

   withContext(Dispatchers.IO) { ... } switches the coroutine to the IO pool for the duration of the block and returns automatically when it completes. Wrapping only the file-write line (rather than the entire function body) keeps the dispatch scope as small as possible. ### suspend fun main as the coroutine entry point

   Kotlin allows main to be declared as a suspend function, which makes it the top-level coroutine entry point without requiring runBlocking. This is the idiomatic choice for coroutine-first CLI apps: declaring suspend fun main() tells the Kotlin compiler to generate a blocking bridge automatically, keeping the application entry point clean.

5. Challenge

   Step 5: Validate concurrent behavior

   The application runs correctly end-to-end, but concurrent code has subtle timing behaviors that informal observation cannot reliably surface. Feeds appearing to run in parallel on your machine doesn't prove they always will, and catching a supervisorScope regression manually is nearly impossible. In this step, you'll use runTest and TestCoroutineScheduler to write three focused automated tests: one that proves concurrent execution using virtual time, one that proves timeout isolation, and one that proves aggregation correctness with deterministic inputs.

   Asserting concurrency with virtual time

   runTest replaces the real clock with a TestCoroutineScheduler that you control. When you create a StandardTestDispatcher(testScheduler) and pass it to runPipeline, every delay call inside the pipeline is intercepted by the scheduler rather than waiting for real time to pass.

   currentTime inside runTest reports how much virtual time has elapsed. After calling runPipeline with a warehouse list where the sum of all delays is much larger than the longest single delay, currentTime should be close to the longest single delay, not the sum. That gap is the proof of concurrent execution.

   StandardTestDispatcher does not auto-advance virtual time. Time only moves when the coroutine suspends or you call testScheduler.advanceUntilIdle(). Inside a runTest block, idle advancement happens automatically when you await or awaitAll, so you typically do not need to call it manually. ### What this test is actually proving

   The timeout isolation test has two distinct assertions, and the second one is more important than the first. Asserting that the slow warehouse returns Unavailable verifies that withTimeout is working. Asserting that the fast warehouses return Success verifies supervisorScope. If coroutineScope were still in place, the timeout cancellation would propagate to siblings and they would not complete, so the Success assertions would fail even if the Unavailable assertion passed. Both checks are needed; neither is redundant. ### The natural evolution: Kotlin Flow

   The pipeline in this lab uses async/awaitAll() to collect all results into a List<FeedResult> before writing the report. This is the right model when you need a complete snapshot first. In production systems, results often need to be processed as they arrive: streaming a large dataset, updating a UI incrementally, or forwarding to a downstream consumer before every source has responded.

   kotlinx.coroutines provides Flow<T> for this pattern. A Flow-based pipeline emits each FeedResult as it completes rather than blocking on awaitAll(). flatMapMerge is the Flow equivalent of async/awaitAll() for concurrent fan-out, and the same exception-handling rules apply: TimeoutCancellationException caught first, other CancellationException rethrown, remaining Exception types mapped to failure values.

   Flow is the idiomatic Kotlin choice for any async sequence of values. The async/awaitAll() approach you built here is the right foundation for understanding structured concurrency; Flow is where most production pipeline code lives and is the natural next step. You've built a fully concurrent inventory pipeline in Kotlin using structured concurrency, per-feed timeouts, and isolated failure handling. The pipeline fetches data from multiple simulated warehouse feeds in parallel, converts slow or failing feeds into typed Unavailable results without cancelling sibling coroutines, and aggregates the outcomes into a formatted console and file report.

   Along the way you applied the core coroutines patterns that appear in real production systems: async/awaitAll() for concurrent fan-out, withContext for dispatcher injection, withTimeout and supervisorScope for resilience, and Dispatchers.IO for safe blocking I/O. You also practiced using runTest and TestCoroutineScheduler to write timing-sensitive tests that prove parallel execution and failure isolation with deterministic virtual-time assertions, a skill that translates directly to testing any coroutine-based code.