•  간단한 추상화를 통해 distribution, concurrency, parallelism 을 위한 layer 를 제공
• Asynchronous, non-blocking 으로 구현되어진 높은 성능의 message driven 개발 모델을 제공
• Lighweight 한 event driven 으로 동작(1GB 의 메모리 위에서 수 백만의 actor 들이 동작 가능)

구현 하고자 하는 시스템

이 글의 작성 목적은 외부 또는 내부에서 생성되는 event, web log, application log 등의 데이터 수집을 원활하게 하기 위해서였습니다. 아래는 간략한 구성도입니다.

수집된 데이터를 전달받을 대상은, 최근 대용량 데이터 전송에 주로 사용되고 있는 distributed queue인 Apache Kafka입니다. 위에서 언급한 데이터를 전송하는 가장 쉬운 방법은 실제 데이터를 생산하는 주체, 또는 1차 수집 주체가 Kafka의 특정 Topic으로 바로 전송하는 것입니다. 이 경우 Kafka만이 관리의 대상이 되므로, 데이터를 수집하는 입장에서는 굉장히 편리한 선택일 수 있습니다.

Kafka direct 사용시 단점

하지만, 실제로 데이터를 수집하고 송신해야 하는 end client에게는 데이터 전송을 위해 Kafka의 client library dependency를 추가해야 하는 등 추가 작업이 발생하게 됩니다. 물론, end client에게는 데이터를 다음 단계로 안전하게 전달할 의무가 있습니다. 그렇다 하더라도 자신에게 익숙한 Protocol이 아닌 다른 수단을 활용해야 하는 것이 부담스러울 수 있습니다.  이러한 부담을 줄이기 위해 데이터를 수신하는 Gateway 역할의 Web application layer를 두고, 이 application이 Kafka로의 데이터 전송을 책임지도록 하는 방법을 적용하게 되었습니다.

단점 해결 방향

