Kafka

MirrorMaker의 내부 동작과 무손실 보장

MirrorMaker는 핵심만 보면 컨슈머의 집합이다. MirrorMaker 문서에서는 역사적 이유로 이 컨슈머들을 stream이라 부른다. 이 컨슈머들은 모두 같은 컨슈머 그룹에 속하며, 복제하기로 선택한 토픽 집합에서 데이터를 읽는다. 그리고 MirrorMaker 프로세스마다 프로듀서는 단 하나다.

워크플로는 단순하다. MirrorMaker는 컨슈머마다 스레드 하나를 돌린다. 각 컨슈머는 소스 클러스터에서 자신에게 할당된 토픽·파티션의 이벤트를 소비하고, 공유 프로듀서를 사용해 그 이벤트를 대상 클러스터로 보낸다. 기본값으로 60초마다 컨슈머들은 프로듀서에게 가진 이벤트를 전부 Kafka로 보내라고 지시하고, Kafka가 그 이벤트들을 확인(acknowledge)할 때까지 기다린다. 그런 다음 컨슈머는 소스 Kafka 클러스터에 접속해 그 이벤트들의 오프셋을 커밋한다.

이 순서가 두 가지를 보장한다. 첫째, 데이터 유실이 없다. 오프셋이 소스에 커밋되기 전에 메시지가 대상 Kafka에서 확인되기 때문이다. 둘째, MirrorMaker 프로세스가 죽더라도 중복은 60초치를 넘지 않는다.

MirrorMaker는 단순해 보이지만, 매우 효율적이면서 exactly-once에 최대한 가깝게 만들려다 보니 제대로 구현하기가 까다로웠다. Apache Kafka 0.10.0.0 릴리스 시점까지 MirrorMaker는 네 번 재작성되었다. 이 장의 설명은 0.9.0.0부터 0.10.2.0까지의 MirrorMaker에 해당한다.

명령행 인자로 보는 MirrorMaker 설정

MirrorMaker는 프로듀서 하나와 컨슈머 여러 개를 쓰므로, 프로듀서·컨슈머의 모든 설정 프로퍼티를 그대로 쓸 수 있다. 여기에 MirrorMaker 자체의 설정 옵션도 상당히 많고 서로 복잡한 의존 관계를 갖는다.

bin/kafka-mirror-maker --consumer.config etc/kafka/consumer.properties \
  --producer.config etc/kafka/producer.properties \
  --new.consumer --num.streams=2 --whitelist ".*"

consumer.config는 소스 클러스터에서 데이터를 가져올 모든 컨슈머의 설정이다. 모두 하나의 설정 파일을 공유하므로 소스 클러스터도 하나, group.id도 하나뿐이다. 즉 모든 컨슈머가 같은 컨슈머 그룹에 속하는데, 이것이 바로 우리가 원하는 바다. 필수 설정은 (소스 클러스터의) bootstrap.servers와 group.id다.

건드리면 안 되는 설정이 하나 있다. auto.commit.enable=false다. MirrorMaker는 이벤트가 대상 Kafka 클러스터에 안전하게 도착한 뒤 자기 오프셋을 직접 커밋할 수 있어야 동작한다. 이 설정을 바꾸면 데이터 유실이 생길 수 있다.

반대로 바꾸고 싶어질 설정도 하나 있다. auto.offset.reset이다. 기본값이 latest이므로 MirrorMaker는 자신이 시작된 이후에 소스 클러스터에 도착한 이벤트만 미러링한다. 기존 데이터도 미러링하고 싶다면 earliest로 바꾼다.

producer.config는 MirrorMaker가 대상 클러스터에 쓸 때 사용할 프로듀서의 설정이다. 필수 설정은 (대상 클러스터의) bootstrap.servers 하나뿐이다.

new.consumer는 0.8 컨슈머 대신 0.9 새 컨슈머를 쓰겠다는 표시다. 책은 이 시점에서 더 안정적이라는 이유로 0.9 컨슈머를 권한다.

num.streams는 소스 클러스터에서 읽는 컨슈머의 수다. 같은 MirrorMaker 프로세스의 컨슈머들은 프로듀서를 공유한다는 점을 기억하자. 프로듀서 하나를 포화시키려면 스트림이 여러 개 필요하다. 그 지점을 넘어 더 많은 처리량이 필요하면 MirrorMaker 프로세스를 추가해야 한다.

