Kafka

사전 준비: OS, Java, Zookeeper

Apache Kafka는 Java 애플리케이션이라 Windows, MacOS, Linux 등 여러 OS에서 동작한다. 그러나 가장 흔하게 설치되고 일반적인 배포에 권장되는 OS는 Linux다.

Zookeeper든 Kafka든 설치하기 전에 Java 환경이 필요하다. 이 책의 기준은 Java 8이며, OS가 제공하는 버전이나 java.com에서 직접 받은 버전 모두 가능하다. Zookeeper와 Kafka는 런타임 에디션으로도 동작하지만, 도구와 애플리케이션을 개발할 때는 전체 JDK가 편리하다. 설치 예제는 JDK 8 update 51을 /usr/java/jdk1.8.0_51에 설치했다고 가정한다.

Kafka는 Kafka 클러스터에 대한 메타데이터와 컨슈머 클라이언트 정보를 저장하는 데 Zookeeper를 사용한다. Kafka 배포판에 포함된 스크립트로 Zookeeper 서버를 띄울 수도 있지만, 정식 배포판을 설치하는 것이 어렵지 않다. Kafka는 Zookeeper 3.4.6 안정 릴리스로 광범위하게 테스트되었다.

독립 실행 서버 설정은 간단하다. /usr/local/zookeeper에 설치하고 데이터를 /var/lib/zookeeper에 저장하는 예는 다음과 같다.

tar -zxf zookeeper-3.4.6.tar.gz
mv zookeeper-3.4.6 /usr/local/zookeeper
mkdir -p /var/lib/zookeeper
cat > /usr/local/zookeeper/conf/zoo.cfg << EOF
tickTime=2000
dataDir=/var/lib/zookeeper
clientPort=2181
EOF
export JAVA_HOME=/usr/java/jdk1.8.0_51
/usr/local/zookeeper/bin/zkServer.sh start

정상 동작 여부는 클라이언트 포트에 접속해 네 글자 명령 srvr을 보내 확인할 수 있다. 응답에 Mode: standalone 같은 정보가 나온다.

Zookeeper 앙상블 구성

Zookeeper 클러스터는 앙상블(ensemble)이라고 부른다. 사용하는 알고리즘 특성상 앙상블은 홀수 개의 서버(3, 5 등)로 구성하는 것이 권장되는데, Zookeeper가 요청에 응답하려면 앙상블 멤버의 과반(쿼럼)이 살아 있어야 하기 때문이다. 3노드 앙상블은 1대가 빠져도 동작하고, 5노드 앙상블은 2대가 빠져도 동작한다.

앙상블 크기를 정할 때는 5노드를 고려하는 것이 좋다. 노드를 교체하는 것을 포함해 앙상블 설정을 변경하려면 노드를 한 번에 하나씩 재기동해야 하는데, 앙상블이 1대 이상 다운을 견딜 수 없다면 유지보수 작업 자체가 추가 위험을 만들기 때문이다. 반대로 7노드를 넘기는 것은 권장되지 않는다. 합의 프로토콜의 특성상 그 이상에서는 성능이 저하되기 시작한다.

앙상블을 구성하려면 모든 서버를 나열한 공통 설정 파일이 필요하고, 각 서버는 자신의 ID 번호를 지정하는 myid 파일을 데이터 디렉터리에 가져야 한다. 호스트명이 zoo1/zoo2/zoo3.example.com이라면 설정은 다음과 같다.

tickTime=2000
dataDir=/var/lib/zookeeper
clientPort=2181
initLimit=20
syncLimit=5
server.1=zoo1.example.com:2888:3888
server.2=zoo2.example.com:2888:3888
server.3=zoo3.example.com:2888:3888

initLimit은 팔로워가 리더에 접속하는 데 허용되는 시간이고, syncLimit은 팔로워가 리더로부터 얼마나 out-of-sync 상태여도 되는지를 제한한다. 두 값 모두 tickTime 단위의 개수이므로, initLimit은 20 × 2000ms = 40초가 된다.

server.X=hostname:peerPort:leaderPort 형식의 각 항목은 다음을 의미한다.

