Kafka

저장의 기본 단위와 log.dirs

Kafka의 기본 저장 단위는 파티션 복제본이다. 파티션은 여러 브로커에 걸쳐 쪼갤 수 없고, 같은 브로커의 여러 디스크에 나눌 수도 없다. 따라서 파티션의 크기는 하나의 마운트 포인트에서 쓸 수 있는 공간으로 제한된다. 마운트 포인트는 JBOD 구성이면 디스크 한 장, RAID 구성이면 여러 디스크가 될 수 있다.

관리자는 파티션을 저장할 디렉터리 목록을 `log.dirs` 파라미터로 정의한다(Kafka의 에러 로그 위치를 정하는 log4j.properties의 설정과 혼동하지 말 것). 보통 Kafka가 사용할 마운트 포인트마다 디렉터리를 하나씩 지정한다.

파티션 할당 알고리즘과 랙 인식

토픽을 만들면 Kafka는 먼저 파티션을 브로커에 어떻게 배치할지 결정한다. 브로커 6대에 파티션 10개, 복제 팩터 3인 토픽을 만든다면 30개의 파티션 복제본을 6대에 배분해야 하며, 목표는 세 가지다.

- 복제본을 브로커에 고르게 분산한다(이 예에서는 브로커당 5개) - 각 파티션의 복제본들이 서로 다른 브로커에 놓이게 한다. 파티션 0의 리더가 브로커 2에 있다면 팔로워는 3과 4에 둘 수 있지만 2에 두거나 둘 다 3에 둘 수는 없다 - 브로커에 랙 정보가 있으면(0.10.0 이상) 각 파티션의 복제본을 가능한 한 서로 다른 랙에 배치한다. 랙 전체가 내려가도 파티션 전체가 불가용해지지 않게 하기 위함이다

구체적으로는 임의의 브로커(예: 4번)부터 시작해 라운드로빈으로 리더 위치를 정한다. 파티션 0의 리더는 브로커 4, 파티션 1은 5, 파티션 2는 (브로커가 6대뿐이므로) 0번으로 넘어간다. 그다음 각 파티션마다 리더에서 한 칸씩 증가하는 위치에 복제본을 둔다. 파티션 0의 리더가 4면 첫 팔로워는 5, 두 번째는 0에 놓인다.

랙 인식이 켜져 있으면 브로커를 번호순으로 고르는 대신 랙이 번갈아 나오는 목록을 먼저 만든다. 브로커 0·1·2가 한 랙, 3·4·5가 다른 랙이라면 0,1,2,3,4,5 순서 대신 0,3,1,4,2,5 순서로 배열해 각 브로커 뒤에 다른 랙의 브로커가 오게 한다. 이렇게 하면 파티션 0의 리더가 브로커 4일 때 첫 복제본이 브로커 2, 즉 완전히 다른 랙에 놓인다.

디렉터리 선택 규칙과 디스크 공간의 함정

브로커가 정해지면 새 파티션을 어느 디렉터리에 둘지 결정한다. 규칙은 파티션마다 독립적으로 적용되며 매우 단순하다. 각 디렉터리에 있는 파티션의 개수를 세어 가장 적은 디렉터리에 새 파티션을 추가한다.

이 규칙 때문에 새 디스크를 추가하면 새로 만들어지는 파티션이 전부 그 디스크에 몰린다. 균형이 맞을 때까지는 새 디스크의 파티션 개수가 항상 가장 적기 때문이다.

주의할 점이 있다. 브로커로의 파티션 할당은 가용 공간이나 기존 부하를 고려하지 않고, 디스크로의 할당은 파티션 개수만 볼 뿐 파티션의 크기는 보지 않는다. 따라서 브로커마다 디스크 용량이 다르거나(구형·신형 서버가 섞인 클러스터), 일부 파티션이 비정상적으로 크거나, 한 브로커 안에서 디스크 크기가 서로 다르면 파티션 배치에 신경을 써야 한다.

세그먼트와 액티브 세그먼트

Kafka는 데이터를 영원히 보관하지 않고, 모든 컨슈머가 메시지를 읽을 때까지 기다렸다가 지우지도 않는다. 관리자는 토픽별로 리텐션 기간을 설정한다. 메시지를 얼마 동안 보관할지, 또는 오래된 메시지를 지우기 전까지 얼마나 많은 데이터를 보관할지를 정하는 방식이다.

거대한 파일에서 지워야 할 메시지를 찾아 파일 일부를 삭제하는 것은 느리고 오류가 나기 쉬우므로, Kafka는 각 파티션을 세그먼트로 쪼갠다. 기본적으로 각 세그먼트는 1GB의 데이터 또는 1주일치 데이터 중 먼저 도달하는 쪽을 담는다. 브로커가 파티션에 쓰다가 세그먼트 한도에 도달하면 파일을 닫고 새 파일을 연다.

