ZIO - 카프카 컨슈머에서 ZSink.collectAllN 무한 대기

-

 이슈

zio-kafka 의 컨슈머에서 컨슘하는 레코드들을 N 개씩 Chunk 로 묶어 일괄 처리하고 싶었다.
그래서 ZSink.collectAllN 함수를 사용했는데, N 개씩 Chunk 로 잘 묶다가 마지막 레코드들이 N 개를 채우지 못하면 무한히 대기하는 현상이 발생했다.

ex) n = 3, 레코드를 5개 컨슘한다고 했을 때, 3 개는 Chunk 로 잘 묶어 처리하지만 나머지 2 개는 처리하지 않고 대기했다.

ZSink.collectAllN 함수는 ZStream 을 N 개씩 Chunk 로 묶어주는 함수이다. N 개를 채우지 못해도 Chunk로 묶는다. (이 부분 때문에 헷갈렸다.)

코드 예시
ZStream(1, 2, 3, 4, 5).run(
  ZSink.collectAllN(3)
)
// Output: Chunk(1,2,3), Chunk(4,5)

Why?

생각해보니 답은 간단했다.
위 코드 예시에선 ZStream 의 끝을 알 수 있어 (1, 2, 3), (4, 5) 로 나눠주지만, 컨슈머는 끝이 없다. 
무한히 컨슘하기 때문에 ZSink.collectAllN 함수 입장에선 N 개가 확정된 순간에만 Chunk 로 묶을 수 있다.


결국 ZSink.collectAllN 함수말고 다른 함수를 사용했지만, 어떤 방식으로 컨슘을 하길래 collect 하지 못하고 기다리게 되는지가 궁금했다.
그래서 먼저 ZSink.collectAllN 함수를 봤는데
def collectAllN[In](n: => Int)(implicit trace: Trace): ZSink[Any, Nothing, In, In, Chunk[In]] =
    fromZIO(ZIO.succeed(ChunkBuilder.make[In](n)))
      .flatMap(cb => foldUntil[In, ChunkBuilder[In]](cb, n.toLong)(_ += _))
      .map(_.result())
아무리 살펴보아도 이 함수에서 뭔가를 기다리거나 하지는 않는 듯 했다.
단순히 n 개씩 돌며 Chunk 를 만들어준다.

다음으로 Consumer 코드를 찾아봤다. 아래는 실제 consume 이 일어나는 코드이다.
override def partitionedAssignmentStream[R, K, V](
      subscription: Subscription,
      keyDeserializer: Deserializer[R, K],
      valueDeserializer: Deserializer[R, V]
    ): Stream[Throwable, Chunk[(TopicPartition, ZStream[R, Throwable, CommittableRecord[K, V]])]] = {
			/.../
      ZStream.unwrapScoped {
        for {
          stream <- ZStream.fromHubScoped(partitionAssignments) // Hub 에서 stream 을 가져옴.
          _      <- extendSubscriptions.withFinalizer(_ => reduceSubscriptions.orDie)
        } yield stream
          .map(_.exit)
          .flattenExitOption
          .flattenChunks
          .map {
            _.collect {
              case (tp, partition) if Subscription.subscriptionMatches(subscription, tp) =>
                val partitionStream =
                  if (settings.perPartitionChunkPrefetch <= 0) partition
                  else partition.bufferChunks(settings.perPartitionChunkPrefetch)

                val stream: ZStream[R, Throwable, CommittableRecord[K, V]] =
                  if (onlyByteArraySerdes) partitionStream.asInstanceOf[ZStream[R, Throwable, CommittableRecord[K, V]]]
                  else partitionStream.mapChunksZIO(_.mapZIO(_.deserializeWith(keyDeserializer, valueDeserializer)))

                tp -> stream
            }
          }
      }
    }
다른 코드는 볼 필요 없고

stream <- ZStream.fromHubScoped(partitionAssignments)

요 부분이 중요하다. Hub 에서 Stream 을 가져온다.
Hub 는 또 뭐지..? 하고 더 찾아봤다. 아래는 internal Hub 이다.
private[zio] abstract class Hub[A] extends Serializable {
		/.../
		def subscribe(): Hub.Subscription[A]
		/.../
}

private[zio] abstract class Subscription[A] extends Serializable {

    /.../
    def poll(default: A): A
		/.../
}
Hub 는 구독(subscribe) 메서드가 있고 이 메서드는 구독 객체를 반환한다. Hub 를 구독한 객체(Subscription)는 poll 메서드로 구독 값을 땡겨올 수 있다.
(Hub 가 zio-kafka의 본체? 라는 생각이 들었다.)

