>From the initial look both of these example should behave the same. Especially since you explicitly set the supervision strategy. Could that there is a supervision strategy missing somewhere. What Akka version did you use? There has been quite a bit of changes how attributes are handled in Akka 2.5. I wonder if that is going to make a difference.
Otherwise I think this deserves an ticket on the issue tracker. On Mon, May 29, 2017 at 11:39 AM Stephen Duncan <[email protected]> wrote: > Before I assume and report this as a bug, I wanted to see if there was > something I was missing regarding the error handling with Akka streams > here. I am surprised to see two different outputs for the two example > streams: > > import akka.actor.ActorSystem > import akka.stream.scaladsl.{Flow, Sink, Source} > import akka.stream.{ActorAttributes, ActorMaterializer, Supervision} > > import scala.collection.immutable.Seq > import scala.concurrent.duration._ > import scala.concurrent.{Await, Future} > > object MapMergeConcatError extends App { > implicit val system = ActorSystem("Main") > implicit val materializer = ActorMaterializer() > implicit val ec = system.dispatcher > > val subFlow = { > Flow[Int] > .mapAsyncUnordered(5)(i => Future { > if (i == 4) sys.error("☠") > i * 5 > }) > > .withAttributes(ActorAttributes.supervisionStrategy(Supervision.stoppingDecider)) > .reduce(_ + _) > } > > val subStreamFuture = Source(Seq(Seq(1, 2), Seq(3, 4, 5), Seq(6))) > .flatMapMerge(5, m => > Source.single(m).mapConcat(identity).via(subFlow)) > > .withAttributes(ActorAttributes.supervisionStrategy(Supervision.resumingDecider)) > .runWith(Sink.seq) > > val mapAsyncFuture = Source(Seq(Seq(1, 2), Seq(3, 4, 5), Seq(6))) > .mapAsyncUnordered(5)(m => > Source.single(m).mapConcat(identity).via(subFlow).runWith(Sink.head)) > > .withAttributes(ActorAttributes.supervisionStrategy(Supervision.resumingDecider)) > .runWith(Sink.seq) > > val f1 = Await.ready(subStreamFuture, 10.seconds) > val f2 = Await.ready(mapAsyncFuture, 10.seconds) > > println(s"Using flatMapMerge: $f1") > println(s"Using mapAsyncUnordered: $f2") > > system.terminate() > } > > The output is: > Using flatMapMerge: Future(Failure(java.lang.RuntimeException: ☠)) > Using mapAsyncUnordered: Future(Success(Vector(30, 15))) > The mapAsyncUnordered output is the desired output (the whole item from > the top-level is dropped when there is a failure in the sub-flow, but the > other items that did not have a failure in the sub-flow make it through). > > Is this a bug, or something subtle that could be explained? > > -- > >>>>>>>>>> Read the docs: http://akka.io/docs/ > >>>>>>>>>> Check the FAQ: > http://doc.akka.io/docs/akka/current/additional/faq.html > >>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user > --- > You received this message because you are subscribed to the Google Groups > "Akka User List" group. > To unsubscribe from this group and stop receiving emails from it, send an > email to [email protected]. > To post to this group, send email to [email protected]. > Visit this group at https://groups.google.com/group/akka-user. > For more options, visit https://groups.google.com/d/optout. > -- >>>>>>>>>> Read the docs: http://akka.io/docs/ >>>>>>>>>> Check the FAQ: >>>>>>>>>> http://doc.akka.io/docs/akka/current/additional/faq.html >>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user --- You received this message because you are subscribed to the Google Groups "Akka User List" group. To unsubscribe from this group and stop receiving emails from it, send an email to [email protected]. To post to this group, send email to [email protected]. Visit this group at https://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
