Kafka

이진 프로토콜과 요청 헤더

Kafka는 TCP 위에서 동작하는 이진 프로토콜로 요청 형식과 응답 방식을 규정한다. 연결은 항상 클라이언트가 시작하고 요청을 보내며, 브로커는 이를 처리해 응답한다. 특정 클라이언트가 브로커에 보낸 모든 요청은 도착한 순서대로 처리되는데, 바로 이 보장이 Kafka가 메시지 큐처럼 동작하며 순서 보장을 제공할 수 있게 하는 근거다.

모든 요청에는 표준 헤더가 있고 다음을 포함한다.

- 요청 타입(API key) - 요청 버전 — 브로커가 서로 다른 버전의 클라이언트를 구분해 응답할 수 있게 한다 - Correlation ID — 요청을 유일하게 식별하는 번호로 응답과 에러 로그에도 함께 나타나 트러블슈팅에 쓰인다 - Client ID — 요청을 보낸 애플리케이션을 식별한다

acceptor → 네트워크 스레드 → 요청 큐 → IO 스레드

브로커는 리스닝하는 포트마다 acceptor 스레드를 하나 실행한다. acceptor는 연결을 만들어 처리용 processor 스레드에게 넘긴다. processor 스레드(네트워크 스레드라고도 한다)의 개수는 설정 가능하다.

네트워크 스레드는 클라이언트 연결에서 요청을 꺼내 요청 큐에 넣고, 응답 큐에서 응답을 집어 클라이언트에게 돌려보내는 일을 담당한다. 요청 큐에 들어간 요청을 실제로 꺼내 처리하는 것은 IO 스레드다. 즉 네트워크 스레드는 입출력 중계를, IO 스레드는 요청 처리를 맡는 이원 구조이며, 모니터링에서 보게 되는 큐 크기와 스레드 관련 설정들이 정확히 이 구조를 가리킨다.

가장 흔한 요청 타입은 두 가지다. Produce 요청은 프로듀서가 보내며 브로커에 기록할 메시지를 담고 있고, Fetch 요청은 컨슈머와 팔로워 복제본이 메시지를 읽을 때 보낸다.

요청은 반드시 리더에게 가야 한다 — Metadata 요청

Produce 요청과 Fetch 요청은 모두 해당 파티션의 리더 복제본에게 보내야 한다. 리더가 다른 브로커에 있는데 요청이 잘못 도착하면 클라이언트는 "Not a Leader for Partition" 오류 응답을 받는다. 올바른 브로커로 요청을 보내는 책임은 Kafka 클라이언트에게 있다.

그렇다면 클라이언트는 어디로 보내야 할지 어떻게 알까? Metadata 요청을 쓴다. 클라이언트가 관심 있는 토픽 목록을 담아 보내면, 서버는 그 토픽에 어떤 파티션이 있는지, 각 파티션의 복제본은 무엇인지, 그중 리더가 누구인지를 응답한다. 모든 브로커가 이 정보를 담은 메타데이터 캐시를 갖고 있으므로 Metadata 요청은 아무 브로커에게나 보내도 된다.

클라이언트는 이 정보를 캐시해 두었다가 파티션별로 올바른 브로커에 요청을 보내고, `metadata.max.age.ms` 설정 주기에 따라 갱신한다. 또한 "Not a Leader" 오류를 받으면 자신이 낡은 정보를 쓰고 있다는 뜻이므로 재시도 전에 메타데이터를 먼저 갱신한다.

Produce 요청 처리와 purgatory

리더 복제본을 가진 브로커가 Produce 요청을 받으면 먼저 몇 가지 검증을 수행한다.

- 데이터를 보내는 사용자가 해당 토픽에 쓰기 권한이 있는가 - 요청에 지정된 acks 값이 유효한가(0, 1, all만 허용) - acks가 all이면, 메시지를 안전하게 쓰기에 충분한 in-sync 복제본이 있는가 (브로커는 ISR 수가 설정된 값 아래로 떨어지면 새 메시지를 거부하도록 구성할 수 있다)

검증을 통과하면 브로커는 새 메시지를 로컬 디스크에 쓴다. 다만 Linux에서 메시지는 파일시스템 캐시에 기록될 뿐이며 언제 실제 디스크에 내려갈지는 보장되지 않는다. Kafka는 데이터가 디스크에 영속화되기를 기다리지 않고 메시지 내구성을 복제에 의존한다.

리더에 기록이 끝나면 브로커는 acks 설정을 본다. acks가 0이나 1이면 즉시 응답하고, all이면 요청을 purgatory라는 버퍼에 담아 두었다가 팔로워 복제본들이 메시지를 복제한 것을 리더가 확인한 시점에 응답을 보낸다.

Fetch 요청, zero-copy, 그리고 상·하한

Fetch 요청은 "토픽 Test의 파티션 0에서 오프셋 53부터, 파티션 3에서 오프셋 64부터 보내 달라"처럼 토픽·파티션·오프셋 목록으로 구성된다. 클라이언트는 파티션별로 브로커가 반환할 수 있는 데이터 양의 상한도 지정한다. 응답을 담을 메모리를 클라이언트가 미리 할당해야 하므로, 이 상한이 없으면 브로커가 큰 응답을 보내 클라이언트가 OOM에 빠질 수 있기 때문이다.

