Kafka Consumer Design

Background

대부분이 메시징 시스템이 그렇듯이, 프로듀서는 비교적 쉽습니다. 그냥 메시지를 발행하면 됩니다. 

하지만 컨슈머의 경우 비즈니스의 요구사항에 따라 고려할 점이 많아집니다.

Fail over를 처리해야되는지 또는 메시지를 중복 소모를 해도 되는지 또는 병렬처리가 필요한지 또는 비동기처리를 해야되는지 등.. 

제가 이번에 경험한 상황은 기존에 RabbitMQ로 50개의 병렬 메시지를 동시에 처리하던 작업을 Kafka 로 마이그레이션 하는 작업입니다. 

각 메시지는 처리시간이 초 단위에서 길게는 30분 까지 다양합니다. 

그리고 메시지는 최대 50개를 병렬 처리해야하며, 각각의 메시지는 끝나면 큐에서 다음 메시지를 가져와서 처리해야합니다. 

Smart pipe and Dump pipe

RabbitMQ와 Kafka는 매우 특징이 다릅니다.

RabbitMQ 는 흔히 말하는 스마트 파이프 중 하나이고 Kafka는 덤브 파이프 - 스마트 컨슈머 모델입니다. 

RabbitMQ에서는 메시지 오프셋을 관리해주고 잘못 처리된 메시지에 대해 Re-queue 기능도 제공합니다. 

따라서 RabbitMQ는 스마트합니다. 그만큼 컨슈머는 쉽게 구현 가능하며 코드또한 깔끔합니다.

하지만 카프카는 덤브파이프 - 스마트컨슈머 모델로, 카프카 자체에서 제공해주는 것은 그저 메시지를 보관하고 전달하는 기능입니다.

그만큼 간단하게 구현되어서 확장성과 속도면에서 우월하지만, 스마트 파이프에서 제공해주던 오프셋관리 라던가 Fail over 등 여러가지 기능들을 컨슈머를 구현할때 고려해야 합니다. 

그래서 요약하자면, 이번 포스트에서는 하나의 메시지 처리가 초~30분정도 걸리고, 50개의 병렬처리가 필요한 요구사항을 카프카에서는 어떻게 처리할 수 있을지 알아보고자 합니다.

 

KAFKA 컨슈머 구현 방법들

카프카에 대한 기본적인 정보는 이곳을 참조하세요.

카프카는 같은 컨슈머 그룹이라면, 파티션 1개당 컨슈머 1개만 연결될 수 있습니다. 

따라서 기본적으로 파티션 개수가 병렬처리의 기준이 됩니다. 

파티션이 1개라면, 하나의 컨슈머 그룹으로는 워커-스레드 모델이 아니라면 동시에 여러 메시지를 처리할 수 없습니다. 

멀티컨슈머 모델을 이용해서 병렬처리를 하고 싶다면, 파티션 개수를 늘려서 컨슈머당 파티션에 대한 커넥션을 따로 가져가야 합니다. 

카프카는 배치 컨슈밍이 가능하지만, 이것도 마찬가지로 파티션이 1개라면 워커-스레드 모델로 배치로 가져온 메시지를 병렬처리 하지 않는 한, 병렬처리 할 수 없습니다. 

파티션을 늘리는 방법이 컨슈머 입장에서는 구현이 쉽습니다. 

워커-스레드 모델을 구현하지 않아도 되고 단순히 파티션과 컨슈머만 늘려주면 됩니다. 

때문에 비동기처리도 고려할 필요가 없습니다. 

하나의 컨슈머는 하나의 메시지를 각 파티션에서 가져와서 처리 후 커밋합니다. 

오프셋 관리도 쉽습니다. 자동으로 증가시키면 됩니다. 

물론 Failover 관련해서는 추가로직이 필요해보이지만, 워커스레드에 비하면 매우 간단합니다. 

그래서 파티션수를 늘리면 좋겠지만...병렬처리만을 위해 카프카의 파티션을 늘리는 것에는 한계가 있습니다. 

오버헤드가 커진다고 판단해서 카프카담당 팀에서 파티션 수에 제한을 둡니다. 

그리고 파티션은 증가만 가능하고 수축이 불가능합니다. 따라서 병렬처리 개수 N 개가 확장성이 좋지 못합니다. 

지금까지 정리하자면

-  50개의 메시지를 병렬처리해야 되는 상황으로 가정합니다. 각 메시지는 1초~30분 처리시간을 갖습니다.

- 병렬처리를 위한 방법으로는 크게 두 가지 방법이 있습니다. 

    1) 워커스레드를 늘리는 방법 - 아래 모델에선 Single Consumer(워커-스레드) 모델에 해당

    2) 파티션&컨슈머 수를 늘리는 방법  - 아래 모델에선 Multi Consumer 모델에 해당

 - 멀티 컨슈머 모델은 컨슈머 스레드를 이용해서 비즈니스 로직을 처리하므로 아무래도 별도의 스레드를 생성하는 워커-스레드 모델보다 구현이 쉽습니다. 

 - 스케일-아웃이 자주 발생하면, 워커스레드 모델을 사용할 것을 권장하고, 그렇지 않고 파티션 수에 확장이 없다면 컨슈머 스레드를 사용하는 것도 괜찮습니다.

카프카 컨슈머 디자인

위의 요구사항들을 해결할 컨슈머 구현 방안들 입니다. 위에서 설명한 멀티컨슈머, 워커스레드 모델도 포함됩니다. 