위 Hub 는 internal 이고, 외부에서 사용되는 Hub 의 subscribe 메서드는 Dequeue 를 반환한다.
abstract class Hub[A] extends Enqueue[A] {
		/.../
		def subscribe(implicit trace: Trace): ZIO[Scope, Nothing, Dequeue[A]]
		/.../
}

외부에서 Hub 를 사용한다면 hub.subscribe() 로 구독하고, 구독 대상의 Dequeue 를 받아 Dequeue 의 메서드로 구독 값들을 관리할 수 있을 거다.

Hub 에서 반환하는 Dequeue 메서드의 구현체에서 take 메서드는 다음과 같다.
def take(implicit trace: Trace): UIO[A] =
        ZIO.fiberIdWith { fiberId =>
          if (shutdownFlag.get) ZIO.interrupt
          else {
            val empty   = null.asInstanceOf[A]
            val message = if (pollers.isEmpty()) subscription.poll(empty) else empty // poll 메서드로 message 를 가져옴.
            message match {
              case null =>
                val promise = Promise.unsafe.make[Nothing, A](fiberId)(Unsafe.unsafe)
                ZIO.suspendSucceed {
                  pollers.offer(promise)
                  subscribers.add(subscription -> pollers)
                  strategy.unsafeCompletePollers(hub, subscribers, subscription, pollers)
                  if (shutdownFlag.get) ZIO.interrupt else promise.await // 값이 없다면 wait...
                }.onInterrupt(ZIO.succeed(unsafeRemove(pollers, promise)))
              case a =>
                strategy.unsafeOnHubEmptySpace(hub, subscribers)
                ZIO.succeed(a)
            }
          }
        }
dequeue.take() 로 외부에서 구독값을 가져올터라 Dequeue 구현체에서 위 메서드만 살펴보았다.
ZSink.collectAllN 함수에서 계속 대기하는 원인의 코드가 위에 있었다.
주석으로 달아 놓았듯이, 구독 대상을 poll() 했을 때 empty 라면 promise.await 으로 구독 값(A) 를 가져올 때까지 기다린다.
구독 값(A)를 가져오는 Promise 를 pollers.offer(promise) 코드에서 queue 에 집어넣는데, 이를 unsafeCompletePollers 함수에서 처리한다.

unsafeCompletePollers 함수는 다음과 같다.
final def unsafeCompletePollers(
      hub: internal.Hub[A],
      subscribers: Set[(internal.Hub.Subscription[A], MutableConcurrentQueue[Promise[Nothing, A]])],
      subscription: internal.Hub.Subscription[A],
      pollers: MutableConcurrentQueue[Promise[Nothing, A]]
    ): Unit = {
      var keepPolling = true
      val nullPoller  = null.asInstanceOf[Promise[Nothing, A]]
      val empty       = null.asInstanceOf[A]

      while (keepPolling && !subscription.isEmpty()) {
        val poller = pollers.poll(nullPoller)  // 1. Promise poll()
        if (poller eq nullPoller) {
          subscribers.remove(subscription -> pollers)
          if (pollers.isEmpty()) keepPolling = false
          else subscribers.add(subscription -> pollers)
        } else {
          subscription.poll(empty) match {  // 2. Subscription.poll()
            case null =>
              unsafeOfferAll(pollers, poller +: unsafePollAll(pollers))  // 3. 다시 pollers 에 offer
            case a =>
              unsafeCompletePromise(poller, a)
              unsafeOnHubEmptySpace(hub, subscribers)
          }
        }
      }
    }
간단히 요약하면
1. 구독 값(A)을 가져오는 Promise 를 poll 한다.
2. 구독 값(A)을 구독 객체(Subscription)에서 poll 해본다.
3. null, 즉 비어있다면 다시 Promise 를 pollers 에 집어넣는다. (1 번부터 반복)

위 동작으로 구독 값(A)가 구독 객체(Subscription)에 쌓일 때까지 계속 poll 을 하며 기다리게 된다.

ZStream, ZSink 에서 출발한 의문이었지만 카프카 컨슈밍의 동작 원리에 대해서 좀 더 이해를 할 수 있게 된 것 같다!

댓글

댓글 쓰기

이 블로그의 인기 게시물

HTML - input file 버튼 꾸미기

