👀 Views: 76 💬 Answers: 1 📅 Created: 2025-06-16

akka scala futures streams Scala

I need some guidance on I'm building a feature where I'm stuck trying to I'm working on a personal project and I'm working with inconsistent behavior when trying to handle exceptions in an Akka Streams pipeline that uses Scala's `Future`. I have a flow that processes elements and returns a `Future`, but when an exception is thrown in that future, it seems to not propagate as expected through the stream. For example, consider the following code snippet: ```scala import akka.actor.ActorSystem import akka.stream.scaladsl.{Sink, Source} import akka.{Done, NotUsed} import scala.concurrent.{ExecutionContext, Future} import scala.util.{Failure, Success} implicit val system: ActorSystem = ActorSystem("example") implicit val ec: ExecutionContext = system.dispatcher val source: Source[Int, NotUsed] = Source(1 to 10) val flow = source.mapAsync(4) { num => Future { if (num == 5) throw new RuntimeException("behavior on 5") num * 2 } } val result: Future[Done] = flow.runWith(Sink.ignore) result.onComplete { case Success(_) => println("Stream completed successfully.") case Failure(ex) => println(s"Stream failed with exception: ${ex.getMessage}") } ``` In this example, I expect the stream to unexpected result when it encounters the number 5, but instead, it prints that the stream completed successfully. However, if I change it to use `map` instead of `mapAsync`, the exception propagates correctly. Is there a specific setting or a best practice I should be aware of when dealing with `Future` and exception handling in Akka Streams? I've also tried using `recover` and `recoverWith`, but they don't seem to capture the exception as expected either. Any insights on this would be greatly appreciated! For context: I'm using Scala on Linux. Am I missing something obvious? I'm working in a Linux environment. Any feedback is welcome! Thanks for any help you can provide! Has anyone dealt with something similar?