Kafka

신뢰성과 메모리에 영향을 주는 설정

acks는 프로듀서가 쓰기를 성공으로 간주하기 전에 몇 개의 파티션 복제본이 레코드를 받아야 하는지를 제어한다. 메시지 유실 가능성에 가장 큰 영향을 주는 설정이다.

- acks=0: 브로커의 응답을 기다리지 않고 성공으로 간주한다. 문제가 생겨 브로커가 메시지를 못 받아도 프로듀서는 알 수 없고 메시지는 유실된다. 대신 네트워크가 감당하는 만큼 빠르게 보낼 수 있어 매우 높은 처리량을 낼 수 있다. - acks=1: 리더 복제본이 메시지를 받는 순간 성공 응답을 받는다. 리더에 기록할 수 없으면(예: 리더가 죽고 새 리더가 아직 선출되지 않은 상태) 오류 응답을 받아 재시도할 수 있어 유실을 피할 수 있다. 다만 리더가 죽고 이 메시지를 갖지 못한 복제본이 unclean leader election으로 새 리더가 되면 여전히 유실될 수 있다. - acks=all: 모든 in-sync 복제본이 메시지를 받으면 성공 응답을 받는다. 한 대 이상의 브로커가 메시지를 갖고 있음을 보장하므로 가장 안전하지만, 여러 브로커를 기다리므로 지연이 acks=1보다 더 커진다.

buffer.memory는 브로커로 보내기를 기다리는 메시지를 버퍼링하는 데 쓸 메모리 양이다. 애플리케이션이 전송 속도보다 빠르게 메시지를 만들면 공간이 부족해지고, 이후 send() 호출은 블로킹되거나 예외를 던진다. 이 동작은 block.on.buffer.full 파라미터가 결정했는데, 0.9.0.0 릴리스에서 max.block.ms로 대체되어 일정 시간 블로킹한 뒤 예외를 던지는 방식이 되었다.

max.block.ms는 send()를 호출할 때, 그리고 partitionsFor()로 메타데이터를 명시적으로 요청할 때 프로듀서가 블로킹하는 시간을 제어한다. 이 두 메서드는 전송 버퍼가 가득 찼거나 메타데이터를 사용할 수 없을 때 블로킹하며, max.block.ms에 도달하면 타임아웃 예외를 던진다.

client.id는 임의의 문자열로, 브로커가 클라이언트를 식별하는 데 쓴다. 로깅과 메트릭, 그리고 쿼터에 사용된다.

처리량과 지연을 조율하는 설정 — 압축, 재시도, 배치

compression.type은 기본적으로 압축하지 않는다. snappy, gzip, lz4로 설정하면 브로커로 보내기 전에 해당 알고리즘으로 압축한다. Snappy는 Google이 만든 것으로 CPU 오버헤드가 낮으면서 준수한 압축률과 성능을 내므로 성능과 대역폭을 모두 신경 써야 할 때 권장된다. gzip은 CPU와 시간을 더 쓰지만 압축률이 좋아 네트워크 대역폭이 제한적일 때 권장된다. 압축은 네트워크 사용량과 저장 공간을 줄여 준다.

retries는 서버가 일시적 오류(예: 파티션에 리더가 없는 상태)를 반환했을 때 프로듀서가 몇 번까지 재전송할지를 제어한다. 재시도 사이에는 기본적으로 100ms를 기다리며 retry.backoff.ms로 조정한다. 권장되는 접근은 브로커가 죽었을 때 복구되는 데(즉 모든 파티션이 새 리더를 갖게 되는 데) 걸리는 시간을 실제로 측정한 뒤, 재시도 횟수와 간격의 곱이 그 복구 시간보다 길어지도록 설정하는 것이다. 그렇지 않으면 프로듀서가 너무 일찍 포기한다.

batch.size는 같은 파티션으로 갈 레코드들을 묶는 배치 하나에 쓸 메모리 양을 바이트 단위로 지정한다. 메시지 개수가 아니라 바이트라는 점이 중요하다. 배치가 가득 차면 배치의 모든 메시지가 전송된다. 하지만 이것이 프로듀서가 배치가 찰 때까지 기다린다는 뜻은 아니다. 프로듀서는 절반만 찬 배치도, 메시지 한 건만 든 배치도 그냥 보낸다. 따라서 batch.size를 너무 크게 잡아도 전송이 지연되지는 않고 배치에 메모리를 더 쓸 뿐이며, 너무 작게 잡으면 전송 횟수가 늘어 오버헤드가 커진다.