-
이미지
 이번 글에선 HTML 에서 기본적으로 제공하는 input 태그의 file 속성일 때의 모양을 바꿔보려고 한다. 열심히 구글링을 해본 결과 보안상의 문제로 직접 input 태그에 변경을 가할 수는 없다고 한다. 따라서 input 태그에 속성을 넣는 게 아닌 약간의 편법?을 이용해서 업로드 버튼처럼 보이게 만들어야 한다. 먼저, 파일의 이름을 출력할 가짜 input 태그를 넣자. <input type= "text" id= "file_route" disabled= "disabled" value= "파일 선택" > value 에 안내 문구를 넣었고, disabled를 지정했기 때문에 커서를 옮겨도 아무 것도 할 수 없다. 잘 생성됐다. 이제 오른쪽 버튼과 "선택된 파일 없음" 문구를 없애고 새 버튼을 만들어야 하는데 label과 style 속성을 잘 활용해서 없애보자. <input type= "text" id= "file_route" disabled= "disabled" value= "파일 선택" > <label for= "upload_file" style= "border: solid 1px black;" > 업로드 </label> <input type= "file" id= "upload_file" accept= "image/*" required= true style= "position:absolute; clip:rect(0, 0, 0, 0);" > label 태그는 for 속성으로 연결한 id의 태그와 자신(라벨)을 연결해 준다. 즉, "업로드" 를 클릭해도 input 태그를 클릭한 것...
자세한 내용 보기

HTML - 이미지 미리보기(jQuery 없이)

-
이미지
 최근에 웹 클론 코딩을 하면서 이미지 미리보기 기능을 구현해야 됐는데 구글링해서 나온 대부분이 제이쿼리를  이용한 코드여서 찾는데 애를 좀 먹었다. 이 글에선 제이쿼리를 쓰지 않고 이미지 미리보기를 구현해보려고 한다. 먼저, 기본으로 제공되는 input 태그와 미리보기를 하기 전 기본 이미지를 띄워보자. <img src= "고양이.jpg" id= "img_section" style= "width: 400px; height: 400px;" > <br> <input type= "file" id= "upload_file" accept= "image/*" required= true value= "업로드" > src 는 본인이 원하는 기본 이미지 경로로 설정해주면 된다. (어차피 나중에 src 속성을 바꿔서 이미지가 변경됨!) id , style 등등 나머지 속성을 지정해주자. 아직은 이미지와 input 버튼만 만들었기 때문에 파일을 아무리 업로드해도 아무 변화가 없다. 이제 동적으로 img 태그의 src 를 바꿔주는 스크립트를 작성해보자. <img src= "고양이.jpg" id= "img_section" style= "width: 400px; height: 400px;" > <br> <input type= "file" id= "upload_file" accept= "image/*" required= true value= "업로드" > <script> const reader = new FileReader(); reader.o...
자세한 내용 보기

BOJ - DNA 유사도(2612)

-
문제(링크) LCS 알고리즘 응용문제입니다. dp[i][j] = i, j 지점에서 나올 수 있는 최대 유사도 공백을 추가할 때, 현재 유사도가 음수가 되버리면 그 지점부터 다시 시작하는게 이득입니다. 따라서 lcs 알고리즘과 비슷하게 문자가 같지 않다면 왼쪽, 왼쪽 위, 위 값의 최댓값을 현재 값으로 가져오는데 해당 값에 공백을 추가한 값, 즉, -2 를 한 값을 가져오고 최종 값이 음수가 되면 0으로 바꿔줘야 합니다. 문자가 같다면 lcs 알고리즘에서 처럼 +3을 해줍니다. dp 배열을 완성했다면 유사도가 최대인 점을 찾습니다. 최대인 점으로부터 가로, 세로, 대각선을 체크하면서 시작 점을 찾아가는데, 가로, 세로는 +2 , 대각선은 -3 으로 찾아가면서 0이 될때까지 찾아가고, 역으로 시작점부터 끝점까지 출력하면 됩니다. #include<iostream> #include<algorithm> #include<math.h> #include<string> #include<vector> #include<stack> #include<queue> #include<map> using namespace std; #define FIO ios_base::sync_with_stdio(false); cin.tie(NULL) #define pii pair<int, int> #define pdd pair<double, double> #define pic pair<int, char> #define ll long long #define vi vector<int> #define vl vector<long long> #define vc vector<char> #define vii vector<pii> #define IMAX 2000000001 #define LM...
자세한 내용 보기