When you first dive into Akka Streams, it's easy to treat a Source like it's just a lazy List or an Iterator. I've seen a lot of developers make this mistake—they think "streaming" just means "I don't have to load the whole file into memory." But the real power of Akka Streams isn't about memory efficiency; it's about backpressure. If you ignore backpressure, you aren't actually streaming; you're just building a very expensive way to crash your JVM with an OutOfMemoryError.

Let's look at a real scenario: imagine we're building a pipeline that reads a massive stream of user activity logs and enriches each log entry by fetching user metadata from a remote database. It seems straightforward—you read the log, call the DB, and push the result to a sink.

The Danger of Unbounded Futures

The most common "naive" mistake I see is using a standard .map to trigger an asynchronous call. It looks like this:

val logProcessingFlow = Source(logEntries)
  .map { log =>
    // I'm calling a DB service that returns a Future
    userService.getUserMetadata(log.userId).map { metadata =>
      EnrichedLog(log, metadata)
    }
  }
  .to(Sink.foreach(println))
  .run()


At first glance, this looks fine. But look closer: .map is synchronous. In the example above, the stream is actually producing a stream of Future[EnrichedLog], not a stream of EnrichedLog. If you then try to "await" those futures downstream or just push them into a sink, you've broken the fundamental contract of Akka Streams. The source will pump data as fast as it possibly can, spawning thousands of Future objects that all hit your database simultaneously. You've effectively created a self-inflicted Denial of Service attack on your own database.

Taming the Flow with mapAsync
To do this properly, you need to tell the stream that the transformation is asynchronous and, more importantly, how many of these operations can happen at once. This is where mapAsync comes in. I prefer this approach because it forces you to be explicit about your concurrency limits.

val logProcessingFlow = Source(logEntries)
  .mapAsync(parallelism = 4) { log =>
    userService.getUserMetadata(log.userId).map { metadata =>
      EnrichedLog(log, metadata)
    }
  }
  .to(Sink.foreach(println))
  .run()


By setting parallelism = 4, you're telling Akka: "You can have up to four database requests in flight at any given time. Do not pull a fifth element from the source until one of these four completes." This is the essence of backpressure. The mapAsync stage signals the Source to slow down. If the database starts lagging, the mapAsync buffer fills up, and the source is told to stop emitting. Your application remains stable regardless of the data volume.

When the Sink Can't Keep Up
Now, you might wonder: "What if the database is fast, but the thing I'm writing to (the Sink) is slow?" Maybe you're writing to a legacy API that can only handle 10 requests per second. If you use mapAsync to fetch data quickly but your Sink is sluggish, the data will start to pile up between those two stages.

In a standard Scala collection, you'd just have a growing queue in memory. In Akka Streams, the "slow" sink sends a signal upstream. It tells the mapAsync stage, "I'm full, stop sending me stuff." The mapAsync stage then tells the Source, "I'm full, stop sending me stuff." This ripple effect is what prevents your heap from exploding. It's a pull-based model: the sink pulls from the flow, which pulls from the source. Nothing moves unless there is demand at the end of the pipe.

I've spent a lot of time debugging production outages caused by "fire-and-forget" async patterns. The lesson here is simple: whenever you're crossing a boundary—whether it's a DB call, an API request, or a heavy computation—don't just wrap it in a Future inside a map. Use the operators designed for asynchronous boundaries so that your stream can breathe.

📋 Practical Task

Exercise: Implementing a Rate-Limited Log Enricher

You are tasked with fixing a broken log processing stream. Currently, the stream uses map to trigger a Future, which is overwhelming a downstream Rate-Limited API. The API allows a maximum of 5 concurrent requests; any more will result in 429 Too Many Requests errors.

Your Task:

• Refactor the provided code to replace the naive .map with .mapAsync.
• Set the parallelism to 5 to match the API limit.
• Ensure the stream correctly handles the Future[EnrichedLog] so that the final Sink receives EnrichedLog objects, not Future objects.

// BROKEN CODE TO FIX
import akka.stream.scaladsl._
import scala.concurrent.Future
import scala.concurrent.ExecutionContext.Implicits.global

case class Log(id: String, userId: String)
case class EnrichedLog(id: String, userEmail: String)

class UserApi {
  def fetchEmail(userId: String): Future[String] = Future {
    // Simulate API call
    Thread.sleep(100) 
    s"$userId@example.com"
  }
}

val api = new UserApi()
val logs = Source(1 to 100).map(i => Log(i.toString, s"user_$i"))

// FIX THE LINE BELOW
val stream = logs
  .map { log => 
    api.fetchEmail(log.userId).map(email => EnrichedLog(log.id, email))
  }
  .to(Sink.foreach(result => println(s"Processed: $result")))
  .run()