Актер супервизора Akka не обрабатывает исключение, когда дочерний актор создает исключение в onFailure of the future

Question

У меня проблема с актером-супервайзером Акки. Когда дочерний субъект генерирует исключение в методе onFailure будущего результата, супервизор не обрабатывает ошибку (я хочу перезапустить дочерний элемент в случае исключения ConnectException).

Я использую Акку 2.3.7.

Это руководитель актера:

class MobileUsersActor extends Actor with ActorLogging {

  import Model.Implicits._
  import Model.MobileNotifications

  override val supervisorStrategy =
    OneForOneStrategy(maxNrOfRetries = 3, withinTimeRange = 1 minute) {
      case _: java.net.ConnectException => {
        Logger.error("API connection error. Check your proxy configuration.")
        Restart
      }
    }

  def receive = {
    case Start => findMobileUsers
  }

  private def findMobileUsers = {
    val notis = MobileNotificationsRepository().find()
    notis.map(invokePushSender)
  }

  private def invokePushSender(notis: List[MobileNotifications]) = {
    notis.foreach { n =>
      val pushSender = context.actorOf(PushSenderActor.props)
      pushSender ! Send(n)
    }
  }

}

А это детский актер

class PushSenderActor extends Actor with ActorLogging {

  def receive = {
    case Send(noti) => {
      val response = sendPushNotification(noti) onFailure {
        case e: ConnectException => throw e
      }
    }
  }

  private def sendPushNotification(noti: MobileNotifications): Future[WSResponse] = {
    val message = "Push notification message example"
    Logger.info(s"Push Notification >> $message to users " + noti.users)
    PushClient.sendNotification(message, noti.users)
  }

}

Я пытался уведомить отправителя с помощью akka.actor.Status.Failure(e), как предлагается здесь, но не сработало, исключение остается необработанным супервизором.

В качестве обходного пути я нашел способ заставить его работать:

class PushSenderActor extends Actor with ActorLogging {

  def receive = {
    case Send(noti) => {
      val response = sendPushNotification(noti) onFailure {
        case e: ConnectException => self ! APIConnectionError
      }
    }
    case APIConnectionError => throw new ConnectException
  }

  private def sendPushNotification(noti: MobileNotifications): Future[WSResponse] = {
    val message = "Push notification message example"
    Logger.info(s"Push Notification >> $message to users " + noti.users)
    PushClient.sendNotification(message, noti.users)
  }

}

Это ошибка Акки или я что-то не так делаю?

Спасибо!

Answer 1

Я думаю, что проблема заключается в том, что исключение, генерируемое в Future, не относится к тому же потоку (потенциально), что и тот, на котором работает Актер (кто-то более опытный может уточнить это). Итак, проблема в том, что исключение, выброшенное внутри тела Будущего, "проглатывается" и не распространяется на Актера. Так как это так, Актер не терпит неудачу, и поэтому нет необходимости применять стратегию надзора. Итак, первое решение, которое приходит мне в голову, - обернуть исключение внутри Будущего в какое-то сообщение, отправить его себе, а затем выбросить изнутри самого контекста Actor. На этот раз будет выявлено исключение и будет применена стратегия надзора. Тем не менее, обратите внимание, что если вы не отправите сообщение Send(noti) снова, вы не увидите исключения, произошедшего после перезапуска Actor. В общем, код будет выглядеть так:

class PushSenderActor extends Actor with ActorLogging {

  case class SmthFailed(e: Exception)

  def receive = {
    case Send(noti) => {
      val response = sendPushNotification(noti) onFailure {
        case e: ConnectException => self ! SmthFailed(e) // send the exception to yourself
      }
    }

    case SmthFailed(e) =>
      throw e // this one will be caught by the supervisor
  }

  private def sendPushNotification(noti: MobileNotifications): Future[WSResponse] = {
    val message = "Push notification message example"
    Logger.info(s"Push Notification >> $message to users " + noti.users)
    PushClient.sendNotification(message, noti.users)
  }

}

Надеюсь, это помогло.