linger.ms는 현재 배치를 보내기 전에 추가 메시지를 기다리는 시간이다. KafkaProducer는 현재 배치가 가득 찼을 때 또는 linger.ms에 도달했을 때 배치를 보낸다. 기본적으로 프로듀서는 전송 스레드가 가용하면 배치에 메시지가 하나뿐이어도 곧바로 보낸다. linger.ms를 0보다 크게 두면 몇 밀리초를 더 기다려 배치에 메시지를 더 모으게 되며, 지연은 늘지만 메시지당 오버헤드가 줄어 처리량이 올라간다.

max.in.flight.requests.per.connection은 응답을 받기 전에 서버로 보낼 수 있는 메시지 수다. 크게 잡으면 메모리 사용은 늘지만 처리량이 개선되고, 지나치게 크게 잡으면 배치 효율이 떨어져 오히려 처리량이 낮아진다. 이 값을 1로 두면 재시도가 발생하더라도 메시지가 보낸 순서대로 브로커에 기록되는 것이 보장된다.

타임아웃과 크기 제한 설정

request.timeout.ms는 프로듀서가 데이터를 보낸 뒤 서버 응답을 기다리는 시간, metadata.fetch.timeout.ms는 쓰려는 파티션의 현재 리더 같은 메타데이터를 요청했을 때 기다리는 시간이다. 응답 없이 타임아웃에 도달하면 프로듀서는 재시도하거나 오류로 응답한다(예외 또는 send 콜백을 통해).

timeout.ms는 성격이 다르다. 이것은 브로커 쪽 시간으로, acks 설정을 충족하기 위해 브로커가 in-sync 복제본들의 확인을 기다리는 시간이다. 이 시간이 지나도 필요한 확인이 오지 않으면 브로커가 오류를 반환한다.

max.request.size는 프로듀서가 보내는 produce 요청의 크기를 제어한다. 이 값은 보낼 수 있는 가장 큰 메시지의 크기와, 한 요청에 담을 수 있는 메시지 개수를 동시에 제한한다. 예를 들어 기본값이 1MB라면 보낼 수 있는 최대 메시지는 1MB이거나, 1KB짜리 메시지 1,000개를 한 요청으로 묶을 수 있다.

여기서 놓치기 쉬운 점은 브로커에도 별도의 상한이 있다는 것이다. 브로커는 message.max.bytes로 받아들일 최대 메시지 크기를 제한한다. 프로듀서가 브로커에 거부당할 크기의 메시지를 보내려 시도하지 않도록, 두 설정을 맞춰 두는 것이 좋다.

receive.buffer.bytes와 send.buffer.bytes는 데이터를 읽고 쓸 때 소켓이 사용하는 TCP 수신·송신 버퍼 크기다. -1로 두면 OS 기본값을 쓴다. 프로듀서나 컨슈머가 다른 데이터센터의 브로커와 통신할 때는 그 네트워크 링크가 대개 지연이 높고 대역폭이 낮으므로 이 값들을 키우는 것이 좋다.

순서 보장이 깨지는 지점

Kafka는 파티션 내에서 메시지 순서를 보존한다. 프로듀서가 특정 순서로 보냈다면 브로커는 그 순서로 파티션에 기록하고 모든 컨슈머도 그 순서로 읽는다. 계좌에 100달러를 입금한 뒤 출금하는 것과 그 반대는 완전히 다른 일이므로, 어떤 유스케이스에서는 순서가 매우 중요하다.

그런데 retries를 0이 아닌 값으로 두고 in-flight 요청을 1보다 크게 두면 순서가 뒤집힐 수 있다. 브로커가 첫 번째 배치 기록에 실패하고, 이미 전송 중이던 두 번째 배치는 기록에 성공한 뒤, 첫 번째 배치를 재시도해서 성공하는 시나리오가 가능하기 때문이다. 결과적으로 두 번째 배치가 첫 번째보다 앞에 기록된다.

신뢰성 있는 시스템에서 재시도 횟수를 0으로 두는 것은 보통 선택지가 아니다. 따라서 순서 보장이 결정적으로 중요하다면 in-flight 요청 수를 1로 설정해, 한 배치가 재시도되는 동안 추가 메시지가 전송되지 않게 해야 한다. 이 설정은 프로듀서 처리량을 크게 떨어뜨리므로 순서가 정말 중요할 때만 써야 한다.

커스텀 직렬화의 함정과 Avro

프로듀서 설정에는 직렬화기가 필수다. String, Integer, ByteArray 직렬화기는 기본 제공되지만 대부분의 유스케이스를 덮지 못한다. 결국 좀 더 일반적인 레코드를 직렬화해야 한다.

