Kafka

컨슈머 그룹이 확장을 만드는 방식

토픽에서 메시지를 읽어 검증한 뒤 다른 저장소에 쓰는 애플리케이션을 생각해 보자. 프로듀서가 쓰는 속도가 이 애플리케이션이 검증하는 속도를 넘어서면, 단일 컨슈머로는 점점 더 뒤처지게 된다. 그래서 여러 컨슈머가 같은 토픽을 읽으며 데이터를 나눠 갖는 방식이 필요하다.

Kafka 컨슈머는 보통 컨슈머 그룹의 일원이다. 같은 그룹에 속한 여러 컨슈머가 한 토픽을 구독하면, 각 컨슈머는 토픽 파티션의 서로 다른 부분집합에서 메시지를 받는다.

파티션 4개짜리 토픽 T1이 있다고 하자.

- 그룹 G1에 컨슈머 C1 하나뿐이면 C1이 4개 파티션의 모든 메시지를 받는다. - C2를 G1에 추가하면 각 컨슈머가 2개 파티션에서만 메시지를 받는다. - G1에 컨슈머가 4개면 각각 파티션 하나씩을 읽는다. - 파티션 수보다 많은 컨슈머를 한 그룹에 추가하면, 일부 컨슈머는 유휴 상태가 되어 메시지를 전혀 받지 못한다.

토픽에서 데이터 소비를 확장하는 주된 방법이 바로 그룹에 컨슈머를 추가하는 것이다. Kafka 컨슈머는 DB 쓰기나 시간이 많이 드는 연산 같은 고지연 작업을 하는 경우가 흔해서, 컨슈머 하나로는 유입 속도를 따라갈 수 없다. 각 컨슈머가 파티션과 메시지의 일부만 담당하게 해 부하를 나누는 것이 주된 확장 수단이다. 이 때문에 토픽을 만들 때 파티션을 넉넉히 두는 것이 좋다. 부하가 늘었을 때 컨슈머를 더 붙일 수 있기 때문이다. 다만 파티션 수보다 컨슈머를 더 붙이는 것은 의미가 없다.

또 다른 축이 있다. 단일 애플리케이션 확장이 아니라, 여러 애플리케이션이 같은 토픽의 데이터를 읽어야 하는 경우다. Kafka의 주요 설계 목표 중 하나가 토픽에 쓰인 데이터를 조직 전체의 많은 유스케이스에서 쓸 수 있게 하는 것이었다. 이 경우 각 애플리케이션이 부분집합이 아니라 전체 메시지를 받아야 하므로, 애플리케이션마다 자기 컨슈머 그룹을 갖게 하면 된다. 그룹 G2를 추가하면 G1이 무엇을 하든 상관없이 G2는 T1의 모든 메시지를 받는다. 많은 전통적 메시징 시스템과 달리 Kafka는 컨슈머와 컨슈머 그룹 수가 많아져도 성능이 떨어지지 않는다.

요약하면, 모든 메시지가 필요한 애플리케이션마다 새 컨슈머 그룹을 만들고, 읽기와 처리를 확장하려면 기존 그룹에 컨슈머를 추가한다.

리밸런스 — 필요하지만 반갑지 않은 일

그룹 내 컨슈머들은 구독한 토픽 파티션의 소유권을 공유한다. 새 컨슈머가 그룹에 들어오면 다른 컨슈머가 소비하던 파티션을 이어받기 시작한다. 컨슈머가 종료되거나 크래시해 그룹을 떠나면, 그 컨슈머가 소비하던 파티션을 남은 컨슈머 중 하나가 가져간다. 관리자가 새 파티션을 추가하는 등 그룹이 소비하는 토픽이 변경될 때도 파티션 재할당이 일어난다.

파티션 소유권이 한 컨슈머에서 다른 컨슈머로 옮겨 가는 것을 리밸런스라 한다. 리밸런스는 컨슈머 그룹에 고가용성과 확장성을 제공하므로(컨슈머를 쉽고 안전하게 추가·제거할 수 있게 해 준다) 중요하지만, 평상시에는 그리 달갑지 않은 일이다.

