Kafka

CLI 도구를 쓰기 전에 알아야 할 두 가지

Kafka는 클러스터에 관리 변경을 가하는 데 유용한 여러 CLI 유틸리티를 제공한다. 도구들은 Java 클래스로 구현되어 있고, 그 클래스를 올바르게 호출해 주는 스크립트 집합이 함께 제공된다. 이 도구들은 기본 기능을 제공하지만, 더 복잡한 작업에는 부족하다고 느낄 수 있다.

첫 번째 주의점은 권한이다. Apache Kafka는 토픽 조작을 통제하기 위한 인증·인가를 구현하고 있지만, 대부분의 클러스터 조작은 아직 지원되지 않는다. 즉 이 CLI 도구들은 아무런 인증 없이 사용할 수 있으며, 토픽 변경 같은 작업이 보안 검사나 감사 없이 실행될 수 있다는 뜻이다. 이 기능은 개발 중이었다.

두 번째 주의점은 버전이다. Kafka의 많은 명령행 도구는 브로커에 접속하는 대신 Zookeeper에 저장된 메타데이터를 직접 조작한다. 그래서 사용하는 도구의 버전이 클러스터 브로커의 버전과 일치하는지 확인하는 것이 중요하다. 가장 안전한 방법은 Kafka 브로커 자체에서, 배포된 버전의 도구를 실행하는 것이다.

토픽 조작은 kafka-topics.sh가 담당한다(설정 변경 기능은 deprecated되어 kafka-configs.sh로 옮겨졌다). 이 명령을 쓰려면 --zookeeper 인자로 클러스터의 Zookeeper 연결 문자열을 제공해야 한다. 이후 예시에서 Zookeeper 연결 문자열은 zoo1.example.com:2181/kafka-cluster로 가정한다.

토픽 생성과 이름 규칙

새 토픽을 만들려면 세 가지 인자가 필요하다. 일부는 브로커 레벨 기본값이 이미 설정되어 있더라도 반드시 제공해야 한다.

- 토픽 이름: 만들려는 토픽의 이름 - Replication Factor: 클러스터 안에서 유지할 토픽 복제본의 수 - Partitions: 토픽에 만들 파티션의 수

kafka-topics.sh --zookeeper <zookeeper connect> --create --topic <string> \
  --replication-factor <integer> --partitions <integer>

이 명령은 지정한 이름과 파티션 수로 토픽을 만들게 한다. 각 파티션마다 클러스터가 지정된 수의 복제본을 적절히 선택한다. 즉 클러스터가 rack-aware 복제본 할당으로 설정되어 있다면 각 파티션의 복제본이 서로 다른 rack에 놓인다. rack-aware 할당을 원하지 않으면 --disable-rack-aware 인자를 준다.

# kafka-topics.sh --zookeeper zoo1.example.com:2181/kafka-cluster --create \
  --topic my-topic --replication-factor 2 --partitions 8
Created topic "my-topic".

자동화 스크립트에서 이 명령을 쓸 때는 --if-not-exists 인자를 쓰면 토픽이 이미 있어도 오류를 반환하지 않는다.

토픽 이름에는 영숫자, 밑줄, 대시, 마침표를 쓸 수 있다. 두 가지 권고가 있다. 첫째, 밑줄 두 개로 시작하는 이름은 허용되지만 권장되지 않는다. 이런 형태는 클러스터의 내부 토픽으로 간주된다(컨슈머 그룹 오프셋 저장용 __consumer_offsets가 그 예다). 둘째, 한 클러스터에서 마침표와 밑줄을 함께 쓰는 것도 권장되지 않는다. 토픽 이름이 Kafka 내부의 메트릭 이름에 쓰일 때 마침표가 밑줄로 바뀌기 때문이다(예: "topic.1"은 메트릭에서 "topic_1"이 된다). 즉 topic.1과 topic_1이 같은 클러스터에 있으면 메트릭에서 구분되지 않는다.

생성 시 복제본을 명시적으로 지정하거나 토픽 설정 오버라이드를 --config 파라미터로 함께 줄 수도 있다.

파티션 추가는 되지만 감소는 안 된다

토픽의 파티션 수를 늘려야 할 때가 있다. 파티션은 토픽을 클러스터에 걸쳐 확장·복제하는 수단이고, 파티션 수를 늘리는 가장 흔한 이유는 토픽을 더 넓게 퍼뜨리거나 단일 파티션의 처리량을 낮추기 위해서다. 컨슈머가 한 그룹 안에서 더 많은 복사본을 돌려야 할 때도 파티션을 늘린다. 하나의 파티션은 그룹 안의 단 한 멤버만 소비할 수 있기 때문이다.

# kafka-topics.sh --zookeeper zoo1.example.com:2181/kafka-cluster \
  --alter --topic my-topic --partitions 16