whitelist는 미러링할 토픽 이름의 정규식이다. 정규식에 매치되는 모든 토픽이 미러링된다. 예시에서는 모든 토픽을 복제하도록 했지만, 실무에서는 prod.* 같은 패턴으로 테스트 토픽을 제외하는 것이 좋다. Active-Active 구성에서는 NYC에서 SF로 복제하는 MirrorMaker에 whitelist="NYC.\*"를 설정해, SF에서 출발한 토픽이 되돌아 복제되지 않게 한다.

프로덕션 배포 — 어느 데이터센터에서 돌릴 것인가

프로덕션에서는 MirrorMaker를 서비스로, nohup으로 백그라운드에서 돌리고 콘솔 출력을 로그 파일로 리다이렉트하는 형태로 운영한다. 기술적으로 -daemon 옵션이 이 모든 것을 해 주기로 되어 있지만, 최근 릴리스에서는 기대대로 동작하지 않았다. 대부분의 회사는 자체 기동 스크립트에 설정 파라미터를 넣어 쓰고, Ansible·Puppet·Chef·Salt 같은 배포 시스템으로 자동화한다.

요즘 인기 있는 방식은 MirrorMaker를 Docker 컨테이너에서 돌리는 것이다. MirrorMaker는 완전히 stateless이고 디스크 저장소가 전혀 필요 없다(모든 데이터와 상태는 Kafka 자체에 저장된다). Docker로 감싸면 한 머신에서 여러 인스턴스를 돌릴 수 있다. MirrorMaker 인스턴스 하나는 프로듀서 하나의 처리량에 묶여 있으므로, 여러 인스턴스를 띄우는 것이 중요할 때가 많다. 또한 피크에 컨테이너를 늘리고 한가할 때 줄이는 식의 스케일 업/다운도 쉬워진다.

가장 중요한 배포 판단은 위치다. 가능하다면 MirrorMaker를 목적지 데이터센터에서 돌린다. NYC에서 SF로 데이터를 보낸다면, MirrorMaker는 SF에서 돌면서 미국을 가로질러 NYC의 데이터를 소비해야 한다. 장거리 네트워크는 데이터센터 내부보다 덜 안정적이기 때문이다. 네트워크 파티션이 생겨 데이터센터 간 연결이 끊겼을 때, 클러스터에 접속하지 못하는 쪽이 컨슈머인 편이 프로듀서인 편보다 훨씬 안전하다. 컨슈머가 접속하지 못하면 이벤트를 읽지 못할 뿐이고 이벤트는 소스 Kafka 클러스터에 오래도록 남는다. 유실 위험이 없다. 반대로 이벤트가 이미 소비되었는데 네트워크 파티션 때문에 MirrorMaker가 생산하지 못한다면, 그 이벤트들이 MirrorMaker에서 유실될 위험이 늘 존재한다. 요컨대 원격 소비가 원격 생산보다 안전하다.

그럼 로컬에서 소비하고 원격으로 생산해야 하는 경우는 언제인가? 데이터센터 사이를 지날 때는 암호화가 필요하지만 데이터센터 내부에서는 필요 없을 때다. 컨슈머는 SSL 암호화로 Kafka에 접속할 때 성능 손해가 크고, 프로듀서보다 훨씬 크며, 그 손해가 Kafka 브로커 자체에도 영향을 준다. 크로스 데이터센터 트래픽에 암호화가 필요하다면 MirrorMaker를 소스 데이터센터에 두고 로컬에서 암호화 없이 소비한 뒤, SSL 연결로 원격 데이터센터에 생산하는 편이 낫다. 이 구성을 쓸 때는 MirrorMaker가 이벤트를 절대 잃지 않도록 acks=all과 충분한 retries를 설정하고, 이벤트 전송에 실패하면 계속 진행하다 데이터를 잃는 대신 종료(exit)하도록 설정해야 한다.

소스와 대상 사이의 랙을 아주 낮게 유지하는 것이 중요하다면, 서로 다른 서버에서 MirrorMaker 인스턴스를 최소 두 개 돌리고 둘 다 같은 컨슈머 그룹을 쓰게 한다. 한 서버가 어떤 이유로든 멈춰도 다른 인스턴스가 미러링을 계속한다.

