Akka performer Akka fails exception when child actor throws exception in future onFailure
I'm having a problem with the Akka akker actor. When the child actor throws an exception in the onFailure method for the future result, the supervisor does not handle the error (I want to restart the child in case of a ConnectException).
I am using Akka 2.3.7.
This is the observer actor:
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)
}
}
}
And this is a child actor:
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)
}
}
I tried to notify the sender with akka.actor.Status.Failure (e) as suggested here but didn't work, the exception is not handled by the supervisor.
As a workaround, I found this way to get it working:
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)
}
}
Is this an Akka bug or am I doing something wrong?
Thank!
source to share
I think the problem is that the exception thrown inside the Future does not belong to the same thread (potentially) as the one the Actor is running on (someone more experienced can figure this out). So the problem is that the exception thrown inside the body of the Future is "swallowed" and does not apply to the Actor. Since this is the case, the Actor does not fail and therefore there is no need to apply a surveillance strategy. So the first solution that comes to my mind is to wrap the future exception in some kind of message, send it to itself, and then throw the Actor out of context. This time, an Exception will be found and the surveillance strategy will be applied. Note, however, that if you have not already sent the "Submit" (noti) message, you will not see that "The exception "happened because the Actor was restarted. In general, the code will be like this:
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)
}
}
Hope it helped.
source to share