Почему реки Акка проглатывают мои исключения?

Почему исключение в

import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.Source

object TestExceptionHandling {
  def main(args: Array[String]): Unit = {
    implicit val actorSystem = ActorSystem()
    implicit val materializer = ActorMaterializer()(defaultActorSystem)

    Source(List(1, 2, 3)).map { i =>
      if (i == 2) {
        throw new RuntimeException("Please, don't swallow me!")
      } else {
        i
      }
    }.runForeach { i =>
      println(s"Received $i")
    }
  }
}

молча игнорируется? Я вижу, что поток прекращается после печати Received 1, но ничего не регистрируется. Обратите внимание, что проблема не в конфигурации ведения журнала вообще, так как я вижу много результатов, если я установил akka.log-config-on-start = on в мой файл application.conf.

Ответы

Ответ 1

Теперь я использую пользовательский Supervision.Decider, который позволяет правильно регистрировать исключения, которые можно настроить следующим образом:

val decider: Supervision.Decider = { e =>
  logger.error("Unhandled exception in stream", e)
  Supervision.Stop
}

implicit val actorSystem = ActorSystem()
val materializerSettings = ActorMaterializerSettings(actorSystem).withSupervisionStrategy(decider)
implicit val materializer = ActorMaterializer(materializerSettings)(actorSystem)

Кроме того, как было указано в Vikor Klang, в приведенном выше примере исключение также можно "поймать" через

Source(List(1, 2, 3)).map { i =>
  if (i == 2) {
    throw new RuntimeException("Please, don't swallow me!")
  } else {
    i
  }
}.runForeach { i =>
  println(s"Received $i")
}.onComplete {
  case Success(_) =>
    println("Done")
  case Failure(e) =>
    println(s"Failed with $e")
}

Обратите внимание, что этот подход не поможет вам с

Source(List(1, 2, 3)).map { i =>
  if (i == 2) {
    throw new RuntimeException("Please, don't swallow me!")
  } else {
    i
  }
}.to(Sink.foreach { i =>
  println(s"Received $i")
}).run()

поскольку run() возвращает Unit.

Ответ 2

У меня были подобные вопросы, когда я начал использовать akk-потоки. Supervision.Decider помогает, но не всегда.

К сожалению, он не перехватывает исключения, сброшенные в ActionPublisher. Я вижу, что он обрабатывается, ActorPublisher.onError вызывается, но он не достигает Supervision.Decider. Он работает с простым потоком, представленным в документации.

Ошибки также не достигают актера, если я использую Sink.actorRef.

И ради эксперимента я попробовал следующий образец

val stream = Source(0 to 5).map(100 / _)
stream.runWith(Sink.actorSubscriber(props))

В этом случае исключение было поймано Decider, но так и не было подписано подписчиком.

В целом я считаю это непоследовательным поведением. Я не могу использовать один механизм для обработки ошибок в потоке.

Мой оригинальный вопрос SO: Custom Supervision.Decider не видит исключения, созданного ActorPublisher

И вот проблема akka, где она отслеживается: https://github.com/akka/akka/issues/18359