- X: 서버의 ID 번호. 정수여야 하지만 0부터 시작할 필요도, 연속일 필요도 없다 - hostname: 서버의 호스트명 또는 IP 주소 - peerPort: 앙상블의 서버들이 서로 통신하는 TCP 포트 - leaderPort: 리더 선출이 수행되는 TCP 포트

클라이언트는 clientPort로만 앙상블에 접속할 수 있으면 되지만, 앙상블 멤버들끼리는 세 포트 모두로 통신할 수 있어야 한다. 공통 설정 파일에 더해 각 서버의 dataDir에는 myid 파일이 있어야 하며, 그 안의 ID 번호는 설정 파일의 값과 일치해야 한다.

브로커 설치와 동작 검증

Java와 Zookeeper가 준비되면 Kafka를 설치한다. 이 책 집필 시점의 버전은 Scala 2.11.0 기반의 0.9.0.1이다. 다음 예는 Kafka를 /usr/local/kafka에 설치하고, 앞서 띄운 Zookeeper를 사용하며, 메시지 로그 세그먼트를 /tmp/kafka-logs에 저장하도록 구성한다.

tar -zxf kafka_2.11-0.9.0.1.tgz
mv kafka_2.11-0.9.0.1 /usr/local/kafka
mkdir /tmp/kafka-logs
export JAVA_HOME=/usr/java/jdk1.8.0_51
/usr/local/kafka/bin/kafka-server-start.sh -daemon /usr/local/kafka/config/server.properties

브로커가 뜨면 테스트 토픽을 만들고 메시지를 생성·소비해 동작을 확인한다.

# 토픽 생성과 확인
bin/kafka-topics.sh --create --zookeeper localhost:2181 \
  --replication-factor 1 --partitions 1 --topic test
bin/kafka-topics.sh --zookeeper localhost:2181 --describe --topic test
# Topic:test  PartitionCount:1  ReplicationFactor:1  Configs:
#     Topic: test  Partition: 0  Leader: 0  Replicas: 0  Isr: 0

# 메시지 생성 (브로커의 9092로 접속)
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

# 메시지 소비
bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning

토픽 관리 명령은 Zookeeper 주소(2181)를, 콘솔 프로듀서는 브로커 주소(9092)를 사용한다는 점에 주목할 만하다. describe 출력의 Leader/Replicas/Isr는 각각 파티션 리더 브로커, 복제본이 배치된 브로커 목록, in-sync 상태인 복제본 목록을 뜻한다.

반드시 검토해야 할 일반 브로커 설정

배포판에 포함된 예제 설정은 개념 증명용 단독 서버에는 충분하지만 대부분의 설치에는 부족하다. 단일 서버의 단독 브로커가 아닌 환경에서는 다음 설정들을 반드시 검토해야 한다.

broker.id — 모든 Kafka 브로커는 정수 식별자를 가져야 하며 broker.id로 설정한다. 기본값은 0이고 어떤 값이든 가능하지만, 단일 Kafka 클러스터 안에서 유일해야 한다는 것이 가장 중요하다. 선택은 임의지만, 유지보수 시 브로커 ID를 호스트에 대응시키기 쉽도록 호스트에 내재된 값을 쓰는 것이 좋은 가이드라인이다. 예를 들어 호스트명이 host1.example.com, host2.example.com처럼 고유 번호를 포함한다면 그 번호가 좋은 선택이다.

port — 예제 설정은 TCP 9092에서 리스너를 띄운다. port 파라미터로 아무 포트나 지정할 수 있지만, 1024 미만 포트를 고르면 Kafka를 root로 실행해야 한다. Kafka를 root로 실행하는 것은 권장되는 구성이 아니다.

zookeeper.connect — 브로커 메타데이터를 저장할 Zookeeper 위치를 지정한다. 형식은 hostname:port/path 문자열을 세미콜론으로 구분한 목록이다. 여기서 /path는 Kafka 클러스터를 위한 chroot 환경으로 쓸 선택적 Zookeeper 경로이며, 생략하면 루트 경로가 쓰인다. chroot 경로가 지정되었는데 존재하지 않으면 브로커가 기동할 때 생성한다.

chroot 경로를 쓰는 것은 일반적으로 좋은 관행으로 여겨진다. 다른 Kafka 클러스터를 포함한 다른 애플리케이션들과 Zookeeper 앙상블을 충돌 없이 공유할 수 있기 때문이다. 또한 이 설정에는 (같은 앙상블에 속한) 여러 Zookeeper 서버를 지정하는 것이 좋다. 그래야 서버 장애 시 브로커가 앙상블의 다른 멤버에 접속할 수 있다.