이유는 두 가지다. 첫째, 리밸런스 동안 컨슈머는 메시지를 소비할 수 없다. 리밸런스는 사실상 컨슈머 그룹 전체가 잠시 멈추는 짧은 가용성 공백이다. 둘째, 파티션이 다른 컨슈머로 옮겨 가면 그 컨슈머는 현재 상태를 잃는다. 캐시를 유지하고 있었다면 캐시를 다시 채워야 하고, 그동안 애플리케이션이 느려진다.

그룹 코디네이터와 하트비트

컨슈머가 그룹 멤버십과 할당받은 파티션의 소유권을 유지하는 방법은, 그룹 코디네이터로 지정된 Kafka 브로커에 하트비트를 보내는 것이다. 이 브로커는 컨슈머 그룹마다 다를 수 있다. 컨슈머가 정기적으로 하트비트를 보내는 한, 살아 있고 정상적으로 파티션의 메시지를 처리 중이라고 간주된다.

(초기 동작에서) 하트비트는 컨슈머가 poll할 때, 그리고 소비한 레코드를 커밋할 때 전송된다. 컨슈머가 충분히 오랫동안 하트비트를 보내지 않으면 세션이 타임아웃되고, 그룹 코디네이터는 그 컨슈머를 죽었다고 판단해 리밸런스를 유발한다. 컨슈머가 크래시해 처리가 멈추면 코디네이터가 죽었다고 판단하기까지 하트비트 없는 몇 초가 필요하고, 그 몇 초 동안 죽은 컨슈머가 소유하던 파티션에서는 아무 메시지도 처리되지 않는다.

반면 컨슈머를 정상적으로 닫으면(clean close), 컨슈머가 그룹 코디네이터에게 떠난다고 알리고 코디네이터는 즉시 리밸런스를 유발한다. 그래서 처리 공백이 줄어든다.

중요한 버전 변화가 하나 있다. 0.10.1 릴리스에서 커뮤니티는 별도의 하트비트 스레드를 도입해 poll과 poll 사이에도 하트비트를 보내게 했다. 덕분에 하트비트 주기(즉 컨슈머 크래시를 감지하는 데 걸리는 시간)와 poll 주기(브로커가 반환한 데이터를 처리하는 데 걸리는 시간이 결정)를 분리할 수 있게 되었다.

이 분리와 함께, 애플리케이션이 poll 없이 얼마나 오래 있을 수 있는지를 설정할 수 있게 되었다(max.poll.interval.ms). 이것은 애플리케이션이 크래시하지는 않았지만 어떤 이유로 진행하지 못하는 라이브락(livelock)을 막기 위한 설정이며, 컨슈머 크래시를 감지해 하트비트 전송이 멈추는 것을 다루는 session.timeout.ms와는 별개다. 처리 시간이 긴 레코드를 다뤄야 한다면 max.poll.interval.ms를 조정하면 된다.

파티션 할당은 누가 계산하는가

컨슈머가 그룹에 들어오려 하면 그룹 코디네이터에게 JoinGroup 요청을 보낸다. 그룹에 가장 먼저 들어온 컨슈머가 그룹 리더가 된다.

리더는 그룹 코디네이터로부터 그룹 내 모든 컨슈머 목록을 받는다(최근에 하트비트를 보내 살아 있다고 간주되는 컨슈머 전부). 그리고 각 컨슈머에게 파티션의 부분집합을 할당하는 책임을 진다. 이때 PartitionAssignor 구현체를 사용해 어떤 파티션을 어떤 컨슈머가 처리할지 결정한다.

할당을 결정한 뒤, 리더 컨슈머는 할당 목록을 GroupCoordinator에게 보내고 코디네이터가 이 정보를 모든 컨슈머에게 전달한다. 각 컨슈머는 자기 할당만 볼 수 있고, 그룹 내 전체 컨슈머 목록과 그 할당을 모두 아는 클라이언트 프로세스는 리더뿐이다. 이 과정은 리밸런스가 일어날 때마다 반복된다.