Gateway의 역할을 하는 web application은 일반적인 클라이언트들이 사용하는 HTTP protocol을 사용하는 것이 가장 적은 비용이 들 것이라고 판단했습니다. 이 Gateway에게는 이미 언급했듯이 클라이언트로부터 수신한 다양한 데이터를 Kafka로 안전하게 전달할 의무가 있으며, 또한 높은 성능 (low latency, high throughput, reliability, etc.)을 보장해야 할 의무도 있습니다. 이러한 web application이 제공해야 하는 기능에 가장 부합하는 것이 Akka 였고, 더불어 데이터의 흐름을 안정적으로 처리하기 위해 Akka stream을 활용하게 된 것입니다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
import akka.stream._
import akka.stream.scaladsl._
import akka.NotUsed
import akka.actor.ActorSystem
import akka.util.ByteString
import scala.concurrent._
import java.nio.file.Paths
object Example {
  /**
    * Create actor system with name, "QuickStart"
    */
  implicit val system = ActorSystem("QuickStart")
  /**
    * Stream 이 동작하도록 하는 Stream Execution Engine 의 factory 로
    * Materializer 로 생성된 Engine 은 실제 stream 이 수행되도록 해주는 핵심입니다.
    */
  implicit val materializer = ActorMaterializer()
  /**
    * Source 의 첫 번째 type Int 는 Source 가 emit 하게 될 data 의 type 을 의미합니다.
    * 두 번째 type NotUsed 는 Source 가 부가적인 데이터를 의미합니다.
    * 여기에선 1 ~ 100 의 Int 이외에 추가적인 데이터가 없으므로 NotUsed 를 사용하였습니다.
    */
  val source: Source[Int, NotUsed] = Source(1 to 100)
  /**
    * 생성된 Source 아래 runForeach 로 실제 Source 에 대한 검증을 해볼 수 있습니다. 
    * Source 를 검증할 때, 위에서 생성한 amterializer 가 implicit 하게 사용됩니다.
    */
  source.runForeach( i => println(_) )
  /**
    * 1 이라는 숫자로 factorial 계산을 시작해서 source 에서 emit 되는 1 에서 100 의 숫자를
    * scan 하며 factorial 값을 만들어내는 과정을 나타내는 Source 를 만들어 냅니다.
    */
  val factorials: Source[BigInt, NotUsed] = source.scan(BitInt(1))((acc, next) => acc * next)
  // Sink 를 FileIO 로 선택하여 넘어온 값들이 저장될 수 있도록 처리
  val fileSink = FileIO.toPath(Paths.get("factorials.txt")
  val result: Future[IOResult] = 
    factorials
      .map(num => ByteString(s"$num\n")) // factorials(Source) 에서 생성된 int stream 데이터를 ByteString 형태로 변환 
      .runWith(fileSink) // runWith 를 통해서 위에서 정의된 Flow 를 실제 실행해준다. 
}

1
2
3
4
5
def lineSink(fileName: Stirng): Sink[BigInt, Future[IOResult]] =
  Flow[BigInt]
    .map(s => ByteString(s"$s\n"))
    .toMat(FileIO.toPath(Paths.get(fileName)))(Keep.right)
val reulst: Future[IOResult] = factorials.runWith(lineSink("factorials2.txt")

1
2
3
val brokers = "host1:9092,host2:902,host3:9092"
val producerSettings = ProducerSettings(syste, new ByteArraySerializer, new StringSerializer)
  .withBootstrapServers(brokers)

1
2
3
4
Source(1 to 10000)
  .map(_.toString)
  .map(elem => new ProducerRecord[Array[Byte], String]("topic1", elem))
  .to(Producer.plainSink(producerSettings))

1
2
3
4
5
6
7
8
9
Source(1 to 1000)
  .map(elem => ProducerMessage.Message(new ProducerRecord[Array[Byte], String]("topic1", elem.toString), elem))
  .via(Producer.flow(producerSettings))
  .map { result => 
    val record = result.message.record
    println(s"${record.topic}/${record.partition} ${result.offset}: ${record.value} (${result.message.passThrough}")
    result
  }
  .to(Sink.ignore)

message를 produce 하는 두 예제 (<예제 4>와 <예제 5>) 간의 큰 차이점을 발견하기 어려울 수도 있습니다. 하지만, stream을 구성하고 사용하는 방법에서 약간의 차이가 있고, 용도도 다를 수 있습니다. 두 예제 모두 1에서 10000까지의 숫자를 Kafka 내의 topic1로 produce 하고 있습니다. 두 번째 예제에는 akka stream kafka에서 제공하는 flow를 사용한 작업이 더해져 있습니다. `via`를 통해 실제 flow로 전달하고자 하는 message를 보내고, 전송에 성공한 결과물을 map을 통해 후 처리를 하는 것입니다.

Kafka로 메시지를 보내는 것은 위 두 가지 예제를 활용하여 처리할 수 있습니다. 이 예제들은 github에서 제공하는 가이드에도 포함되어 있습니다. Kafka를 사용해보신 분들이라면, 여기까지 구현하면 어느 정도 처리가 된다는 것은 알고 계실 것 같구요. Kafka 또는 네트워크 상에서 문제가 발생했을 때, 전송하던 데이터 손실을 어떻게 처리할 수 있는지가 추가로 고민했던 부분입니다. Akka stream에는 message를 처리하는 도중 이러한 문제가 발생하면 stream에 대한 self recover를 수행하는 기능이 포함되어 있습니다. 이 기능을 사용하면 stream이 중단되는 것과 같은 critical한 이슈는 막을 수 있습니다. 이 때 사용하는 것이 Supervision Strategy입니다. 자세한 내용은 Akka stream 문서의 Error Handling(http://doc.akka.io/docs/akka/2.4/scala/stream/stream-error.html#supervision-strategies) 부분을 참조하시면 됩니다.

간략히 정리하자면, message 처리 과정에서 exception 등 예측하지 못한 error case가 발생한 경우, 상위의 supervisor에게 exception을 보고하고, akka stream이 어떻게 행동할지에 대한 정책을 받아오는 과정이라고 할 수 있습니다. 정책은 Resume, Restart, Stop 세 가지로 구성됩니다. Resume과 Restart는 문제를 일으킨 message를 drop시키고, akka stream이 계속해서 처리를 진행할 수 있도록 stream을 관리합니다. Stop은 말 그대로 akka stream의 처리가 더 이상 진행되지 않도록 stream 자체를 중단시키는 것을 의미합니다. Stream이 중단되면, 흘러 들어오던 메시지들은 모두 dead letter mail box로 보내져 사용자가 정의해 놓은 처리 과정을 거치지 못하게 되죠.

1
2
3
4
5
6
val decider: Supervision.Decider = {
  case _: ArithmeticException => Supervision.Resume
  case _                      => Supervision.Stop
}
implicit val materializer = ActorMaterializer(
  ActorMaterializerSettings(system).withSupervisionStrategy(decider))

1
2
3
4
5
6
7
8
9
10
11
12
13
14
/**
  * Publish records to Kafka topics and then continue the flow. Possibility to pass through a message, which
  * can for example be a [[ConsumerMessage.CommitableOffset]] or [[ConsumerMessage.CommittableOffsetBatch]] that can
  * be committed later in the flow.
  */
def flow[K, V, PassThrough](settings: ProducerSettings[K, V]): Flow[K, V, PassThrough], Result[K, V, PassThrough], NotUsed] = {
  val flow = Flow.fromGraph(new ProducerStage[K, V, PassThrough](
    settings,
    () => settings.createKafkaProducer()
  ))
  .mapAsync(settings.parallelism)(identity)
  if (settings.dispatcher.isEmpty) flow
  else flow.withAttributes(ActorAttributes.dispatcher(settings.dispatcher))
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
setHandler(in, new InHandler {
  override def onPush() - {
    val msg = grab(in)
    val r = Promise[Result[K, V, P]]
    producer.send(msg.record, new Callback {
      override def onCompletion(metadata: RecordMetadata, exception: Exception) = {
        if (exception == null)
          r.success(Result(metadata.offset, msg))
        else 
          r.failure(exception)
        decrementConfirmation.invoke(())
      }
    })
    awaitingConfirmation += 1 
    push(out, r.future)
  }
  override def onUpstreamFinish() = {
    completionState = Some(Success(()))
    checkForCompletion()
  }
  override def onUpstreamFailure(ex: Throwable) = {
    completionState = Some(Failure(ex))
    checkForCompletion()
  }
})

1
case class KafkaFallbackException[K, V](exception: Exception, record: ProducerRecord[K, V]) extends Exception(exception)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
setHandler(in, new InHandler {
  override def onPush() - {
    val msg = grab(in)
    val r = Promise[Result[K, V, P]]
    producer.send(msg.record, new Callback {
      override def onCompletion(metadata: RecordMetadata, exception: Exception) = {
        if (exception == null)
          r.success(Result(metadata.offset, msg))
        else 
          r.failure(new KafkaFallbackException(exception, msg.record))
        decrementConfirmation.invoke(())
      }
    })
    awaitingConfirmation += 1 
    push(out, r.future)
  }
  override def onUpstreamFinish() = {
    completionState = Some(Success(()))
    checkForCompletion()
  }

1
2
3
4
5
6
7
8
9
val decider: Supervision.Decider = {
  case e: KafkaFallbackException[String, String] =>
    // fallback message store to local storage file using slf4j logger
    logger.info(e.record.value)
    Supervision.Resume
  case e: Exception =>
    logger.error(s"Unspected exception occurred : $e", e)
    Supervision.Resume // Just ignore data with unhandled exception, in case of non-pre-defined
}

얻을 수 있는 이득