log.dirs — Kafka는 모든 메시지를 디스크에 영속화하며, 이 로그 세그먼트는 log.dirs에 지정된 디렉터리들에 저장된다. 로컬 시스템 경로를 쉼표로 구분한 목록이다. 경로가 둘 이상이면 브로커는 'least-used' 방식으로 파티션을 저장하는데, 한 파티션의 로그 세그먼트는 같은 경로 안에 저장된다. 여기서 주의할 점은 브로커가 새 파티션을 배치할 때 보는 기준이 디스크 사용량이 아니라 '현재 그 경로에 저장된 파티션 개수'라는 것이다. 파티션 수가 가장 적은 경로에 새 파티션이 놓인다.

num.recovery.threads.per.data.dir — Kafka는 로그 세그먼트를 다루기 위해 설정 가능한 스레드 풀을 사용한다. 이 풀은 (1) 정상 기동 시 각 파티션의 로그 세그먼트를 여는 데, (2) 장애 후 기동 시 각 파티션의 로그 세그먼트를 검사하고 절단하는 데, (3) 종료 시 로그 세그먼트를 정상적으로 닫는 데 쓰인다. 기본값은 로그 디렉터리당 1개다. 이 스레드는 기동과 종료 때만 쓰이므로 병렬화를 위해 더 큰 값을 주는 것이 합리적이다. 특히 비정상 종료에서 복구할 때, 파티션이 많은 브로커를 재시작하는 데 걸리는 시간이 몇 시간씩 차이 날 수 있다. 이 값이 log.dirs에 지정된 로그 디렉터리 '당' 값이라는 점을 기억해야 한다. 8로 설정하고 log.dirs에 경로가 3개면 총 24개 스레드가 된다.

auto.create.topics.enable — 기본 설정에서 브로커는 (1) 프로듀서가 토픽에 메시지를 쓰기 시작할 때, (2) 컨슈머가 토픽에서 메시지를 읽기 시작할 때, (3) 어떤 클라이언트든 그 토픽의 메타데이터를 요청할 때 자동으로 토픽을 생성한다. 많은 상황에서 이는 바람직하지 않은 동작인데, 특히 Kafka 프로토콜에서 토픽 생성을 유발하지 않고 존재 여부만 확인할 방법이 없기 때문이다. 토픽 생성을 수동으로든 프로비저닝 시스템으로든 명시적으로 관리한다면 이 값을 false로 두는 것이 좋다.

토픽 기본값 (1): 파티션 수 정하기

Kafka 서버 설정에는 새로 생성되는 토픽에 적용될 여러 기본값이 있다. 파티션 개수나 메시지 리텐션 같은 항목은 관리 도구로 토픽별 설정이 가능하다. 서버 설정의 기본값은 클러스터 내 대다수 토픽에 적합한 baseline 값으로 잡아야 한다. 참고로 예전 버전에서 쓰이던 log.retention.hours.per.topic, log.retention.bytes.per.topic, log.segment.bytes.per.topic 같은 브로커 설정의 토픽별 오버라이드는 더 이상 지원되지 않으며, 오버라이드는 관리 도구로 지정해야 한다.

num.partitions는 새 토픽이 몇 개의 파티션으로 생성될지를 정하며, 주로 자동 토픽 생성이 켜져 있을 때(기본 설정) 의미가 있다. 기본값은 파티션 1개다. 반드시 기억할 점은 토픽의 파티션 수는 늘릴 수만 있고 줄일 수는 없다는 것이다. 따라서 num.partitions보다 적은 파티션이 필요한 토픽이라면 수동으로 생성해야 한다.

파티션은 클러스터 안에서 토픽을 확장하는 수단이므로, 브로커가 추가될 때 메시지 부하가 클러스터 전체에 고르게 퍼지도록 파티션 수를 정하는 것이 중요하다. 많은 사용자가 토픽의 파티션 수를 클러스터의 브로커 수와 같게 하거나 그 배수로 잡는다. 그러면 파티션이 브로커에 고르게 분배되어 메시지 부하도 고르게 분산된다. 다만 필수는 아니며, 토픽을 여러 개 두는 방식으로도 부하를 분산할 수 있다.