모니터링 — 랙 측정의 두 방법은 모두 부정확하다

랙 모니터링. 대상 클러스터가 소스보다 얼마나 뒤처지는지 반드시 알아야 한다. 랙은 소스 Kafka의 최신 메시지 오프셋과 대상의 최신 메시지 오프셋의 차이다. 책의 그림에서 소스의 마지막 오프셋이 7이고 대상이 5이면 랙은 2다.

랙을 추적하는 방법은 두 가지인데, 어느 쪽도 완전하지 않다.

첫째, MirrorMaker가 소스 Kafka 클러스터에 커밋한 최신 오프셋을 확인하는 방법이다. kafka-consumer-groups 도구로 MirrorMaker가 읽고 있는 파티션마다 파티션의 마지막 이벤트 오프셋, MirrorMaker가 커밋한 마지막 오프셋, 그 차이를 볼 수 있다. 이 지표는 100% 정확하지 않다. MirrorMaker가 오프셋을 항상 커밋하지는 않기 때문이다. 기본적으로 1분마다 커밋하므로 1분 동안 랙이 커지다가 갑자기 떨어지는 모양이 된다. 위 예시에서 실제 랙은 2지만, MirrorMaker가 더 최근 메시지의 오프셋을 아직 커밋하지 않았기 때문에 kafka-consumer-groups 도구는 랙을 4로 보고한다. LinkedIn의 Burrow는 같은 정보를 보되 이 랙이 진짜 문제인지 판단하는 더 정교한 방법을 갖고 있어 거짓 경보가 덜 생긴다.

둘째, MirrorMaker가 읽은 최신 오프셋을(커밋 여부와 무관하게) 확인하는 방법이다. MirrorMaker에 내장된 컨슈머는 핵심 지표를 JMX로 게시하는데, 그중 하나가 컨슈머 최대 랙(소비 중인 모든 파티션에 대한)이다. 이 랙도 100% 정확하지 않다. 컨슈머가 읽은 것을 기준으로 갱신될 뿐, 프로듀서가 그 메시지들을 대상 Kafka로 보냈는지, 성공적으로 확인받았는지는 반영하지 않기 때문이다. 위 예시에서 MirrorMaker 컨슈머는 랙을 2가 아닌 1로 보고한다. 메시지 6을 이미 읽었기 때문이다 — 그 메시지가 대상에 아직 생산되지 않았더라도 말이다.

주의할 점은 MirrorMaker가 메시지를 건너뛰거나 버리면 두 방법 모두 문제를 감지하지 못한다는 것이다. 둘 다 최신 오프셋만 추적하기 때문이다. Confluent의 Control Center는 메시지 개수와 체크섬을 모니터링해 이 모니터링 공백을 메운다.

메트릭 모니터링. MirrorMaker에는 프로듀서와 컨슈머가 들어 있고, 둘 다 많은 지표를 제공한다. 튜닝에 유용했던 지표는 다음과 같다.

- 컨슈머: fetch-size-avg, fetch-size-max, fetch-rate, fetch-throttle-time-avg, fetch-throttle-time-max - 프로듀서: batch-size-avg, batch-size-max, requests-in-flight, record-retry-rate - 공통: io-ratio, io-wait-ratio

카나리. 나머지를 다 모니터링한다면 카나리가 꼭 필요하지는 않지만, 모니터링 계층을 여러 겹 두기 위해 추가할 만하다. 매분 소스 클러스터의 특별한 토픽에 이벤트를 보내고 대상 클러스터에서 그 이벤트를 읽어 보는 프로세스다. 이벤트 도착이 허용 시간을 넘기면 경보를 울린다. 이는 MirrorMaker가 뒤처져 있거나 아예 살아 있지 않다는 뜻일 수 있다.

튜닝 — num.streams, 리눅스 네트워크, 프로듀서·컨슈머 설정

MirrorMaker 클러스터의 사이징은 필요한 처리량과 감내 가능한 랙에 달려 있다. 랙을 전혀 감내할 수 없다면 최대 처리량을 따라갈 만큼의 용량으로 사이징해야 한다. 어느 정도의 랙을 감내할 수 있다면 시간의 95~99% 동안 75~80% 사용률이 되도록 사이징한다. 그러면 피크 처리량에서 랙이 생기겠지만, 평소 여유 용량이 있으므로 피크가 지나면 따라잡는다.