리더는 요청이 유효한지 — 해당 파티션에 그 오프셋이 실제로 존재하는지 — 먼저 확인한다. 이미 삭제될 만큼 오래된 메시지를 요구하거나 아직 존재하지 않는 오프셋을 요구하면 오류로 응답한다. 오프셋이 유효하면 브로커는 상한까지 메시지를 읽어 클라이언트에게 보내는데, 이때 zero-copy 방식을 사용해 파일(정확히는 Linux 파일시스템 캐시)에서 네트워크 채널로 중간 버퍼 없이 바로 전송한다. 데이터를 로컬 캐시에 담았다가 보내는 대부분의 데이터베이스와 다른 점이며, 바이트 복사와 메모리 버퍼 관리 오버헤드를 없애 성능을 크게 끌어올린다.

클라이언트는 상한뿐 아니라 하한도 지정할 수 있다. 하한을 10K로 두면 "보낼 데이터가 최소 10K는 모였을 때만 응답하라"는 뜻이다. 트래픽이 적은 토픽을 읽을 때 몇 밀리초마다 요청을 던지고 빈손으로 돌아오는 낭비를 없애 CPU와 네트워크 사용량을 줄인다. 물론 영원히 기다릴 수는 없으므로 클라이언트는 타임아웃도 함께 지정해 "x밀리초 안에 최소량을 못 채웠으면 있는 만큼만 보내라"고 알린다.

컨슈머는 모든 ISR에 복제된 메시지만 볼 수 있다

리더에 존재하는 데이터가 전부 클라이언트에게 읽히는 것은 아니다. 대부분의 클라이언트는 모든 in-sync 복제본에 기록된 메시지까지만 읽을 수 있다(팔로워 복제본은 컨슈머이지만 이 제약에서 예외다. 그렇지 않으면 복제 자체가 불가능하다). 리더는 어떤 메시지가 어떤 복제본까지 복제되었는지 알고 있고, 모든 ISR에 기록되기 전까지는 그 메시지를 컨슈머에게 보내지 않는다. 이 구간의 메시지를 가져오려는 시도는 오류가 아니라 빈 응답을 받는다.

이유는 아직 충분히 복제되지 않은 메시지가 "안전하지 않기" 때문이다. 리더가 죽고 다른 복제본이 그 자리를 대신하면 그 메시지들은 Kafka에서 사라진다. 만약 리더에만 존재하는 메시지를 읽게 허용한다면, 어떤 컨슈머는 읽었지만 이후 아무도 읽을 수 없는 메시지가 생겨 일관성이 깨진다.

이 동작에는 대가가 있다. 브로커 간 복제가 어떤 이유로 느려지면 새 메시지가 컨슈머에게 도달하는 데도 그만큼 더 걸린다. 이 지연은 `replica.lag.time.max.ms`, 즉 복제본이 뒤처져 있어도 여전히 in-sync로 인정되는 시간만큼으로 제한된다.

기타 요청과 프로토콜 버전 호환성

같은 프로토콜이 브로커들 사이의 통신에도 쓰인다. 예를 들어 컨트롤러가 어떤 파티션에 새 리더가 생겼음을 알릴 때, 새 리더에게는 클라이언트 요청을 받기 시작하라는 의미로, 팔로워들에게는 새 리더를 따르라는 의미로 LeaderAndIsr 요청을 보낸다. 이런 요청은 내부용이며 클라이언트가 사용해서는 안 된다. 책 집필 시점에 프로토콜은 20종의 요청 타입을 다루었고 계속 늘고 있다.

프로토콜은 계속 진화한다. 과거에는 컨슈머가 오프셋을 ZooKeeper에 저장했지만, 이를 Kafka의 특수 토픽에 저장하도록 바꾸면서 OffsetCommitRequest, OffsetFetchRequest, ListOffsetsRequest가 프로토콜에 추가되었다. 이제 애플리케이션이 commitOffset()을 호출하면 클라이언트는 ZooKeeper가 아니라 Kafka에 OffsetCommitRequest를 보낸다.

기존 요청에 버전을 올려 기능을 추가하기도 한다. 0.9.0과 0.10.0 사이에 Metadata 응답에 현재 컨트롤러 정보를 담기로 하면서 Metadata 요청/응답에 새 버전이 생겼다. 0.9.0 클라이언트는 버전 0 요청을 보내고 브로커는 컨트롤러 정보가 없는 버전 0으로 응답하며, 0.10.0 클라이언트는 버전 1 요청을 보내 컨트롤러 정보를 받는다. 반대로 0.10.0 클라이언트가 0.9.0 브로커에 버전 1 요청을 보내면 브로커가 처리하지 못해 오류로 응답한다. 새 브로커는 옛 요청을 처리할 줄 알지만 그 역은 성립하지 않는다. 그래서 클라이언트보다 브로커를 먼저 업그레이드하라는 권고가 나온다. 0.10.0에서는 ApiVersionRequest가 추가되어, 클라이언트가 브로커에게 각 요청의 지원 버전을 물어보고 적절한 버전을 골라 쓸 수 있게 되었다.