WARNING: If partitions are increased for a topic that has a key,
the partition logic or ordering of the messages will be affected
Adding partitions succeeded!

키가 있는 메시지를 생산하는 토픽은 컨슈머 관점에서 파티션 추가가 매우 곤란하다. 파티션 수가 바뀌면 키에서 파티션으로의 매핑이 바뀌기 때문이다. 그래서 키 있는 메시지를 담을 토픽은 생성 시점에 파티션 수를 한 번 정하고 이후 크기를 바꾸지 않는 것이 좋다.

--alter 명령에도 --if-exists 인자가 제공되지만 사용은 권장되지 않는다. 이 인자를 쓰면 변경하려는 토픽이 존재하지 않아도 오류를 반환하지 않아, 만들어졌어야 할 토픽이 없는 문제를 가릴 수 있다.

파티션 수는 줄일 수 없다. 지원되지 않는 이유는, 토픽에서 파티션을 삭제하면 그 토픽 데이터의 일부도 함께 삭제되어 클라이언트 관점에서 일관성이 깨지기 때문이다. 게다가 데이터를 남은 파티션들로 재분배하려 해도 어렵고 메시지 순서가 뒤섞인다. 파티션 수를 줄여야 한다면 토픽을 삭제하고 다시 만들어야 한다.

토픽 삭제, 목록 조회, describe와 진단 필터

메시지가 하나도 없는 토픽도 디스크 공간, 열린 파일 핸들, 메모리 같은 클러스터 자원을 쓴다. 더 이상 필요 없는 토픽은 삭제해 자원을 회수할 수 있다. 다만 이 작업을 하려면 클러스터의 브로커가 delete.topic.enable 옵션을 true로 설정한 상태여야 한다. false로 되어 있으면 토픽 삭제 요청은 무시된다. 토픽 삭제는 그 안의 모든 메시지도 삭제하며, 되돌릴 수 없다.

# kafka-topics.sh --zookeeper zoo1.example.com:2181/kafka-cluster \
  --delete --topic my-topic
Topic my-topic is marked for deletion.
Note: This will have no impact if delete.topic.enable is not set to true.

--list는 클러스터의 모든 토픽을 한 줄에 하나씩, 특별한 순서 없이 출력한다. 삭제 표시된 토픽은 "my-topic - marked for deletion"처럼 나온다.

--describe는 하나 이상의 토픽에 대한 상세 정보를 준다. 출력에는 파티션 수, 토픽 설정 오버라이드, 그리고 각 파티션과 그 복제본 할당 목록이 포함된다. --topic 인자로 특정 토픽 하나로 한정할 수 있다.

describe에는 출력을 걸러 주는 유용한 옵션들이 있고, 클러스터 문제 진단에 쓰인다. 이 옵션들을 쓸 때는 --topic 인자를 주지 않는다(클러스터 전체에서 조건에 맞는 토픽·파티션을 찾는 것이 목적이기 때문이다). 또한 이 옵션들은 --list 명령과는 동작하지 않는다.

- --topics-with-overrides: 클러스터 기본값과 다른 설정을 가진 토픽만 describe한다. - --under-replicated-partitions: 복제본 중 하나 이상이 리더와 in-sync가 아닌 모든 파티션을 보여 준다. - --unavailable-partitions: 리더가 없는 모든 파티션을 보여 준다. 이는 더 심각한 상황으로, 그 파티션이 현재 오프라인이며 produce·consume 클라이언트가 사용할 수 없다는 뜻이다.

# kafka-topics.sh --zookeeper zoo1.example.com:2181/kafka-cluster \
  --describe --under-replicated-partitions
 Topic: other-topic Partition: 2 Leader: 0 Replicas: 1,0 Isr: 0
 Topic: other-topic Partition: 4 Leader: 0 Replicas: 1,0 Isr: 0

컨슈머 그룹 조회와 랙 읽기

Kafka의 컨슈머 그룹은 두 곳에서 관리된다. 구버전 컨슈머의 정보는 Zookeeper에, 새 컨슈머의 정보는 Kafka 브로커 안에 유지된다. kafka-consumer-groups.sh 도구는 두 종류를 모두 나열하고 describe할 수 있다. 다만 컨슈머 그룹과 오프셋 정보를 삭제하는 것은 구버전 컨슈머(Zookeeper 관리) 그룹에 대해서만 가능하다.

구버전 컨슈머 그룹을 다룰 때는 --zookeeper 파라미터로 클러스터에 접근한다. 새 컨슈머 그룹은 대신 --bootstrap-server 파라미터에 접속할 Kafka 브로커의 호스트명과 포트를 준다.

# 구버전 그룹 목록
kafka-consumer-groups.sh --zookeeper zoo1.example.com:2181/kafka-cluster --list

# 새 컨슈머 그룹 목록
kafka-consumer-groups.sh --new-consumer \
  --bootstrap-server kafka1.example.com:9092/kafka-cluster --list

