Classic Fault Tolerance 1 포스트에서 supervisor strategy를 만들고 적용하는 방법에 대해 정리했었다. 이번 포스트에서는 해당 내용들을 적용하는 것을 실제 예제로 다룰 것이다.

Test Application

가장 먼저 예제를 위한 적당한 supervisor 액터를 선언한다.

import akka.actor.Actor

class Supervisor extends Actor {
  import akka.actor.OneForOneStrategy
  import akka.actor.SupervisorStrategy._
  import scala.concurrent.duration._

  override val supervisorStrategy =
    OneForOneStrategy(maxNrOfRetries = 10, withinTimeRange = 1 minute) {
      case _: ArithmeticException      => Resume
      case _: NullPointerException     => Restart
      case _: IllegalArgumentException => Stop
      case _: Exception                => Escalate
    }

  def receive = {
    case p: Props => sender() ! context.actorOf(p)
  }
}

위의 예제는 OneForOneStrategy를 적용하였으며, 최대 10번 재시도 및 최대 1분 동안 실행하는 전략을 가지고 있으며 예외 종류에 따라 행위를 나누었다. 이 supervisor는 자손 액터를 만드는데 사용할 수 있다.

그리고 이를 알아보기 위해 Child라는 액터를 만들었다. 이 액터는 정수형 메시지를 받으면 state를 변경해주며, "get"이라는 메시지를 받으면 현재 state를 보내주는 액터이다.

import akka.actor.Actor

class Child extends Actor {
  var state = 0
  def receive = {
    case ex: Exception => throw ex
    case x: Int        => state = x
    case "get"         => sender() ! state
  }
}

우리는 위의 코드들에서 원하는 대로 예외 처리가 되는지 테스트해 볼 예정이다.
아직 따로 정리는 하지는 않았지만 testing actor system에서 정리할 내용을 미리 사용하였다.

import com.typesafe.config.{ Config, ConfigFactory }
import org.scalatest.BeforeAndAfterAll
import org.scalatest.matchers.should.Matchers
import org.scalatest.wordspec.AnyWordSpecLike
import akka.testkit.{ EventFilter, ImplicitSender, TestKit }

class FaultHandlingDocSpec(_system: ActorSystem)
    extends TestKit(_system)
    with ImplicitSender
    with AnyWordSpecLike
    with Matchers
    with BeforeAndAfterAll {

  def this() =
    this(
      ActorSystem(
        "FaultHandlingDocSpec",
        ConfigFactory.parseString("""
      akka {
        loggers = ["akka.testkit.TestEventListener"]
        loglevel = "WARNING"
      }
      """)))

  override def afterAll(): Unit = {
    TestKit.shutdownActorSystem(system)
  }

  "A supervisor" must {
    "apply the chosen strategy for its child" in {
      // 내용
    }
  }
}

위의 코드는 supervisor에 선언한 정책이 Child 액터에 적용되었는지 테스트하는 코드이다. 액터를 생성하고, 여러 메시지들을 Child 액터에게 보내본다.

val supervisor = system.actorOf(Props[Supervisor](), "supervisor")

supervisor ! Props[Child]()
val child = expectMsgType[ActorRef]

child ! 42
child ! "get"
expectMsg(42)

child ! new ArithmeticException 
child ! "get"
expectMsg(42)

child ! new NullPointerException 
child ! "get"
expectMsg(0)

watch(child)
child ! new IllegalArgumentException // break it
expectMsgPF() { case Terminated(`child`) => () }

supervisor ! Props[Child]() 
val child2 = expectMsgType[ActorRef]
watch(child2)
child2 ! "get" // verify it is alive
expectMsg(0)

child2 ! new Exception("CRASH") 
expectMsgPF() {
  case t @ Terminated(`child2`) if t.existenceConfirmed => ()
}

val child = expectMsgType[ActorRef]를 통해 Testkit의 결과값을 얻어오는 역할을 한다.
이제 이렇게 만든 액터에 여러가지 메시지를 보낼 것이다.

이때 눈에 띄는 점은 new ArithmeticException이나 new NullPointerException과 같은 예외를 액터에게 보내는 점이다. 하지만, 이러한 예외들을 보내면 액터는 crash를 보낼 것이다. 특히, NullPointerException을 보낸다면 심하게 crash날 것이다.

그리고, new IllegalArgumentException을 보내면 그제서야 supervisor를 멈출것이다.

여기까지만 보면 supervisor가 실패를 escalate하는 것이 아니라고 생각할 수 있다.

supervisor 자체는 top-level 시스템인 ActorSystem에 의해 감독 받는다. 즉, ActorSystem의 기본 정책인 Exception case에 대해 재시작하는 것에 영향을 받는것이다. 디폴트 방향이 자손 액터를 모두 죽이는 것이므로 우리는 failure에 빠진 자손 액터들이 살아남지 못하는 것이다.

Reference

• Akka Classic Fault Tolerance
• Testing Classic Actors