다음으로 컨슈머 스레드 수(num.streams)를 바꿔 가며 처리량을 측정한다. 책은 LinkedIn이 컨슈머 스레드 8개로 6MB/s, 16개로 12MB/s를 얻었다는 대략적인 수치를 제시하지만, 이는 하드웨어·데이터센터·클라우드 제공자에 크게 좌우되므로 직접 테스트해야 한다. Kafka에 함께 오는 kafka-performance-producer 도구로 소스 클러스터에 부하를 만든 뒤, MirrorMaker를 붙여 그 부하를 미러링하게 한다. 컨슈머 스레드를 1, 2, 4, 8, 16, 24, 32로 바꿔 가며 테스트하고, 성능이 꺾이는 지점을 찾아 그 바로 아래로 num.streams를 설정한다.

압축된 이벤트를 소비하거나 생산한다면(크로스 데이터센터 미러링에서는 대역폭이 주요 병목이므로 권장된다) MirrorMaker가 이벤트를 압축 해제하고 다시 압축해야 한다. 이는 CPU를 많이 쓰므로 스레드를 늘리면서 CPU 사용률을 계속 지켜봐야 한다. 이 과정으로 단일 MirrorMaker 인스턴스의 최대 처리량을 찾고, 부족하면 인스턴스를 늘리고 그 다음에 서버를 늘린다.

또한 지연이 절대적으로 낮아야 하고 미러가 소스에 최대한 가까워야 하는 민감한 토픽들은 자체 컨슈머 그룹을 가진 별도의 MirrorMaker 클러스터로 분리하는 것이 좋다. 그래야 비대해진 토픽이나 통제 불능인 프로듀서가 가장 민감한 데이터 파이프라인을 느리게 만들지 못한다.

MirrorMaker 자체에 대한 튜닝은 사실상 여기까지다. 그 다음은 각 컨슈머 스레드와 인스턴스의 처리량을 올리는 일이다.

데이터센터를 가로질러 돌린다면 리눅스 네트워크 설정을 최적화한다. - TCP 버퍼 크기 증가: net.core.rmem_default, net.core.rmem_max, net.core.wmem_default, net.core.wmem_max, net.core.optmem_max - 자동 윈도우 스케일링 활성화: sysctl -w net.ipv4.tcp_window_scaling=1 (또는 /etc/sysctl.conf에 추가) - TCP slow start 시간 감소: /proc/sys/net/ipv4/tcp_slow_start_after_idle을 0으로

프로듀서와 컨슈머 중 어느 쪽이 병목인지 먼저 판단해야 한다. 프로듀서가 컨슈머의 데이터를 기다리는가, 반대인가? 모니터링 중인 프로듀서·컨슈머 지표를 보고 한쪽이 놀고 있고 다른 쪽이 꽉 차 있다면 답이 나온다. 다른 방법은 jstack으로 스레드 덤프를 여러 번 떠서 MirrorMaker 스레드가 대부분의 시간을 poll에 쓰는지 send에 쓰는지 보는 것이다. poll에 더 많은 시간을 쓰면 보통 컨슈머가 병목이고, send에 더 많은 시간을 쓰면 프로듀서가 병목이다.

프로듀서 튜닝: - max.in.flight.requests.per.connection: MirrorMaker는 기본적으로 in-flight 요청을 하나만 허용한다. 즉 프로듀서가 보낸 모든 요청이 대상 클러스터에서 확인되어야 다음 메시지를 보낸다. 브로커의 확인까지 지연이 크면 처리량이 크게 제한된다. MirrorMaker가 in-flight 요청 수를 제한하는 이유는, 일부 메시지가 성공적으로 확인되기까지 여러 번 재시도해야 하는 상황에서도 Kafka가 메시지 순서를 보존하도록 보장하는 유일한 방법이기 때문이다. 유스케이스에서 메시지 순서가 중요하지 않다면 이 값을 올려 처리량을 크게 높일 수 있다. - linger.ms와 batch.size: batch-size-avg와 batch-size-max 지표가 설정된 batch.size보다 낮다면 프로듀서가 계속 덜 찬 배치를 보내고 있다는 뜻이다. linger.ms를 올려 프로듀서가 배치가 찰 때까지 몇 밀리초 기다리게 하면 약간의 지연을 대가로 처리량이 오른다. 이미 꽉 찬 배치를 보내고 있고 메모리에 여유가 있다면 batch.size를 키워 더 큰 배치를 보낸다.