현재 쓰고 있는 세그먼트를 액티브 세그먼트라 하며 절대 삭제되지 않는다. 그래서 리텐션을 1일로 설정했더라도 세그먼트 하나에 5일치가 담긴다면 실제로는 5일치 데이터를 보관하게 된다. 세그먼트가 닫히기 전에는 그 안의 데이터를 지울 수 없기 때문이다. 반대로 1주일 보관에 매일 새 세그먼트를 롤링한다면, 매일 새 세그먼트를 만들면서 가장 오래된 세그먼트를 지우므로 파티션에는 대체로 7개의 세그먼트가 존재한다.

또한 브로커는 비활성 세그먼트를 포함해 모든 파티션의 모든 세그먼트에 대해 파일 핸들을 열어 둔다. 그래서 열린 파일 핸들 수가 매우 많아지고, OS 설정을 그에 맞게 튜닝해야 한다.

파일 포맷과 래퍼 메시지

각 세그먼트는 하나의 데이터 파일에 저장되고, 파일 안에는 Kafka 메시지와 오프셋이 들어 있다. 디스크에 저장되는 데이터의 형식은 프로듀서가 브로커로 보내는 형식, 그리고 브로커가 컨슈머에게 보내는 형식과 완전히 동일하다. 디스크와 네트워크에서 같은 메시지 포맷을 쓰기 때문에 Kafka는 컨슈머에게 보낼 때 zero-copy 최적화를 쓸 수 있고, 프로듀서가 이미 압축한 메시지를 풀었다가 다시 압축하는 낭비도 피할 수 있다.

각 메시지는 키·값·오프셋 외에 메시지 크기, 손상 감지용 체크섬, 메시지 포맷 버전을 나타내는 매직 바이트, 압축 코덱(Snappy, GZip, LZ4), 타임스탬프(0.10.0에서 추가)를 담는다. 타임스탬프는 설정에 따라 프로듀서가 보낸 시각이거나 브로커에 도착한 시각이다.

프로듀서가 압축을 사용하면 한 프로듀서 배치의 모든 메시지가 함께 압축되어 "래퍼 메시지"의 값으로 전송된다. 브로커는 메시지 하나를 받아 컨슈머에게 그대로 하나로 보내지만, 컨슈머가 그 값의 압축을 풀면 배치에 담겨 있던 메시지들이 각자의 타임스탬프와 오프셋을 가진 채로 나타난다. 그러므로 프로듀서에서 압축을 쓴다면(권장된다) 배치를 크게 보낼수록 네트워크와 브로커 디스크 양쪽에서 압축 효율이 좋아진다.

브로커에는 DumpLogSegment 도구가 함께 제공되어, 파일시스템의 세그먼트를 열어 메시지별 오프셋·체크섬·매직 바이트·크기·압축 코덱을 확인할 수 있다.

bin/kafka-run-class.sh kafka.tools.DumpLogSegments

`--deep-iteration` 파라미터를 주면 래퍼 메시지 안에 압축되어 있는 메시지들의 정보까지 보여준다.

인덱스: 재생성 가능한 보조 자료

컨슈머는 사용 가능한 어떤 오프셋에서든 메시지를 읽기 시작할 수 있다. 오프셋 100부터 1MB를 달라는 요청이 오면 브로커는 그 파티션의 여러 세그먼트 중 어디에 오프셋 100이 있는지 빠르게 찾아야 한다. 이를 위해 Kafka는 파티션마다 인덱스를 유지하며, 인덱스는 오프셋을 세그먼트 파일과 파일 내 위치로 매핑한다.

인덱스도 세그먼트 단위로 쪼개져 있어서 메시지가 정리될 때 오래된 인덱스 항목도 함께 삭제할 수 있다. Kafka는 인덱스에 대해서는 체크섬을 유지하지 않는다. 인덱스가 손상되면 대응하는 로그 세그먼트를 다시 읽어 오프셋과 위치를 기록함으로써 자동으로 재생성한다. 그래서 관리자가 인덱스 세그먼트를 삭제해도 완전히 안전하며, 자동으로 다시 만들어진다.

로그 컴팩션이 필요한 이유

보통 Kafka는 정해진 기간 동안 메시지를 보관하고 리텐션이 지난 것을 지운다. 그런데 고객의 배송 주소를 Kafka에 저장하는 경우를 생각해 보자. 이때는 최근 1주일이나 1년치가 아니라 고객별 마지막 주소를 보관하는 편이 맞다. 그러면 오래된 주소를 걱정할 필요가 없고, 한동안 이사하지 않은 고객의 주소도 유지된다. 애플리케이션이 자신의 현재 상태를 Kafka에 저장하는 경우도 마찬가지다. 상태가 바뀔 때마다 새 상태를 쓰고, 크래시에서 복구할 때 그 메시지들을 읽어 최신 상태를 복원한다. 이때 필요한 것은 크래시 직전의 최신 상태뿐이지 그 사이의 모든 변화가 아니다.

Kafka는 토픽의 리텐션 정책을 delete(리텐션이 지난 이벤트 삭제) 대신 compact(키별 최신 값만 보관)로 설정할 수 있게 해 이런 사용 사례를 지원한다. 당연히 compact는 애플리케이션이 키와 값을 모두 담은 이벤트를 생산하는 토픽에서만 의미가 있다. 토픽에 null 키가 들어 있으면 컴팩션은 실패한다.