예를 들어 customerID와 customerName을 가진 Customer 클래스를 만들고, 4바이트 int(ID) + 4바이트 int(이름 길이) + N바이트 UTF-8 이름 형태로 직접 직렬화하는 CustomerSerializer를 작성할 수 있다. Serializer 인터페이스의 configure, serialize, close를 구현하면 된다.

문제는 이 코드가 매우 취약하다는 것이다. 고객이 많아져 customerID를 Long으로 바꿔야 하거나 startDate 필드를 추가해야 하면, 구버전과 신버전 메시지 사이의 호환성 유지가 심각한 문제가 된다. 직렬화기와 역직렬화기 버전 간 호환성 디버깅은 원시 바이트 배열을 비교해야 하는 일이라 매우 어렵다. 게다가 같은 회사의 여러 팀이 Customer 데이터를 Kafka에 쓴다면, 모두가 같은 직렬화기를 써야 하고 코드를 정확히 같은 시점에 수정해야 한다.

그래서 책은 JSON, Apache Avro, Thrift, Protobuf 같은 기존 직렬화 라이브러리를 쓸 것을 강하게 권장한다.

Avro는 언어 중립적인 데이터 직렬화 포맷이다. Doug Cutting이 데이터 파일을 넓은 사용자층과 공유하기 위해 만들었다. 스키마는 보통 JSON으로 기술하고 직렬화는 보통 바이너리로 하며, JSON으로 직렬화하는 것도 지원한다. Avro는 읽고 쓸 때 스키마가 존재한다고 가정하는데, 파일의 경우 대개 스키마를 파일 자체에 임베드한다.

Kafka 같은 메시징 시스템에 잘 맞는 이유는, 메시지를 쓰는 애플리케이션이 새 스키마로 전환해도 읽는 애플리케이션은 아무 변경 없이 계속 처리할 수 있다는 점이다. 예를 들어 원래 스키마에 id, name, faxNumber(옵션, 기본값 null)가 있었고 새 스키마에서 faxNumber를 빼고 email(옵션, 기본값 null)을 넣었다고 하자. 구버전 읽기 애플리케이션이 신버전 레코드를 만나면 getName()과 getId()는 그대로 동작하고 getFaxNumber()는 null을 반환한다. 반대로 신버전 읽기 애플리케이션이 구버전 레코드를 만나면 getEmail()이 null을 반환한다. 예외도, 기존 데이터의 값비싼 갱신도 필요 없다.

다만 두 가지 전제가 있다. 첫째, 쓰기에 사용한 스키마와 읽는 애플리케이션이 기대하는 스키마가 호환되어야 한다(호환 규칙은 Avro 문서에 있다). 둘째, 역직렬화기는 애플리케이션이 기대하는 스키마와 다르더라도 데이터를 쓸 때 사용된 스키마에 접근할 수 있어야 한다.

Kafka에서 Avro 쓰기 — 스키마 레지스트리

Avro 파일에서는 전체 스키마를 데이터 파일에 저장해도 오버헤드가 합리적이지만, 레코드마다 전체 스키마를 저장하면 레코드 크기가 보통 두 배 이상이 된다. 그런데 Avro는 레코드를 읽을 때 전체 스키마가 있어야 하므로, 스키마를 다른 곳에서 찾아야 한다.

이를 위해 스키마 레지스트리(Schema Registry)라는 아키텍처 패턴을 쓴다. 스키마 레지스트리는 Apache Kafka의 일부가 아니며 여러 오픈소스 선택지가 있다(책의 예제는 Confluent Schema Registry를 사용한다).

아이디어는 이렇다. Kafka에 데이터를 쓸 때 사용한 모든 스키마를 레지스트리에 저장하고, Kafka로 보내는 레코드에는 스키마의 식별자만 담는다. 컨슈머는 그 식별자로 레지스트리에서 스키마를 가져와 데이터를 역직렬화한다. 핵심은 이 모든 작업(레지스트리에 스키마 저장하기, 필요할 때 꺼내 오기)이 직렬화기와 역직렬화기 안에서 이루어진다는 점이다. Kafka에 데이터를 쓰는 코드는 다른 직렬화기를 쓰듯 그냥 Avro 직렬화기를 쓰면 된다.

props.put("key.serializer",
   "io.confluent.kafka.serializers.KafkaAvroSerializer");
props.put("value.serializer",
   "io.confluent.kafka.serializers.KafkaAvroSerializer");
props.put("schema.registry.url", schemaUrl);

