Uber에서 Kafka를 MSA 큐로 사용하기 위해 고려한 것들

Kafka를 ETL이 아니라 서비스에 적용했다는 사례가 종종 들린다.
마지막 commit 이후 처리가 완료된 message가 연속적으로 있을 때 offset commit 돼야 하고
exactly-once, 그리고 consumer group 중 한 개 consumer만 재시작돼도 rebalance가 일어나면서
처리가 멈춰서 개인적으로 서비스에 Kafka를 도입에 부정적이다. 

차라리 RabbitMQ처럼 온전히 Queue 목적으로 만들어진 제품을 쓰는 게 여러모로 좋다고 생각했다.

 

이런 나를 비웃듯이 우버에서 Kafka를 서비스에 어떻게 적용하는지 소개했다.  크흐~~~

https://eng.uber.com/kafka-async-queuing-with-consumer-proxy/

결론부터 말하면 Kafka Consumer client 쓰지 않고, 메시지를 분배해주고 처리 완료되는걸 offset commit 해주는
Consumer Proxy를 만들어 문제를 해결했다.

어떠한 문제가 있었고 자체 개발로 어떤 이득을 얻었는지 궁금하니 알아보기로 하자

---

하루에 수 페타바이트 크기의 1 조개 메시지를 처리하는 Uber의 Kafka

Uber 기술 스택에 없어서는 안 될 존재가 되었다. `cornerstone of our technology stack`

다양하게 사용되지만 이번에는 MSA에 pub-sub으로 쓰이면서 마주한 문제와 해결한 방법을 공유한다.

 

이번엔 MSA에서 pub-sub 메시지 큐로 사용하는 것에 대해 알아보기로 한다.

Consumer 입장에서 순서를 보장해서 처리하기 위해선 다음을 만족해야 한다.
※ 예를 들어 1, 2, 3, 4 message가 있는데 1 message 처리가 완료됐을 때 offset commit.
만약 아직 1, 2 message가 처리 중인데 3 message 처리가 완료됐다고 offset commit 하게 되면
1, 2번이 실패하는 경우 메시지 유실이 발생한다. 그래서 파티션 한 개 유지

1. 같은 토픽에 같은 consumer group에서 consumer 한 개만 존재
2. Kafka consumer
   partition에서 한 개의 메시지를 읽고 처리해야 하고, 처리가 끝나기 전에 다음 메시지를 읽으면 안 된다.

 

문제 시작!!

5년간 kafka as pub-sub을 사용했는데 사용량이 12배 증가하면서 두 가지 문제를 만났다.