즉 이 판본의 프로토콜에서 할당 계산은 브로커가 아니라 클라이언트(그룹 리더) 쪽에서 이루어진다.

컨슈머 생성과 구독

KafkaConsumer 생성은 KafkaProducer와 비슷하다. Properties에 필수 프로퍼티 세 가지 bootstrap.servers, key.deserializer, value.deserializer를 넣는다. bootstrap.servers는 프로듀서와 동일하게 쓰이고, 역직렬화기는 Java 객체를 바이트 배열로 바꾸는 대신 바이트 배열을 Java 객체로 바꾸는 클래스를 지정한다.

네 번째 프로퍼티 group.id는 엄밀히 말해 필수는 아니지만 사실상 필수처럼 다룬다. 이 컨슈머 인스턴스가 속한 컨슈머 그룹을 지정한다. 어떤 그룹에도 속하지 않는 컨슈머를 만드는 것도 가능하지만 흔하지 않다.

Properties props = new Properties();
props.put("bootstrap.servers", "broker1:9092,broker2:9092");
props.put("group.id", "CountryCounter");
props.put("key.deserializer",
  "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer",
  "org.apache.kafka.common.serialization.StringDeserializer");
KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);

컨슈머를 만들었으면 토픽을 구독한다. subscribe()는 토픽 목록을 인자로 받는다.

consumer.subscribe(Collections.singletonList("customerCountries"));

subscribe는 정규표현식으로도 호출할 수 있다. 표현식은 여러 토픽 이름에 매칭될 수 있고, 누군가 매칭되는 이름의 새 토픽을 만들면 거의 즉시 리밸런스가 일어나며 컨슈머들이 그 새 토픽을 소비하기 시작한다. 여러 토픽을 소비하면서 서로 다른 종류의 데이터를 처리할 수 있는 애플리케이션에 유용하며, 실제로는 Kafka와 다른 시스템 사이에서 데이터를 복제하는 애플리케이션에서 가장 흔히 쓰인다.

poll 루프가 실제로 하는 일

컨슈머 API의 핵심은 서버에서 데이터를 가져오는 단순한 루프다. 구독이 끝나면 poll 루프가 코디네이션, 파티션 리밸런스, 하트비트, 데이터 페칭의 모든 세부 사항을 처리해 준다.

try {
  while (true) {
      ConsumerRecords<String, String> records = consumer.poll(100);
      for (ConsumerRecord<String, String> record : records) {
          // record.topic(), record.partition(), record.offset(),
          // record.key(), record.value()
      }
  }
} finally {
  consumer.close();
}

책은 poll 호출을 두고 "이 장에서 가장 중요한 줄"이라고 말한다. 상어가 계속 헤엄쳐야 살듯이, 컨슈머는 계속 Kafka를 poll해야 한다. 그렇지 않으면 죽은 것으로 간주되어 그 컨슈머가 소비하던 파티션이 그룹 내 다른 컨슈머에게 넘어간다.

poll()에 넘기는 파라미터는 타임아웃 간격이며, 컨슈머 버퍼에 데이터가 없을 때 poll()이 얼마나 블로킹할지를 제어한다. 0으로 두면 즉시 반환하고, 그렇지 않으면 지정한 밀리초만큼 브로커에서 데이터가 오기를 기다린다. 이 값은 폴링하는 스레드에게 얼마나 빨리 제어를 돌려주고 싶은지 같은 애플리케이션 요구에 따라 정한다.

poll()은 레코드 목록을 반환하며, 각 레코드는 자신이 온 토픽과 파티션, 파티션 내 오프셋, 그리고 키와 값을 담고 있다.

poll 루프는 데이터를 가져오는 것 이상의 일을 한다. 새 컨슈머로 poll()을 처음 호출할 때, GroupCoordinator를 찾고 컨슈머 그룹에 들어가고 파티션 할당을 받는 일이 여기서 일어난다. 리밸런스가 발생하면 그 처리도 poll 루프 안에서 이루어진다. 컨슈머를 살아 있게 하는 하트비트도 poll 루프 안에서 전송된다. 그래서 반복 사이의 처리는 빠르고 효율적이어야 한다.

