AbruptTerminationException happens when the actor running the stream (the
"graph interpreter") is terminated without the stream first completing or
failing, for example if terminating an actor system with a running stream
in it.
The paths of the supervisors are pretty much an implementation detail, but
perhaps you can provide some addition to the string "end-of-lookup" to give
you context in the log entry?
--
Johan
Akka Team
On Wed, Nov 9, 2016 at 6:33 PM, Richard Rodseth <rrods...@gmail.com> wrote:
> I'm trying to write two ScalaTest tests to compare two approaches to
> driving a Flow that sends out emails. In one approach the flow is sourced
> using eventsByTag from akka-persistence-query. In the other, I uses
> SourceQueue.offer after receiving a message via AtLeastOnceDelivery. The
> test ends with a check of mock JavaMail mailbox size.
>
> eventually {
>
> mailBoxSize("ema...@foo.com") should be(expectedEmails)
>
> }
>
> eventually {
>
> mailBoxSize("ema...@foo.com") should be(expectedEmails)
>
> }
>
> The test that uses eventsByTag to drive the flow succeeds with 10000
> events. The one that uses SourceQueue.offer gets stuck at 11. In the flow
> below, the println appears 12 times but the log("end-of-lookup") only
> appears 11 times.
>
> def flowLookupEmails()(implicit lookup: EmailsByDefinition, ec:
> ExecutionContext): Flow[AlarmOccurrence, AlarmOccurrenceWithRecipients,
> NotUsed] = {
> val parallelism = 5
> val result = Flow[AlarmOccurrence].mapAsync(parallelism) { occurrence
> =>
> val emailsF: Future[List[String]] = lookup.recipientsForDefinition(
> occurrence.definitionId)
> val lookupResult: Future[AlarmOccurrenceWithRecipients] =
> emailsF.map { emails =>
> println(s"##### Found emails $emails")
> AlarmOccurrenceWithRecipients(occurrence, emails)
> }
>
> lookupResult
> }
> .log("end-of-lookup").withAttributes(Attributes.logLevels(onElement =
> Logging.InfoLevel))
> .named("lookup-emails")
> result
> }
>
> When I turn on DEBUG logging, I get a lot of noise, but do see
>
> [DEBUG] [11/09/2016 08:58:02.827] [OccurrencesIntegrationSpec-
> akka.actor.default-dispatcher-12] [akka://OccurrencesIntegrationSpec/
> user/StreamSupervisor-3/flow-25-0-unknown-operation] stopped
>
> [ERROR] [11/09/2016 08:59:02.455] [OccurrencesIntegrationSpec-
> akka.actor.default-dispatcher-21] [akka.stream.Log(akka://
> OccurrencesIntegrationSpec/user/StreamSupervisor-0)] [end-of-lookup]
> Upstream failed. (akka.stream.AbruptTerminationException)
> Any ideas? What are some good techniques to find the cause of the
> AbruptTerminationException, or name the StreamSupervisors so I know which
> stream they are associated with?
>
> Thanks.
>
> --
> >>>>>>>>>> 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 akka-user+unsubscr...@googlegroups.com.
> To post to this group, send email to akka-user@googlegroups.com.
> 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 akka-user+unsubscr...@googlegroups.com.
To post to this group, send email to akka-user@googlegroups.com.
Visit this group at https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
