Thanks.
I actually found the bug that was causing my stream to get stuck. I was
running a Source.queue in an actor, and calling context.become (without
stashing) immediately afterwards.
I found that one of the stream elements was reaching the initial receive
handler, and thus not getting acked.
On Fri, Nov 11, 2016 at 3:48 AM, Akka Team <akka.offic...@gmail.com> wrote:
> 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-ak
>> ka.actor.default-dispatcher-12] [akka://OccurrencesIntegration
>> Spec/user/StreamSupervisor-3/flow-25-0-unknown-operation] stopped
>>
>> [ERROR] [11/09/2016 08:59:02.455] [OccurrencesIntegrationSpec-ak
>> ka.actor.default-dispatcher-21] [akka.stream.Log(akka://Occurr
>> encesIntegrationSpec/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/c
>> urrent/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.
>
--
>>>>>>>>>> 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.