컴팩션 동작: clean/dirty와 오프셋 맵

각 로그는 두 부분으로 나뉘어 다뤄진다. clean 영역은 이전에 컴팩션된 메시지들로, 키마다 지난 컴팩션 시점의 최신 값 하나씩만 들어 있다. dirty 영역은 마지막 컴팩션 이후에 쓰인 메시지들이다.

Kafka가 시작할 때 컴팩션이 활성화되어 있으면(다소 어색한 이름의 `log.cleaner.enabled` 설정) 각 브로커는 컴팩션 매니저 스레드 하나와 여러 개의 컴팩션 스레드를 띄운다. 각 스레드는 전체 파티션 크기 대비 dirty 메시지 비율이 가장 높은 파티션을 골라 정리한다.

클리너 스레드는 파티션의 dirty 영역을 읽어 인메모리 맵을 만든다. 맵의 각 엔트리는 메시지 키의 16바이트 해시와, 같은 키를 가진 이전 메시지의 8바이트 오프셋으로 구성된다. 즉 엔트리 하나가 24바이트만 사용한다. 1GB 세그먼트에서 메시지 하나가 1KB라면 세그먼트에는 100만 개의 메시지가 있고, 이를 컴팩션하는 데 24MB 맵이면 충분하다(키가 반복되면 같은 해시 엔트리를 재사용하므로 훨씬 덜 쓸 수도 있다). 상당히 효율적이다.

관리자는 컴팩션 스레드들이 이 오프셋 맵에 쓸 수 있는 총 메모리를 설정한다. 스레드마다 자기 맵을 가지지만 설정값은 전체 스레드의 합계 기준이다. 1GB를 설정하고 클리너 스레드가 5개면 스레드당 200MB를 갖는다. 파티션의 dirty 영역 전체가 맵에 들어갈 필요는 없지만, 최소한 하나의 세그먼트는 통째로 들어가야 한다. 그렇지 못하면 Kafka는 오류를 로그에 남기고, 관리자는 메모리를 늘리거나 클리너 스레드 수를 줄여야 한다. 몇 개의 세그먼트만 들어간다면 Kafka는 맵에 들어가는 가장 오래된 세그먼트부터 컴팩션하고 나머지는 dirty 상태로 다음 컴팩션을 기다린다.

맵이 완성되면 클리너 스레드는 가장 오래된 것부터 clean 세그먼트를 읽어 내용을 맵과 대조한다. 메시지의 키가 맵에 없으면 방금 읽은 값이 여전히 최신이라는 뜻이므로 대체(replacement) 세그먼트로 복사한다. 키가 맵에 있으면 파티션 뒤쪽에 같은 키의 더 새로운 값이 있다는 뜻이므로 그 메시지는 생략한다. 최신 값을 가진 메시지를 모두 복사하고 나면 대체 세그먼트를 원본과 교체하고 다음 세그먼트로 넘어간다. 전 과정이 끝나면 키마다 최신 값 하나씩만 남는다.

삭제 이벤트(톰스톤)와 컴팩션 시점

키마다 최신 메시지를 항상 남긴다면, 사용자가 서비스를 떠나 법적으로 모든 흔적을 지워야 하는 경우처럼 특정 키의 메시지를 정말로 전부 지우고 싶을 때는 어떻게 할까?

마지막 메시지조차 남기지 않고 키를 시스템에서 완전히 삭제하려면, 애플리케이션이 그 키와 null 값을 담은 메시지를 생산해야 한다. 클리너 스레드는 그런 메시지를 발견하면 우선 일반적인 컴팩션을 수행해 null 값을 가진 메시지만 남긴다. 그리고 이 특별한 메시지(톰스톤)를 설정된 시간 동안 유지한다. 그 사이 컨슈머들은 이 메시지를 보고 값이 삭제되었음을 알 수 있다. 예컨대 Kafka의 데이터를 관계형 데이터베이스로 복사하는 컨슈머는 톰스톤을 보고 DB에서 해당 사용자를 삭제해야 한다는 것을 안다. 설정된 시간이 지나면 클리너 스레드가 톰스톤도 제거하고 그 키는 파티션에서 사라진다. 컨슈머가 톰스톤을 볼 시간을 충분히 주는 것이 중요한 이유는, 컨슈머가 몇 시간 멈춰 있다가 톰스톤을 놓치면 그 키가 소비 대상에서 그냥 보이지 않게 될 뿐 삭제되었다는 사실을 알 수 없기 때문이다.

한편 delete 정책이 현재 액티브 세그먼트를 절대 지우지 않는 것과 마찬가지로, compact 정책도 현재 세그먼트를 절대 컴팩션하지 않는다. 메시지는 비활성 세그먼트에 있을 때만 컴팩션 대상이 된다. 0.10.0 이하 버전에서 Kafka는 토픽의 dirty 레코드 비율이 50%에 도달하면 컴팩션을 시작했다. 너무 자주 컴팩션하면 토픽의 읽기/쓰기 성능에 영향을 주고, 그렇다고 dirty 레코드를 너무 많이 남겨 두면 디스크를 낭비하므로 그 절충으로 정해진 값이며 관리자가 조정할 수 있다.