컨슈머 튜닝: - 파티션 할당 전략: MirrorMaker의 파티션 할당 전략은 기본이 range다. range에도 이점이 있어 컨슈머의 일반적인 기본값이지만, 파티션이 컨슈머에 고르지 않게 배분될 수 있다. MirrorMaker에서는, 특히 많은 토픽과 파티션을 미러링할 때는 round robin으로 바꾸는 편이 대개 낫다. 컨슈머 프로퍼티 파일에 partition.assignment.strategy=org.apache.kafka.clients.consumer.RoundRobinAssignor를 추가한다. - fetch.max.bytes: fetch-size-avg와 fetch-size-max가 fetch.max.bytes에 근접해 있다면 컨슈머가 허용된 만큼 최대로 읽고 있다는 뜻이다. 메모리 여유가 있다면 fetch.max.bytes를 올려 요청당 더 많은 데이터를 읽게 한다. - fetch.min.bytes와 fetch.max.wait: fetch-rate가 높다면 컨슈머가 브로커에 요청을 너무 많이 보내면서 요청당 충분한 데이터를 받지 못하고 있다는 뜻이다. 두 값을 올려 컨슈머가 요청당 더 많은 데이터를 받고 브로커가 충분한 데이터가 모일 때까지 기다렸다 응답하게 한다.

대안 — Uber uReplicator와 Confluent Replicator

MirrorMaker는 Apache Kafka에 포함되어 있지만 실전에서 몇 가지 한계가 있다. 두 가지 대안이 각각 다른 문제를 겨냥한다.

Uber uReplicator. Uber는 MirrorMaker를 매우 큰 규모로 운영했는데, 토픽·파티션 수와 클러스터 처리량이 늘면서 두 가지 문제에 부딪혔다.

첫째, 리밸런싱 지연이다. MirrorMaker의 컨슈머는 그냥 컨슈머다. MirrorMaker 스레드를 추가하거나, 인스턴스를 추가하거나, 인스턴스를 재기동하거나, 심지어 whitelist 정규식에 매치되는 새 토픽을 추가하기만 해도 컨슈머 리밸런싱이 일어난다. 리밸런싱은 새 파티션이 각 컨슈머에 할당될 때까지 모든 컨슈머를 멈춘다. 토픽·파티션이 아주 많으면 오래 걸린다. Uber가 쓰던 구버전 컨슈머에서는 특히 그랬다. 어떤 경우에는 5~10분의 무활동이 생겨 미러링이 뒤처지고 큰 백로그가 쌓였으며, 이를 회복하는 데 오랜 시간이 걸려 대상 클러스터에서 읽는 컨슈머의 지연이 매우 높아졌다.

둘째, 토픽 추가의 어려움이다. 토픽 whitelist에 정규식을 쓰면 누군가 매치되는 토픽을 소스 클러스터에 추가할 때마다 MirrorMaker가 리밸런싱한다. 이를 피하려고 Uber는 미러링할 토픽을 전부 나열하기로 했다. 그러면 예기치 않은 리밸런싱은 없어지지만, 새 토픽을 모든 MirrorMaker 인스턴스의 whitelist에 수동으로 추가하고 인스턴스를 재기동해야 하며 이는 다시 리밸런싱을 부른다. 최소한 그 리밸런싱은 예정된 유지보수 시점에 일어나지만 여전히 손이 많이 간다. 게다가 유지보수가 잘못되어 인스턴스마다 토픽 목록이 다르면, 컨슈머들이 구독 토픽에 합의하지 못해 MirrorMaker가 시작 후 끝없이 리밸런싱한다.