그리고 하나 익숙하지 않은 디자인이 있는데, 이것은 컨슈머그룹을 이용한 약간의 트릭이라고 할 수 있습니다. 

파티션 수가 제한된 상황에서 파티션을 아끼기 위해 컨슈머그룹이 다르면 같은 파티션에 여러 컨슈머가 붙을 수 있다는 점을 이용합니다. 

컨슈머 그룹 1,2로 나누면 25개의 파티션만 이용해도 그룹1,2가 상태를 공유해가며 50개의 병렬처리가 가능합니다. 

자세한 설명은 아래 디자인에서 설명합니다.

1. Worker thread model - MAX_POLL_RECORDS(1)

메시지를 처리하는 작업은 별도의 워커스레드가 담당합니다. 

카프카 컨슈머스레드는 메시지를 읽기만 합니다. 

컨슈머 스레드는 메시지를 1개씩 가져와서 워커에게 할당 후 커밋합니다. 

컨슈머는 가용할 수 있는 워커가 없을 경우 poll을 대기합니다.

장점 

- partition개수에 상관없이 consumer의 scale-out이 가능합니다.

- 메시지를 1개씩 가지고 오므로 배치 컨슈밍과 비교해서 메시지 커밋이 자유롭습니다. 

( * 배치 컨슈밍하면 같이 컨슈밍된 메시지들은 하나의 커밋에 묶입니다. 따라서 모두 다 처리되야 커밋할 수 있기 때문에 지금과 같이 작업시간이 1초~30분 까지 차이나는 상황에서는 좋지 않습니다.)

단점

- 워커-스레드 모델을 구현해야 합니다.

- 1개씩 메시지를 가지고 오므로, 여러개씩 가져와서 처리할 때보다 메시지 오버헤드가 큽니다.

2.Worker thread model - MAX_POLL_RECORDS(N)

MAX N개를 한번에 poll합니다. 쌓여있는 메시지가 없으면 1개가 Poll될 수도 있습니다.

N개는 consumer.commit 을 할 경우 모두 같이 커밋됩니다.

N개가 커밋으로 묶이기 때문에 N개가 모두 끝나야 다음 poll()이 가능합니다. 작업이 1초~30분까지 다양하기 때문에 치명적입니다. 

마찬가지로 사용가능한 워커스레드가 있을때 폴을 합니다. 

1~N개를 가져오기때문에 병렬처리를 정확하게 50개를 맞추려면 추가적인 구현이 필요합니다.

그림은 1번과 비슷하므로 생략합니다. 단 N개씩 가져와서 처리하며 N개를 적절히 워커에게 분배하는 추가적인 로직이 구현되어야 합니다. 

3.Multi consumer model 

파티션 개수를 병렬처리하고 싶은 개수만큼 늘립니다.

각 파티션에는 1개의 컨슈머가 할당됩니다.

장점

- 가장 간단히 구현될 수 있습니다. 거의 설정만 하는 수준입니다.

단점

- 파티션 개수가 늘어납니다.

- consumer의 scale-out 할때마다 파티션개수에 제한이 걸립니다.

4.Multi consumer model with consumer group

컨슈머 그룹을 이용한 방법입니다.

그룹 수로 나눈 만큼 파티션을 절약하면서 병렬처리 효과를 볼 수 있습니다.

해피케이스를 예로들면 다음과 같습니다. 이때 필요한 파티션 수는 50개에서 그룹 수 2로 나눈 25개입니다.

1. 그룹1의 컨슈머들은 메시지 1~25를 가져온다.
2. 그룹1의 컨슈머들은 워커에게 작업을 할당한다.
3. 그룹1의 컨슈머들은 작업상태를 DB에 업데이트한다.
4. 그룹2의 컨슈머들은 메시지 1~25를 가져온다.
5. 그룹2의 컨슈머들은 DB 상태를 확인하여 메시지 1~25가 그룹1에 의해 처리되었음을 확인하고 워커에게 작업을 할당하지 않고 커밋처리한다.
6. 그룹2의 컨슈머들은 다음 poll을 통해 26~50 메시지를 가져와서 처리한다.

장점

- 워커-스레드 모델보다는 구현이 쉽습니다.

- 파티션수를 그나마 절약할 수 있습니다.

단점

- 작업의 상태를 저장할 곳(DB)을 계속 체크해야 합니다.

- 뭔가 일반적인 사용방법을 벗어나는 방법인 것 같습니다.

회고

RabbitMQ 가 자주 뻗어서 확장성이 좋은 카프카로 마이그레이션을 시도해봤습니다. 

하지만 어떤 기술이든 모든 요구사항을 만족시킬 수는 없는 것 같습니다. 

카프카의 경우 덤브 파이프 모델로 확장성은 좋습니다. 

하지만 이번에 적용해야되는 비즈니스 로직이 카프카로 구현하기에는 아래와 같은 문제점들이 있었습니다. 

1. 워커스레드 모델 구현

2. 중복처리 로직 구현

3. 1초~30분 다양한 처리시간으로 인해 오프셋 관리의 복잡함

4. Failover Re-queue 로직 

위와 같이 다양한 이유로 인해 카프카 마이그레이션은 현재 비즈니스 특성에 맞지 않다고 판단되어 중단되었습니다. 

그리고 다음에 이어질 SQS 를 이용해서 구현되었습니다.