파티션 수를 정할 때 고려할 요소는 다음과 같다.

- 이 토픽에서 기대하는 처리량은 얼마인가? 초당 100KB인가 1GB인가? - 단일 파티션에서 소비할 때 기대하는 최대 처리량은 얼마인가? 파티션 하나는 최대 한 컨슈머만 읽으므로, 데이터를 DB에 쓰는 느린 컨슈머가 스레드당 50MB/s를 넘기지 못한다면 파티션 하나에서의 소비 처리량이 그만큼으로 제한된다 - 프로듀서 쪽도 같은 방식으로 추정할 수 있지만, 프로듀서는 보통 컨슈머보다 훨씬 빠르므로 대개 건너뛰어도 안전하다 - 키를 기준으로 파티션에 메시지를 보낸다면 나중에 파티션을 추가하는 것이 매우 어려워질 수 있으므로, 현재 사용량이 아니라 예상되는 미래 사용량을 기준으로 계산한다 - 브로커당 배치할 파티션 수와 브로커별 가용 디스크 공간, 네트워크 대역폭을 고려한다 - 과대 추정을 피한다. 파티션 각각이 브로커의 메모리와 자원을 소모하고 리더 선출 시간을 늘린다

결론적으로 파티션은 많되 너무 많지는 않아야 한다. 토픽의 목표 처리량과 컨슈머의 예상 처리량을 알고 있다면 목표 처리량을 컨슈머 처리량으로 나눠 파티션 수를 도출할 수 있다. 예를 들어 토픽에서 1GB/s를 읽고 쓰고 싶고 컨슈머 하나가 50MB/s만 처리할 수 있다면 최소 20개 파티션이 필요하다. 이렇게 하면 컨슈머 20개가 토픽을 읽어 1GB/s를 달성할 수 있다. 이런 상세 정보가 없다면, 경험상 디스크상의 파티션 크기를 리텐션 하루당 6GB 미만으로 제한하는 것이 대체로 만족스러운 결과를 준다.

토픽 기본값 (2): 리텐션, 세그먼트, 메시지 크기

log.retention.ms — Kafka가 메시지를 얼마나 보관할지를 정하는 가장 흔한 방식은 시간 기준이다. 설정 파일의 기본값은 log.retention.hours로 지정되며 168시간, 즉 1주일이다. log.retention.minutes와 log.retention.ms도 허용된다. 셋 모두 '메시지가 삭제될 수 있게 되기까지의 시간'이라는 같은 설정을 지정하지만, 권장 파라미터는 log.retention.ms다. 둘 이상이 지정되면 더 작은 단위가 우선하기 때문에, log.retention.ms를 쓰면 설정한 값이 항상 실제로 쓰인다.

시간 기준 리텐션은 디스크상 각 로그 세그먼트 파일의 마지막 수정 시각(mtime)을 검사해 수행된다. 정상 운영에서 이 값은 로그 세그먼트가 닫힌 시각이며 파일 내 마지막 메시지의 타임스탬프를 나타낸다. 그러나 관리 도구로 파티션을 브로커 간에 이동하면 이 시각이 정확하지 않게 되어 해당 파티션의 리텐션이 과도해진다.

log.retention.bytes — 보관되는 메시지의 총 바이트 수로 만료시키는 방식이다. 중요한 것은 이 값이 '파티션 단위'로 적용된다는 점이다. 파티션이 8개인 토픽에 log.retention.bytes를 1GB로 설정하면 그 토픽이 보관하는 데이터는 최대 8GB가 된다. 모든 리텐션은 토픽이 아니라 개별 파티션에 대해 수행되므로, 토픽의 파티션 수를 늘리면 크기 기준 리텐션 총량도 함께 늘어난다.

크기와 시간을 모두 설정하면 둘 중 어느 조건이든 충족되는 순간 메시지가 제거될 수 있다. 예를 들어 log.retention.ms가 86400000(1일)이고 log.retention.bytes가 1000000000(1GB)이라면, 하루 동안의 메시지 총량이 1GB를 넘을 경우 하루가 안 된 메시지도 삭제될 수 있다. 반대로 총량이 1GB 미만이라면 파티션 크기가 1GB에 못 미쳐도 하루 뒤 삭제될 수 있다.