이 문제들 때문에 Uber는 uReplicator라는 자체 MirrorMaker 클론을 만들었다. Apache Helix를 중앙(그러나 고가용성인) 컨트롤러로 두어 토픽 목록과 각 uReplicator 인스턴스에 할당된 파티션을 관리하게 했다. 관리자는 REST API로 Helix의 목록에 새 토픽을 추가하고, uReplicator가 컨슈머들에 파티션을 할당한다. 이를 위해 Uber는 MirrorMaker의 Kafka 컨슈머를 자체 제작한 Helix 컨슈머로 교체했다. 이 컨슈머는 컨슈머들 사이의 합의가 아니라 Apache Helix 컨트롤러로부터 파티션 할당을 받는다. 그 결과 리밸런싱을 피하고 Helix에서 오는 할당 변경만 수신하면 된다. 다만 책 집필 시점에 Uber 외에 uReplicator를 쓰는 회사는 알려져 있지 않았다. 대부분의 회사가 Uber 규모로 운영하지 않아 같은 문제를 겪지 않거나, Apache Helix라는 새 컴포넌트에 대한 의존이 프로젝트 전체의 복잡도를 올리기 때문일 것이다.

Confluent Replicator. 이름은 비슷하지만 uReplicator와 공통점이 거의 없다. MirrorMaker의 서로 다른 문제 집합에 대한 다른 해법이다. Confluent는 엔터프라이즈 고객이 멀티클러스터 배포를 MirrorMaker로 관리할 때 겪은 문제를 겨냥했다.

첫째, 클러스터 설정의 발산이다. MirrorMaker는 소스와 대상 사이에서 데이터를 동기화할 뿐 그 외에는 아무것도 동기화하지 않는다. 토픽의 파티션 수, 리플리케이션 팩터, 토픽 레벨 설정이 서로 달라질 수 있다. 소스 클러스터에서 토픽 보관 기간을 1주에서 3주로 늘리고 DR 클러스터를 잊어버리면, 페일오버 시 몇 주치 데이터가 사라진 것을 발견하는 불쾌한 상황이 생긴다. 이 설정들을 수동으로 맞추는 것은 오류가 나기 쉽고, 시스템이 어긋나면 다운스트림 애플리케이션이나 복제 자체가 실패할 수도 있다.

둘째, 클러스터 관리 부담이다. MirrorMaker는 보통 여러 인스턴스의 클러스터로 배포된다. 즉 배포·모니터링·관리해야 할 클러스터가 또 하나 늘어난다. 설정 파일 두 개와 수많은 파라미터 때문에 MirrorMaker 자체의 설정 관리도 만만치 않다. 클러스터가 둘을 넘고 단방향 복제가 아니면 더 심해진다. Active-Active 클러스터가 셋이면 배포·모니터링·설정해야 할 MirrorMaker 클러스터가 여섯이고, 각각 최소 세 인스턴스를 갖는다. Active-Active 클러스터가 다섯이면 MirrorMaker 클러스터는 스무 개가 된다.

Confluent는 Replicator를 Kafka Connect 프레임워크의 소스 커넥터로 구현했다. 데이터베이스가 아니라 다른 Kafka 클러스터에서 데이터를 읽는 소스 커넥터인 셈이다. Connect 아키텍처에서 각 커넥터는 설정 가능한 수의 태스크로 작업을 나눈다. Replicator에서 각 태스크는 컨슈머와 프로듀서 한 쌍이다. Connect 프레임워크가 이 태스크들을 필요에 따라 Connect 워커 노드에 할당하므로, 한 서버에 여러 태스크가 있을 수도 있고 여러 서버에 분산될 수도 있다. 이로써 '인스턴스당 몇 개의 MirrorMaker 스트림을, 머신당 몇 개의 인스턴스를 돌릴 것인가'라는 수작업이 사라진다. Connect에는 커넥터와 태스크 설정을 중앙에서 관리하는 REST API도 있다. 대부분의 Kafka 배포가 (DB 변경 이벤트를 Kafka로 보내는 것이 매우 흔한 유스케이스이므로) 어차피 Kafka Connect를 포함한다고 가정하면, Replicator를 Connect 안에서 돌림으로써 관리해야 할 클러스터 수를 줄일 수 있다. 또 하나의 중요한 개선은, Replicator 커넥터가 토픽 목록의 데이터뿐 아니라 그 토픽들의 설정도 Zookeeper에서 복제한다는 점이다.