--list를 --describe로 바꾸고 --group 파라미터를 더하면 상세 정보를 볼 수 있다. 그 그룹이 소비 중인 모든 토픽과 각 토픽 파티션의 오프셋이 나온다. 출력 필드는 다음과 같다.

- GROUP: 컨슈머 그룹 이름 - TOPIC: 소비 중인 토픽 이름 - PARTITION: 소비 중인 파티션의 ID 번호 - CURRENT-OFFSET: 이 토픽 파티션에 대해 컨슈머 그룹이 마지막으로 커밋한 오프셋. 파티션 안에서 컨슈머의 위치다 - LOG-END-OFFSET: 이 토픽 파티션에 대한 브로커의 현재 high-water mark 오프셋. 클러스터에 생산되어 커밋된 마지막 메시지의 오프셋이다 - LAG: 이 토픽 파티션에 대한 CURRENT-OFFSET과 LOG-END-OFFSET의 차이 - OWNER: 현재 이 토픽 파티션을 소비 중인 그룹 멤버. 그룹 멤버가 제공하는 임의의 ID이며 반드시 컨슈머의 호스트명을 포함하지는 않는다

그룹 삭제는 구버전 컨슈머 클라이언트에서만 지원된다. 삭제하면 그룹 전체가 Zookeeper에서 제거되며 그 그룹이 소비하는 모든 토픽의 저장된 오프셋도 함께 사라진다. 이 작업을 하려면 그룹의 모든 컨슈머를 먼저 종료해야 한다. 그러지 않으면 사용 중인 Zookeeper 메타데이터가 제거되어 컨슈머가 정의되지 않은 동작을 보일 수 있다.

같은 명령에 --topic을 더하면 그룹 전체를 지우지 않고 그 그룹이 소비하는 특정 토픽 하나의 오프셋만 삭제할 수도 있다. 이때도 컨슈머 그룹을 멈추거나 해당 토픽을 소비하지 않도록 설정한 뒤 수행하는 것이 권장된다.

오프셋 내보내기와 들여오기 (구버전 컨슈머 한정)

구버전 컨슈머 클라이언트를 쓸 때는 오프셋을 보고 삭제하는 것 외에, 오프셋을 일괄로 가져오고 새 오프셋을 일괄로 저장하는 것도 가능하다. 문제가 생겨 메시지를 다시 읽어야 하거나, 컨슈머가 처리하지 못하는 잘못된 형식의 메시지를 건너뛰기 위해 오프셋을 앞으로 밀어야 할 때 유용하다.

중요한 제약이 있다. 오프셋을 Kafka에 커밋하는 컨슈머 클라이언트의 오프셋을 관리해 주는 도구는 (이 책 시점에) 존재하지 않는다. 이 기능은 Zookeeper에 커밋하는 컨슈머에만 제공된다. Kafka에 커밋하는 그룹의 오프셋을 관리하려면 클라이언트에서 제공하는 API로 그룹의 오프셋을 커밋해야 한다.

오프셋 내보내기 전용 스크립트는 없고, kafka-run-class.sh로 해당 Java 클래스를 적절한 환경에서 실행한다. 내보내면 그룹의 각 토픽 파티션과 그 오프셋이 import 도구가 읽을 수 있는 정해진 형식으로 파일에 저장된다. 파일은 한 줄에 토픽 파티션 하나씩, /consumers/GROUPNAME/offsets/topic/TOPICNAME/PARTITIONID:OFFSET 형식이다.

# kafka-run-class.sh kafka.tools.ExportZkOffsets \
  --zkconnect zoo1.example.com:2181/kafka-cluster --group testgroup \
  --output-file offsets
# cat offsets
/consumers/testgroup/offsets/my-topic/0:8905
/consumers/testgroup/offsets/my-topic/1:8915
/consumers/testgroup/offsets/my-topic/2:9845

들여오기는 그 반대다. 내보낸 파일을 받아 컨슈머 그룹의 현재 오프셋으로 설정한다. 흔한 방식은 현재 오프셋을 내보내고, 백업용으로 파일을 복사해 둔 뒤, 사본을 편집해 원하는 값으로 바꾸는 것이다. import 명령에서는 --group 옵션을 쓰지 않는다는 점에 주의한다. 컨슈머 그룹 이름이 들여올 파일 안에 이미 들어 있기 때문이다.

# kafka-run-class.sh kafka.tools.ImportZkOffsets \
  --zkconnect zoo1.example.com:2181/kafka-cluster --input-file offsets

이 단계를 수행하기 전에 그룹의 모든 컨슈머를 멈추는 것이 중요하다. 컨슈머 그룹이 활성 상태일 때 새 오프셋을 쓰면 컨슈머는 그 값을 읽지 않고 그냥 덮어써 버린다.