AvroSerializer는 프리미티브 타입도 처리할 수 있어서 키는 String, 값은 Customer 객체로 둘 수 있다. Avro가 생성한 객체 대신 제네릭 객체를 쓰고 싶다면 스키마 문자열을 직접 파싱해 GenericRecord를 만들면 된다. 이 경우에도 직렬화기는 GenericRecord에서 스키마를 꺼내 레지스트리에 저장하고 데이터를 직렬화하는 일을 알아서 한다.

키와 파티셔닝 — 기본 파티셔너의 규칙

Kafka 메시지는 키-값 쌍이다. 토픽과 값만으로 ProducerRecord를 만들 수도 있고 이때 키는 기본적으로 null이 되지만, 대부분의 애플리케이션은 키를 함께 쓴다. 키는 두 가지 역할을 한다. 메시지와 함께 저장되는 추가 정보이며, 동시에 이 메시지가 토픽의 어느 파티션에 기록될지를 결정한다. 같은 키를 가진 모든 메시지는 같은 파티션으로 간다. 즉 어떤 프로세스가 토픽 파티션의 일부만 읽고 있다면, 특정 키의 모든 레코드는 같은 프로세스가 읽게 된다.

키가 null이고 기본 파티셔너를 쓰면 레코드는 토픽의 가용한(available) 파티션 중 하나로 무작위로 간다. 파티션 간 균형을 맞추기 위해 라운드로빈 알고리즘이 사용된다.

키가 있고 기본 파티셔너를 쓰면 Kafka는 키를 해싱해 그 결과로 메시지를 특정 파티션에 매핑한다. 여기서 Kafka는 Java의 해시가 아니라 자체 해시 알고리즘을 쓰는데, Java를 업그레이드해도 해시 값이 바뀌지 않게 하기 위해서다.

놓치기 쉬운 중요한 세부 사항이 하나 있다. 키가 항상 같은 파티션에 매핑되는 것이 중요하기 때문에, 매핑 계산에는 가용한 파티션이 아니라 토픽의 모든 파티션이 사용된다. 그래서 데이터를 쓰려는 시점에 특정 파티션이 가용하지 않으면 오류를 받을 수 있다(실제로는 드문 일이다). 키가 null일 때 가용 파티션 중에서 고르는 것과 대비되는 지점이다.

파티션 증설이 키 매핑을 깨뜨린다

키에서 파티션으로의 매핑은 토픽의 파티션 수가 변하지 않는 동안에만 일관성이 유지된다. 파티션 수가 그대로라면 예를 들어 user 045189에 관한 레코드는 항상 파티션 34에 기록된다고 확신할 수 있고, 이 성질은 파티션에서 데이터를 읽을 때 온갖 최적화를 가능하게 한다.

그러나 토픽에 새 파티션을 추가하는 순간 이 보장은 사라진다. 예전 레코드는 파티션 34에 남아 있는데 새 레코드는 다른 파티션에 기록된다. 즉 같은 키의 이력이 두 파티션에 걸쳐 흩어지고, 파티션 내에서만 보장되는 순서는 증설 시점을 기준으로 끊긴다.

키 기반 파티셔닝이 중요한 경우에 대한 책의 처방은 단순하다. 충분한 수의 파티션으로 토픽을 만들고 절대 파티션을 추가하지 않는 것이다.

커스텀 파티셔너 — 핫 키를 격리하기

Kafka는 해시 파티셔닝만 강요하지 않는다. 데이터를 다르게 분배해야 할 좋은 이유가 있을 수 있다.

책의 예는 이렇다. B2B 벤더인데 가장 큰 고객사 Banana가 일일 트랜잭션의 10% 이상을 차지한다. 기본 해시 파티셔닝을 쓰면 Banana 레코드가 다른 계정들과 같은 파티션에 배정되어 그 파티션 하나가 다른 파티션의 두 배 크기가 된다. 서버 디스크가 부족해지거나 처리가 느려질 수 있다. 원하는 것은 Banana에게 전용 파티션 하나를 주고, 나머지 계정은 해시 파티셔닝으로 나머지 파티션에 매핑하는 것이다.

Partitioner 인터페이스는 configure, partition, close 세 메서드를 갖는다. partition() 안에서 cluster.partitionsForTopic(topic)으로 파티션 목록을 얻고, 키가 "Banana"면 마지막 파티션을 반환하고, 그 외에는 Utils.murmur2(keyBytes)의 절댓값을 (numPartitions - 1)로 나눈 나머지를 반환하는 식이다.

예제 코드는 특정 고객 이름을 partition() 안에 하드코딩하고 있지만, 원래는 configure()를 통해 전달하는 것이 맞다는 점을 책도 지적한다. 또한 키가 없거나 String이 아니면 InvalidRecordException을 던져 계약을 명시한다.