파티션 확장과 HOL blocking (https://en.wikipedia.org/wiki/Head-of-line_blocking)

※ HOL이란? 예로 들면 고속도로 8차선에서 현금 결제하는 차선이 한 개 일 때 현금 준비 못해서 오래 걸리는 차량 하나만 있으면
뒤로 계속 밀린다.  (물론 하이패스 지나가고 나중에 결제해도 되는데 컴퓨터 세상엔 얄짤없음)

같은 큐에 있는 패킷 중 첫 번째 처리가 지연되면서 발생하는 성능 저하를 의미

 

1. 파티션 확장

약 1초 정도 걸리는 결제 서비스가 있다고 하면 Kafka consumer 패턴으로는 1000 events/s를 처리하기 위해선
1000개 파티션이 필요하다.
그리고 네트워크 대역폭이 10MB/s이고 1KB 메시지라고 하면 이론상 최대 10K messages/s가 가능한데
순서를 보장하기 위해 파티션 한 개에서 처리해야 하므로 1 message/s가 된다.

2. HOL

한 개씩 순서대로 처리해야 하므로 처리속도와는 상관없이 특정 처리가 밀리면 이후의 모든 처리가 같이 밀린다.

 

Kafka consumer의 기능 중 다음 두 가지 문제가 있다고 보면 된다.

1. autocommit - 처리가 완료되지 않은 메시지의 offset이 커밋이 될 수 있고 이는 메시지 유실 가능성이 있다.

2. 메시지 처리량 저하 - 내부에서만 사용하는 서비스라면 1초 기다릴 수 있다. 하지만 결제 서비스에서 외부 서비스
latency가 100ms인 경우 1초에 10개밖에 처리하지 못한다.

 

그래서 Uber가 만든 Consumer Proxy!!

1. Kafka binary protocol을 이용하여 Kafka로부터 메시지를 가져온다.

2. message를 처리하는 consumer service에서 gRPC 통신으로 한 개씩 message 전송

3. consumer service는 처리 완료 후 Consumer Proxy에 응답 값을 보낸다.

4. Consumer Proxy는 gRPC status code를 받고

5. 이 값을 취합한 다음

6. 확실히 처리됐다고 판단되는 offset을 커밋한다.

 

Consumer client를 쓰지 않고 Proxy를 한 이유는 다음과 같다.

Uber에서는 Go, Java, Python, NodeJS 등 다양한 언어를 사용하는데 언어별 Kafka-client를 만들고 유지 보수하는 건 꽤 힘든 일이다. 하지만 Consumer Proxy를 이용하면 하나의 언어로 모든 서비스에 적용이 가능하다.

 

Uber에서는 1000개 이상의 마이크로 서비스가 있는데 kafka client 버전업 한다는 것도 곤욕이다.(한 달 이상 걸린다고 한다 덜덜).
하지만 Consumer Proxy를 이용하면 protocol이 변경되지 않는 한 서비스와 무관하게
Kafka team 자체적으로(Kafka + Consumer Proxy만 하면 되니) 가능하다.

 

Consumer group rebalance storm, rolling restart가 시작되면 해당 서비스의 consume은 멈춘다.
한 서비스가 백개 인스턴스가 있다고 하면 환장하겠죠? Consumer proxy는 인스턴스 개수도 훨씬 적게 유지하고
자체 rebalance를 구현해서 rebalance 영향을 제거했다.

 

※ 4번째는 이해 안 됨;;

 

파티션 별 병렬 처리

Kafka 파티션 1개는 Consumer Proxy Node 1개에서 메시지 가져오고
메시지 처리는 consumer service에 전달하여 병렬 처리 한다.

 

Out-of-Order commit

메시지 마다 처리 속도가 다를 수 있다. (Visa, Master 결제 차이)

이 문제를 해결하기 위해 out-of-order commit 개념을 도입했다.

Consumer Proxy는 Kafka에서 여러 message를 가져오고 consumer service에서 처리가 되면 acknowledge 상태로 변경!!
마지막 commit 이후 연속적으로 acknowledge인 메시지가 확인되면 그때 offset commit 해준다.

 

Consumer Proxy가 Kafka로부터 여러 message를 가져오면 out-of-order commit tracker에 넣어준다.

 

. Consumer rebalance 할 때 acknowledge인 message가 다시 처리될 수 있다.

. Poison pill message가 있다면 offset commit은 되지 않고 acknowledge인 상태로 계속 남아있는다.

※ Poson pill message란 처리하는데 굉장히 오랜 시간 걸리는 message

Uber에서는 consumer service에 중복 제거 처리를 항상 하기 때문에 첫 번째는 문제가 아닌데
Poison pill message는 문제가 된다. 이를 위해 DLQ를 준비했다.

 

DLQ

DLQ용 토픽을 만들고 consumer service에서 Poison pill message라고 판단되면 Consumer Proxy에게 gRPC 에러코드를
보내주고 Consumer Proxy는 해당 메시지를 DLQ 토픽으로 보내고 "negative acknowledged" 상태로 변경한다.
그리고 연속적으로 acknowledged, negative acknowledged 인지 판단 후 offset commit을 진행한다.

 

Flow Control

Consumer Proxy에서 consumer service로 메시지를 푸시하는 방식이다.
(Consumer Proxy가 consumer service 정보를 알아야겠네?)

consumer service로 너무 적거나 많은 메시지가 가지 않도록 처리 가능한 속도를 유지하기 위한 flow control이 굉장히 중요하다.

다음 3가지 방식으로 flow control 한다.

 

1. Out-of-order commit tracker의 크기와 message 처리의 timeout을 설정할 수 있다.

※ Out-of-order commit tracker 크기를 크게 하면 offset commit이 되지 않은 message가 있어도
Kafka에서 메시지를 더 가져올 수 있겠구나…

2. Consumer service에서 Consumer Proxy로 메시지 속도를 조절할 수 있는 gRPC 코드를 제공한다.

3. Poison pill message가 아니고 consumer service가 모두 내려간 건데 DLQ로 가는 걸 막기 위해 Consumer Proxy는 circuit braker를 사용한다.

 

우리가 문제 해결을 위해 리서치할 때쯤 Kafka community에서 나온
솔루션(Kafka connect, Kafka Rest Proxy)이 있었지만 서비스에 적용하기에는 부족한 부분이 있어 자체 개발했다.

 

끝!!