The details here are probably too subtle to be useful as a system design tool: there are two kinds of failure handling at play in this example—a caught exception and propagation of teardown—and only one of them is reasonably controlled by the supervisionStrategy. I would recommend using `.recover()` to turn a failed stream into an explicit element instead of relying upon supervision for non-local failure handling.
Regards, Roland > 29 maj 2017 kl. 15:45 skrev Martynas Mickevičius > <[email protected]>: > > 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] > <mailto:[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/ <http://akka.io/docs/> > >>>>>>>>>> Check the FAQ: > >>>>>>>>>> http://doc.akka.io/docs/akka/current/additional/faq.html > >>>>>>>>>> <http://doc.akka.io/docs/akka/current/additional/faq.html> > >>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user > >>>>>>>>>> <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] > <mailto:[email protected]>. > To post to this group, send email to [email protected] > <mailto:[email protected]>. > Visit this group at https://groups.google.com/group/akka-user > <https://groups.google.com/group/akka-user>. > For more options, visit https://groups.google.com/d/optout > <https://groups.google.com/d/optout>. > > -- > >>>>>>>>>> Read the docs: http://akka.io/docs/ <http://akka.io/docs/> > >>>>>>>>>> Check the FAQ: > >>>>>>>>>> http://doc.akka.io/docs/akka/current/additional/faq.html > >>>>>>>>>> <http://doc.akka.io/docs/akka/current/additional/faq.html> > >>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user > >>>>>>>>>> <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] > <mailto:[email protected]>. > To post to this group, send email to [email protected] > <mailto:[email protected]>. > Visit this group at https://groups.google.com/group/akka-user > <https://groups.google.com/group/akka-user>. > For more options, visit https://groups.google.com/d/optout > <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.