루프를 빠져나갈 때는 반드시 close()를 호출한다. 네트워크 연결과 소켓을 닫을 뿐 아니라, 그룹 코디네이터가 하트비트 부재로 컨슈머의 죽음을 알아채기를 기다리지 않고 즉시 리밸런스를 유발하기 때문이다.

스레드 안전성 규칙은 단순하다. 같은 그룹에 속한 여러 컨슈머를 한 스레드에서 돌릴 수 없고, 여러 스레드가 같은 컨슈머를 안전하게 쓸 수도 없다. 컨슈머 하나당 스레드 하나가 규칙이다. 한 애플리케이션에서 같은 그룹의 컨슈머를 여러 개 돌리려면 각각을 자기 스레드에서 실행해야 하며, 컨슈머 로직을 객체로 감싸고 Java의 ExecutorService로 여러 스레드를 띄우는 방식이 유용하다.

페치 크기와 지연을 제어하는 설정

fetch.min.bytes는 컨슈머가 브로커에서 레코드를 페치할 때 받고 싶은 최소 데이터 양이다. 브로커가 레코드 요청을 받았는데 새 레코드가 이 값보다 적으면, 브로커는 더 많은 메시지가 쌓일 때까지 기다렸다가 응답한다. 토픽 활동이 적을 때(또는 하루 중 한산한 시간대에) 오가는 메시지 수를 줄여 컨슈머와 브로커 양쪽의 부하를 낮춘다. 데이터가 별로 없는데 컨슈머가 CPU를 너무 많이 쓰거나, 컨슈머 수가 많아 브로커 부하를 줄이고 싶을 때 기본값보다 높인다.

fetch.max.wait.ms는 그 기다림의 상한이다. 기본적으로 Kafka는 최대 500ms를 기다린다. 토픽에 데이터가 충분히 흐르지 않아 최소 데이터 양을 못 채우면 최대 500ms의 추가 지연이 생긴다는 뜻이다. 지연 상한을 제한하고 싶으면(보통 애플리케이션 SLA 때문에) 이 값을 낮춘다. fetch.max.wait.ms를 100ms, fetch.min.bytes를 1MB로 두면, Kafka는 반환할 데이터가 1MB 모이거나 100ms가 지나거나 둘 중 먼저 오는 시점에 응답한다.

max.partition.fetch.bytes는 서버가 파티션당 반환할 최대 바이트 수이며 기본값은 1MB다. KafkaConsumer.poll()이 ConsumerRecords를 반환할 때, 컨슈머에 할당된 파티션마다 최대 이만큼을 쓴다. 그래서 토픽에 파티션이 20개이고 컨슈머가 5개라면 각 컨슈머는 ConsumerRecords를 위해 4MB의 메모리가 필요하다. 실제로는 그룹 내 다른 컨슈머가 죽으면 각 컨슈머가 더 많은 파티션을 다뤄야 하므로 메모리를 더 잡아 두어야 한다.

이 값에는 두 가지 함정이 있다. 첫째, max.partition.fetch.bytes는 브로커가 허용하는 가장 큰 메시지보다 커야 한다. 그렇지 않으면 브로커에 컨슈머가 소비할 수 없는 메시지가 생기고, 컨슈머는 그 메시지를 읽으려다 멈춰 버린다(hang). 둘째, poll() 한 번이 반환하는 데이터가 너무 크면 처리에 시간이 오래 걸려 다음 poll 루프 반복에 제때 도달하지 못하고 세션 타임아웃이 날 수 있다. 이런 일이 생기면 max.partition.fetch.bytes를 낮추거나 세션 타임아웃을 늘려야 한다.

max.poll.records는 poll() 한 번이 반환할 최대 레코드 수를 제어한다. 폴링 루프에서 애플리케이션이 처리해야 할 데이터 양을 조절하는 데 유용하다.

receive.buffer.bytes와 send.buffer.bytes는 프로듀서와 동일하게 소켓의 TCP 버퍼 크기이며 -1이면 OS 기본값을 쓴다. 다른 데이터센터의 브로커와 통신할 때 키우는 것이 좋다.