log.segment.bytes — 앞의 리텐션 설정들은 개별 메시지가 아니라 로그 세그먼트 단위로 동작한다. 메시지가 브로커에 생성되면 해당 파티션의 현재 로그 세그먼트에 추가되고, 세그먼트가 log.segment.bytes에 지정된 크기(기본 1GB)에 도달하면 닫히고 새 세그먼트가 열린다. 세그먼트는 '닫힌 뒤에야' 만료 대상이 된다. 세그먼트 크기를 작게 하면 파일을 더 자주 닫고 할당해야 하므로 전반적인 디스크 쓰기 효율이 떨어진다.

생성 속도가 낮은 토픽에서는 세그먼트 크기 조정이 중요할 수 있다. 예를 들어 하루 100MB만 받는 토픽에 기본값이 적용되면 세그먼트 하나를 채우는 데 10일이 걸린다. 세그먼트가 닫히기 전에는 메시지를 만료시킬 수 없으므로, log.retention.ms가 604800000(1주)이라면 닫힌 세그먼트가 만료되기까지 실제로는 최대 17일치 메시지가 보관된다. 세그먼트가 10일치 메시지로 닫힌 뒤, 시간 정책에 따라 다시 7일을 보관해야 만료되기 때문이다(세그먼트 안의 마지막 메시지까지 만료 가능해져야 지울 수 있다).

세그먼트 크기는 타임스탬프로 오프셋을 조회하는 동작에도 영향을 준다. 특정 타임스탬프에 대한 파티션 오프셋을 요청하면 Kafka는 그 시각에 쓰이고 있던 로그 세그먼트 파일을 찾는데, 파일의 생성 시각과 마지막 수정 시각을 이용해 해당 타임스탬프 이전에 생성되고 이후에 수정된 파일을 찾는다. 그리고 그 로그 세그먼트의 시작 오프셋(= 파일명)을 응답으로 돌려준다.

log.segment.ms — 로그 세그먼트가 닫히는 시점을 제어하는 또 다른 방법으로, 세그먼트를 닫기까지의 시간을 지정한다. log.segment.bytes와 log.segment.ms는 상호 배타적이지 않으며, Kafka는 크기 한계와 시간 한계 중 먼저 도달한 쪽에서 세그먼트를 닫는다. 기본적으로 log.segment.ms에는 설정값이 없어 크기 기준으로만 세그먼트가 닫힌다. 시간 기준 세그먼트를 쓸 때는 여러 세그먼트가 동시에 닫힐 때의 디스크 성능 영향을 고려해야 한다. 세그먼트 크기 한계에 절대 도달하지 않는 파티션이 많으면, 시간 한계의 시계가 브로커 기동 시점부터 함께 돌기 시작해 저볼륨 파티션들이 항상 같은 시각에 세그먼트를 닫게 되기 때문이다.

message.max.bytes — 브로커가 받아들이는 메시지의 최대 크기를 제한하며 기본값은 1000000, 즉 1MB다. 이보다 큰 메시지를 보내려는 프로듀서는 브로커로부터 에러를 받고 메시지는 거부된다. 브로커에서 지정하는 모든 바이트 크기와 마찬가지로 이 설정은 '압축된' 메시지 크기를 다룬다. 즉 압축 후 이 값 아래로 들어오기만 하면 프로듀서는 훨씬 큰 원본 메시지를 보낼 수 있다.

허용 메시지 크기를 키우면 눈에 띄는 성능 영향이 있다. 메시지가 클수록 네트워크 연결과 요청을 처리하는 브로커 스레드가 요청 하나에 더 오래 매달리고, 디스크 쓰기 크기도 커져 I/O 처리량에 영향을 준다. 또한 브로커의 메시지 크기 설정은 컨슈머 클라이언트의 fetch.message.max.bytes와 반드시 맞춰야 한다. 이 값이 message.max.bytes보다 작으면 더 큰 메시지를 만난 컨슈머는 그 메시지를 가져오지 못해 멈춰 버리고 더 진행하지 못한다. 클러스터로 구성했을 때 브로커의 replica.fetch.max.bytes에도 같은 규칙이 적용된다.