세션 타임아웃, 오프셋 리셋, 할당 전략

session.timeout.ms는 컨슈머가 브로커와 연락이 끊긴 상태로도 살아 있다고 간주되는 시간이며, 이 판본 기준 기본값은 3초다. 이 시간이 지나도록 그룹 코디네이터에게 하트비트를 보내지 않으면 죽은 것으로 간주되고 코디네이터가 리밸런스를 유발한다.

이 설정은 heartbeat.interval.ms와 밀접하게 엮인다. heartbeat.interval.ms는 KafkaConsumer의 poll()이 얼마나 자주 하트비트를 보낼지, session.timeout.ms는 하트비트 없이 얼마나 오래 버틸 수 있는지를 제어한다. 그래서 보통 함께 조정하며, heartbeat.interval.ms는 session.timeout.ms보다 작아야 하고 대개 타임아웃의 3분의 1로 잡는다. session.timeout.ms가 3초면 heartbeat.interval.ms는 1초다.

session.timeout.ms를 기본값보다 낮추면 장애를 더 빨리 감지하고 복구할 수 있지만, 컨슈머가 poll 루프를 마치는 데 오래 걸리거나 GC가 길어질 때 원치 않는 리밸런스가 발생할 수 있다. 반대로 높이면 우발적 리밸런스 가능성은 줄지만 실제 장애 감지가 늦어진다.

auto.offset.reset은 컨슈머가 커밋된 오프셋이 없는 파티션을 읽기 시작할 때, 또는 갖고 있는 커밋 오프셋이 유효하지 않을 때(보통 컨슈머가 너무 오래 죽어 있어서 그 오프셋의 레코드가 브로커에서 이미 만료된 경우)의 동작을 제어한다. 기본값은 latest로, 유효한 오프셋이 없으면 가장 최신 레코드부터(즉 컨슈머가 실행을 시작한 뒤에 쓰인 레코드부터) 읽는다. 대안은 earliest로, 파티션의 맨 처음부터 모든 데이터를 읽는다.

enable.auto.commit은 컨슈머가 오프셋을 자동으로 커밋할지를 제어하며 기본값은 true다. 커밋 시점을 직접 제어하고 싶다면(중복을 최소화하고 데이터 누락을 피하려면 필요하다) false로 둔다. true로 둔다면 auto.commit.interval.ms로 커밋 빈도를 조절할 수 있다.

partition.assignment.strategy는 어떤 파티션을 어떤 컨슈머에 할당할지 결정하는 PartitionAssignor를 지정한다. 이 판본 기준 Kafka에는 두 가지 기본 전략이 있다.

Range는 각 컨슈머에게, 구독한 각 토픽에서 연속된 파티션 부분집합을 할당한다. 컨슈머 C1, C2가 토픽 T1, T2를 구독하고 각 토픽에 파티션이 3개라면, C1은 T1과 T2의 파티션 0과 1을, C2는 두 토픽의 파티션 2를 받는다. 각 토픽의 파티션 수가 홀수이고 할당이 토픽별로 독립적으로 이루어지기 때문에 첫 번째 컨슈머가 더 많은 파티션을 갖게 된다. 컨슈머 수가 각 토픽의 파티션 수를 깔끔하게 나누지 못할 때마다 이 불균형이 발생한다.

RoundRobin은 구독한 모든 토픽의 모든 파티션을 모아 컨슈머들에게 하나씩 순차적으로 할당한다. 위 예에서 C1은 T1의 파티션 0과 2, T2의 파티션 1을 받고, C2는 T1의 파티션 1과 T2의 파티션 0, 2를 받는다. 모든 컨슈머가 같은 토픽들을 구독하는 흔한 상황에서 RoundRobin은 컨슈머들이 같은 수의 파티션(많아야 1개 차이)을 갖도록 만든다.

기본값은 org.apache.kafka.clients.consumer.RangeAssignor이며, org.apache.kafka.clients.consumer.RoundRobinAssignor로 교체하거나 직접 구현한 전략의 클래스 이름을 지정할 수도 있다.