기본 Kafka 운영
기본 Kafka 운영 (Basic Kafka Operations)
이 페이지는 Kafka 클러스터 운영에서 가장 흔하게 하는 작업들을 하나로 모아 다뤄요. 토픽 추가·수정·삭제, 정상 종료, 리더 균형, 파티션 재배치, 컨슈머 그룹·공유 그룹 관리, 팬아웃 확장, 쿼터 설정까지 명령어 예시와 함께 꼼꼼히 설명돼 있어요.
출처: 문서
본문
이 섹션은 Kafka 클러스터에서 수행할 가장 흔한 운영 작업을 다룹니다. 이 섹션에서 다루는 모든 도구는 Kafka 배포판의 bin/ 디렉터리 아래에 있으며, 각 도구는 인자 없이 실행하면 가능한 모든 커맨드라인 옵션에 대한 세부 정보를 출력합니다.
토픽 추가와 제거 (Adding and removing topics)
토픽을 수동으로 추가하거나, 존재하지 않는 토픽에 데이터가 처음 게시될 때 자동으로 생성되게 할 수 있습니다. 토픽이 자동 생성된다면 자동 생성 토픽에 사용되는 기본 토픽 구성을 튜닝하고 싶을 수 있습니다.
토픽은 토픽 도구로 추가되고 수정됩니다.
$ bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic my_topic_name \
--partitions 20 --replication-factor 3 --config x=y
복제 팩터는 기록된 각 메시지를 몇 개의 서버가 복제할지 제어합니다. 복제 팩터가 3이라면 데이터 접근을 잃기 전까지 최대 2개의 서버가 실패할 수 있습니다. 우리는 복제 팩터 2 또는 3을 권장합니다. 데이터 소비를 중단하지 않고 머신을 투명하게 재부팅할 수 있기 때문입니다.
파티션 수는 토픽이 몇 개의 로그로 분할(shard)될지 제어합니다. 파티션 수에는 몇 가지 영향이 있습니다. 첫째, 각 파티션은 단일 서버에 완전히 맞아야 합니다. 따라서 20개의 파티션이 있으면 전체 데이터 셋(및 읽기·쓰기 부하)이 20개 이하의 서버(복제본 제외)가 처리합니다. 마지막으로 파티션 수는 컨슈머의 최대 병렬성을 결정합니다. 이는 개념 섹션에서 더 자세히 다룹니다.
각 분할된 파티션 로그는 Kafka 로그 디렉터리 아래의 자신의 폴더에 배치됩니다. 이러한 폴더의 이름은 토픽 이름 뒤에 대시(-)와 파티션 ID를 붙인 것으로 구성됩니다. 일반적인 폴더 이름은 255자를 넘을 수 없으므로 토픽 이름의 길이에 제한이 있습니다. 파티션 수가 100,000을 넘지 않을 것이라고 가정합니다. 따라서 토픽 이름은 249자를 넘을 수 없습니다. 이는 폴더 이름에 대시와 잠재적으로 5자리 파티션 ID를 위한 충분한 공간을 남겨줍니다.
커맨드 라인에서 추가된 구성은 데이터가 보존되어야 하는 시간 길이 같은 것에 대한 서버의 기본 설정을 오버라이드합니다. 전체 토픽별 구성 집합은 여기에 문서화되어 있습니다.
토픽 수정 (Modifying topics)
동일한 토픽 도구를 사용해 토픽의 구성이나 파티셔닝을 변경할 수 있습니다.
파티션을 추가하려면 다음을 수행할 수 있습니다.
$ bin/kafka-topics.sh --bootstrap-server localhost:9092 --alter --topic my_topic_name \
--partitions 40
참고: 토픽의 파티션 수를 동적으로 늘리는 것은 몇 가지 중요한 고려 사항과 잠재적 부작용이 있습니다.
- 키 분포 변경: 데이터가
hash(key) % number_of_partitions로 파티셔닝된다면, 기본 파티셔너의 매핑 로직은 파티션 수가 증가할 때 변경됩니다. 즉, 같은 키를 가진 메시지가 확장 후 다른 파티션으로 라우팅될 수 있어 기존 키에 대한 메시지 순서 보장에 영향을 줄 수 있습니다. Kafka는 기존 데이터를 자동으로 재분배하려 하지 않습니다. - auto.offset.reset=latest인 경우 잠재적 데이터 손실:
auto.offset.reset=latest로 구성된 기존 컨슈머는 파티션 생성과 컨슈머 발견 사이의 창 동안 새 파티션에 생성된 메시지를 놓칠 수 있습니다. 이는 컨슈머가 새 파티션을 즉시 감지하지 못하고, 컨슈머가 리밸런스되기 전에 그 파티션들에 생성된 메시지가 건너뛰어지기 때문에 발생합니다. - 메타데이터 전파 지연: 새 파티션은 메타데이터 갱신 간격(
metadata.max.age.ms로 제어) 때문에 프로듀서와 컨슈머에게 즉시 보이지 않습니다. 클라이언트가 새 파티션을 인지하지 못하는 짧은 기간이 있어 메시지 분포가 고르지 않거나 컨슈머 랙이 발생할 수 있습니다. - 내부 토픽의 위험: 사용자는
__consumer_offsets,__transaction_state,__share_group_state,__cluster_metadata같은 Kafka 내부 상태 토픽의 파티션을 수동으로 늘려서는 안 됩니다. 그렇게 하면 코디네이터 매핑 로직이 깨지고, 상태 불일치를 일으키며, 데이터 손상이나 시스템 실패로 이어질 수 있습니다. 이러한 토픽은 Kafka가 자동으로 관리하므로 수동으로 수정해서는 안 됩니다.
구성을 추가하려면:
$ bin/kafka-configs.sh --bootstrap-server localhost:9092 --entity-type topics --entity-name my_topic_name --alter --add-config x=y
구성을 제거하려면:
$ bin/kafka-configs.sh --bootstrap-server localhost:9092 --entity-type topics --entity-name my_topic_name --alter --delete-config x
마지막으로 토픽 삭제:
$ bin/kafka-topics.sh --bootstrap-server localhost:9092 --delete --topic my_topic_name
Kafka는 현재 토픽의 파티션 수를 줄이는 것을 지원하지 않습니다.
토픽의 복제 팩터를 변경하는 지침은 여기에서 찾을 수 있습니다.
정상 종료 (Graceful shutdown)
Kafka 클러스터는 브로커 종료나 장애를 자동으로 감지하고 그 머신의 파티션에 대한 새 리더를 선출합니다. 이는 서버가 실패하든, 유지보수나 구성 변경을 위해 의도적으로 내려지든 발생합니다. 후자의 경우 Kafka는 서버를 그냥 죽이는 것보다 더 정상적인 중지 메커니즘을 지원합니다. 서버가 정상적으로 중지되면 활용하는 두 가지 최적화가 있습니다.
- 모든 로그를 디스크에 동기화하여 재시작 시 로그 복구(즉, 로그 꼬리에 있는 모든 메시지의 체크섬 검증)를 하지 않도록 함. 로그 복구는 시간이 걸리므로 이는 의도적 재시작을 빠르게 만듭니다.
- 서버가 리더인 파티션을 종료 전에 다른 복제본으로 마이그레이션하여 리더십 전환을 더 빠르게 만들고 각 파티션이 몇 밀리초만 사용 불가능하게 함. 로그 동기화는 하드 킬 외의 방식으로 서버가 중지될 때마다 자동으로 발생하지만, 제어된 리더십 마이그레이션은 특수 설정을 사용해야 합니다.
controlled.shutdown.enable=true
제어 종료는 브로커가 호스팅하는 모든 파티션에 복제본이 있는 경우(즉 복제 팩터가 1보다 크고 그 복제본 중 적어도 하나가 살아 있는 경우)에만 성공한다는 점에 유의하세요. 마지막 복제본을 종료하면 그 토픽 파티션을 사용할 수 없게 되므로, 이것이 일반적으로 원하는 동작입니다.
리더십 균형 (Balancing leadership)
브로커가 중지되거나 크래시할 때마다 그 브로커 파티션의 리더십은 다른 복제본으로 이전됩니다. 브로커가 재시작되면 모든 파티션에 대해 팔로워일 뿐이므로, 클라이언트의 읽기·쓰기에 사용되지 않습니다.
이 불균형을 피하기 위해 Kafka는 선호 복제본(preferred replicas) 개념이 있습니다. 파티션의 복제본 목록이 1,5,9라면 노드 1은 복제본 목록에서 더 앞에 있으므로 노드 5나 9보다 리더로 선호됩니다. 기본적으로 Kafka 클러스터는 리더십을 선호 복제본으로 복원하려 합니다. 이 동작은 다음으로 구성됩니다.
auto.leader.rebalance.enable=true
이를 false로 설정할 수도 있지만, 그 경우 다음 명령을 실행해 복원된 복제본으로 리더십을 수동으로 복원해야 합니다.
$ bin/kafka-leader-election.sh --bootstrap-server localhost:9092 --election-type preferred --all-topic-partitions
랙 간 복제본 균형 (Balancing replicas across racks)
랙 인식(rack awareness) 기능은 같은 파티션의 복제본을 서로 다른 랙에 분산합니다. 이는 Kafka가 브로커 장애에 대해 제공하는 보장을 랙 장애까지 확장하며, 한 랙의 모든 브로커가 동시에 실패해도 데이터 손실 위험을 제한합니다. 이 기능은 EC2의 가용 영역 같은 다른 브로커 그룹에도 적용할 수 있습니다.
브로커 구성에 속성을 추가해 특정 랙에 속한다고 지정할 수 있습니다.
broker.rack=my-rack-id
토픽이 생성·수정되거나 복제본이 재분배될 때 랙 제약이 존중되어 복제본이 가능한 한 많은 랙에 걸치도록 보장합니다(파티션은 min(#racks, replication-factor)개의 서로 다른 랙에 걸침).
복제본을 브로커에 할당하는 알고리즘은 브로커가 랙에 어떻게 분산되어 있든 브로커당 리더 수가 일정하도록 보장합니다. 이는 균형 잡힌 처리량을 보장합니다.
하지만 랙에 서로 다른 수의 브로커가 할당되면 복제본 할당은 고르지 않습니다. 브로커가 적은 랙이 더 많은 복제본을 받게 되어 저장 공간을 더 많이 사용하고 복제에 더 많은 리소스를 투입하게 됩니다. 따라서 랙당 동일한 수의 브로커를 구성하는 것이 합리적입니다.
클러스터 간 데이터 미러링 & 지리적 복제 (Mirroring data between clusters & Geo-replication)
Kafka 관리자는 개별 Kafka 클러스터, 데이터센터, 또는 지리적 지역의 경계를 넘는 데이터 흐름을 정의할 수 있습니다. 자세한 내용은 지리적 복제(Geo-Replication) 섹션을 참고하세요.
컨슈머 위치 확인 (Checking consumer position)
때로는 컨슈머의 위치를 보는 것이 유용합니다. 컨슈머 그룹의 모든 컨슈머 위치와 로그 끝에서 얼마나 뒤처져 있는지 보여주는 도구가 있습니다. my-topic이라는 토픽을 소비하는 my-group이라는 이름의 컨슈머 그룹에서 이 도구를 실행하면 다음과 같습니다.
$ bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group my-group
TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID
my-topic 0 2 4 2 consumer-1-029af89c-873c-4751-a720-cefd41a669d6 /127.0.0.1 consumer-1
my-topic 1 2 3 1 consumer-1-029af89c-873c-4751-a720-cefd41a669d6 /127.0.0.1 consumer-1
my-topic 2 2 3 1 consumer-2-42c1abd4-e3b2-425d-a8bb-e1ea49b29bb2 /127.0.0.1 consumer-2
그룹 관리 (Managing groups)
GroupCommand 도구로 컨슈머 그룹, 공유 그룹, 스트림 그룹을 포함한 모든 유형의 그룹을 나열할 수 있습니다. 각 그룹 유형에는 그 유형의 그룹을 관리하는 전용 도구가 있습니다. 예를 들어 클러스터의 모든 그룹을 나열하려면:
$ bin/kafka-groups.sh --bootstrap-server localhost:9092 --list
GROUP TYPE PROTOCOL
my-consumer-group Consumer consumer
my-share-group Share share
컨슈머 그룹 관리 (Managing consumer groups)
ConsumerGroupCommand 도구로 컨슈머 그룹을 나열, 설명(describe), 삭제할 수 있습니다. 컨슈머 그룹은 수동으로, 또는 그 그룹의 마지막 커밋 오프셋이 만료되면 자동으로 삭제될 수 있습니다. 수동 삭제는 그룹에 활성 멤버가 없을 때만 작동합니다. 예를 들어 모든 토픽에 걸친 모든 컨슈머 그룹을 나열하려면:
$ bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list
test-consumer-group
오프셋을 보려면 앞서 언급했듯이 컨슈머 그룹을 다음과 같이 "describe"합니다.
$ bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group my-group
TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID
topic3 0 241019 395308 154289 consumer2-e76ea8c3-5d30-4299-9005-47eb41f3d3c4 /127.0.0.1 consumer2
topic2 1 520678 803288 282610 consumer2-e76ea8c3-5d30-4299-9005-47eb41f3d3c4 /127.0.0.1 consumer2
topic3 1 241018 398817 157799 consumer2-e76ea8c3-5d30-4299-9005-47eb41f3d3c4 /127.0.0.1 consumer2
topic1 0 854144 855809 1665 consumer1-3fc8d6f1-581a-4472-bdf3-3515b4aee8c1 /127.0.0.1 consumer1
topic2 0 460537 803290 342753 consumer1-3fc8d6f1-581a-4472-bdf3-3515b4aee8c1 /127.0.0.1 consumer1
topic3 2 243655 398812 155157 consumer4-117fe4d3-c6c1-4178-8ee9-eb4a3954bee0 /127.0.0.1 consumer4
컨슈머 그룹이 consumer 프로토콜을 사용한다면 admin 클라이언트는 그룹에서 사용되는 모든 토픽(멤버들이 구독한 토픽)에 대한 DESCRIBE 접근이 필요하다는 점에 유의하세요. 반대로 classic 프로토콜은 모든 토픽의 DESCRIBE 인가를 요구하지 않습니다. 컨슈머 그룹에 대해 더 자세한 정보를 제공하는 몇 가지 추가 "describe" 옵션이 있습니다.
$ bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group my-group --members
CONSUMER-ID HOST CLIENT-ID #PARTITIONS
consumer1-3fc8d6f1-581a-4472-bdf3-3515b4aee8c1 /127.0.0.1 consumer1 2
consumer4-117fe4d3-c6c1-4178-8ee9-eb4a3954bee0 /127.0.0.1 consumer4 1
consumer2-e76ea8c3-5d30-4299-9005-47eb41f3d3c4 /127.0.0.1 consumer2 3
consumer3-ecea43e4-1f01-479f-8349-f9130b75d8ee /127.0.0.1 consumer3 0
--members: 이 옵션은 컨슈머 그룹의 모든 활성 멤버 목록을 제공합니다.
$ bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group my-group --members --verbose
CONSUMER-ID HOST CLIENT-ID #PARTITIONS ASSIGNMENT
consumer1-3fc8d6f1-581a-4472-bdf3-3515b4aee8c1 /127.0.0.1 consumer1 2 topic1(0), topic2(0)
consumer4-117fe4d3-c6c1-4178-8ee9-eb4a3954bee0 /127.0.0.1 consumer4 1 topic3(2)
consumer2-e76ea8c3-5d30-4299-9005-47eb41f3d3c4 /127.0.0.1 consumer2 3 topic2(1), topic3(0,1)
consumer3-ecea43e4-1f01-479f-8349-f9130b75d8ee /127.0.0.1 consumer3 0 -
--members --verbose: 위 "--members" 옵션이 보고하는 정보에 더해 각 멤버에게 할당된 파티션도 제공.--offsets: 기본 describe 옵션이며 "--describe" 옵션과 동일한 출력 제공.
$ bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group my-group --state
COORDINATOR (ID) ASSIGNMENT-STRATEGY STATE #MEMBERS
localhost:9092 (0) range Stable 4
--state: 이 옵션은 유용한 그룹 수준 정보를 제공.
하나 이상의 컨슈머 그룹을 수동으로 삭제하려면 "--delete" 옵션을 사용할 수 있습니다.
$ bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --delete --group my-group --group my-other-group
Deletion of requested consumer groups ('my-group', 'my-other-group') was successful.
컨슈머 그룹의 오프셋을 리셋하려면 "--reset-offsets" 옵션을 사용할 수 있습니다. 이 옵션은 한 번에 하나의 컨슈머 그룹을 지원합니다. --all-topics 또는 --topic 중 하나의 스코프를 정의해야 합니다. '--from-file' 시나리오를 사용하지 않는다면 스코프 하나를 선택해야 합니다. 또한 먼저 컨슈머 인스턴스가 비활성인지 확인하세요. 자세한 내용은 KIP-122를 참고하세요.
실행 옵션은 3가지입니다.
- (기본) 리셋할 오프셋 표시.
--execute:--reset-offsets프로세스 실행.--export: 결과를 CSV 형식으로 내보내기.
--reset-offsets에는 선택할 수 있는 다음 시나리오도 있습니다.
--to-datetime <String: datetime>: datetime의 오프셋으로 리셋. 형식: 'YYYY-MM-DDThh:mm:ss.sss'.--to-earliest: 가장 이른 오프셋으로 리셋.--to-latest: 가장 최신 오프셋으로 리셋.--shift-by <Long: number-of-offsets>: 현재 오프셋에서 'n'만큼 이동해 오프셋 리셋. 'n'은 양수 또는 음수일 수 있음.--from-file: CSV 파일에 정의된 값으로 오프셋 리셋.--to-current: 현재 오프셋으로 리셋.--by-duration <String: duration>: 현재 타임스탬프에서 duration만큼의 오프셋으로 리셋. 형식: 'PnDTnHnMnS'.--to-offset: 특정 오프셋으로 리셋.
범위를 벗어난 오프셋은 사용 가능한 오프셋 끝으로 조정된다는 점에 유의하세요. 예를 들어 오프셋 끝이 10이고 오프셋 이동 요청이 15라면 실제로 오프셋 10이 선택됩니다.
예를 들어 컨슈머 그룹의 오프셋을 최신 오프셋으로 리셋하려면:
$ bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --reset-offsets --group my-group --topic topic1 --to-latest
TOPIC PARTITION NEW-OFFSET
topic1 0 0
공유 그룹 관리 (Managing share groups)
ShareGroupCommand 도구로 공유 그룹(share group)을 나열, 설명, 삭제할 수 있습니다. 활성 멤버가 없는 공유 그룹만 삭제할 수 있습니다. 예를 들어 클러스터의 모든 공유 그룹을 나열하려면:
$ bin/kafka-share-groups.sh --bootstrap-server localhost:9092 --list
my-share-group
현재 시작 오프셋과 랙을 보려면 "--describe" 옵션을 사용합니다.
$ bin/kafka-share-groups.sh --bootstrap-server localhost:9092 --describe --group my-share-group
GROUP TOPIC PARTITION START-OFFSET LAG
my-share-group topic1 0 4 0
시작 오프셋은 공유 컨슈머에게 전달 여부가 평가 중인 in-flight 레코드의 가장 이른 오프셋입니다. 시작 오프셋 이후의 일부 레코드는 이미 전달을 완료했을 수 있습니다. 참고: admin 클라이언트는 그룹에서 사용되는 모든 토픽에 대한 DESCRIBE 접근이 필요합니다. 공유 그룹에 대한 더 자세한 정보를 제공하는 --describe 옵션이 여러 개 있습니다.
$ bin/kafka-share-groups.sh --bootstrap-server localhost:9092 --describe --group my-share-group --members
GROUP CONSUMER-ID HOST CLIENT-ID #PARTITIONS ASSIGNMENT
my-share-group 94wrSQNmRda9Q6sk6jMO6Q /127.0.0.1 console-share-consumer 1 topic1:0
my-share-group EfI0sha8QSKSrL_-I_zaTA /127.0.0.1 console-share-consumer 1 topic1:0
--members: 공유 그룹의 활성 멤버 설명.
두 멤버 모두 공유 중인 같은 파티션이 할당된 것을 볼 수 있습니다.
--offsets: 기본 describe 옵션. "--describe" 옵션과 동일한 출력 제공.
$ bin/kafka-share-groups.sh --bootstrap-server localhost:9092 --describe --group my-share-group --state
GROUP COORDINATOR (ID) STATE #MEMBERS
my-share-group localhost:9092 (1) Stable 2
--state: 공유 그룹의 상태 요약 설명.
공유 그룹의 오프셋을 리셋하려면 "--reset-offsets" 옵션을 사용합니다.
실행 옵션은 2가지입니다.
--dry-run: 리셋할 오프셋 표시.--execute:--reset-offsets프로세스 실행.
--reset-offsets에는 다음 시나리오도 있습니다.
--to-datetime <String: datetime>: datetime의 오프셋으로 리셋. 형식: 'YYYY-MM-DDThh:mm:ss.sss'.--to-earliest: 가장 이른 오프셋으로 리셋.--to-latest: 가장 최신 오프셋으로 리셋.
예를 들어 공유 그룹의 오프셋을 최신 오프셋으로 리셋하려면:
$ bin/kafka-share-groups.sh --bootstrap-server localhost:9092 --reset-offsets --group my-share-group --topic topic1 --to-latest --execute
GROUP TOPIC PARTITION NEW-OFFSET
my-share-group topic1 0 10
공유 그룹의 개별 토픽 오프셋을 삭제하려면 "--delete-offsets" 옵션을 사용합니다.
$ bin/kafka-share-groups.sh --bootstrap-server localhost:9092 --delete-offsets --group my-share-group --topic topic1
TOPIC STATUS
topic1 Successful
하나 이상의 공유 그룹을 삭제하려면 "--delete" 옵션을 사용합니다.
$ bin/kafka-share-groups.sh --bootstrap-server localhost:9092 --delete --group my-share-group
Deletion of requested share groups ('my-share-group') was successful.
클러스터 확장 (Expanding your cluster)
Kafka 클러스터에 서버를 추가하는 것은 쉽습니다. 고유한 브로커 ID를 할당하고 새 서버에서 Kafka를 시작하기만 하면 됩니다. 하지만 새 서버에는 데이터 파티션이 자동으로 할당되지 않으므로, 파티션이 이동되지 않는 한 새 토픽이 생성될 때까지 아무 작업도 하지 않습니다. 그래서 보통 머신을 클러스터에 추가할 때 기존 데이터의 일부를 이 머신들로 마이그레이션하고 싶을 것입니다.
데이터 마이그레이션 과정은 수동으로 시작되지만 완전히 자동화됩니다. 내부적으로 Kafka는 새 서버를 마이그레이션 중인 파티션의 팔로워로 추가하고 그 파티션의 기존 데이터를 완전히 복제하도록 허용합니다. 새 서버가 이 파티션의 내용을 완전히 복제하고 in-sync 복제본에 합류하면 기존 복제본 중 하나가 자신의 파티션 데이터를 삭제합니다.
파티션 재배치 도구(partition reassignment tool)는 브로커 간에 파티션을 이동하는 데 사용할 수 있습니다. 이상적인 파티션 분포는 모든 브로커에 걸쳐 고른 데이터 부하와 파티션 크기를 보장합니다. 파티션 재배치 도구는 Kafka 클러스터의 데이터 분포를 자동으로 연구하고 균등한 부하 분포를 얻기 위해 파티션을 옮기는 기능은 없습니다. 따라서 관리자가 어떤 토픽이나 파티션을 옮겨야 하는지 알아내야 합니다.
파티션 재배치 도구는 상호 배타적인 3가지 모드로 실행할 수 있습니다.
--generate: 이 모드에서 토픽 목록과 브로커 목록이 주어지면 도구는 지정된 토픽의 모든 파티션을 새 브로커로 이동하는 후보 재배치를 생성합니다. 이 옵션은 토픽 목록과 대상 브로커가 주어졌을 때 파티션 재배치 계획을 생성하는 편리한 방법을 제공할 뿐입니다.--execute: 이 모드에서 도구는 사용자가 제공한 재배치 계획을 기반으로 파티션 재배치를 시작합니다. (--reassignment-json-file옵션 사용). 이는 관리자가 직접 만든 커스텀 재배치 계획이거나--generate옵션으로 제공된 것일 수 있음.--verify: 이 모드에서 도구는 마지막--execute동안 나열된 모든 파티션에 대한 재배치 상태를 검증합니다. 상태는 성공적으로 완료, 실패, 또는 진행 중 중 하나일 수 있음.
새 머신으로 자동 데이터 마이그레이션 (Automatically migrating data to new machines)
파티션 재배치 도구는 일부 토픽을 현재 브로커 집합에서 새로 추가된 브로커로 이동하는 데 사용할 수 있습니다. 한 번에 하나의 파티션을 이동하는 것보다 전체 토픽을 새 브로커 집합으로 이동하는 것이 더 쉽기 때문에 이는 기존 클러스터를 확장할 때 특히 유용합니다. 이 용도로 사용할 때 사용자는 새 브로커 집합으로 이동해야 하는 토픽 목록과 대상 새 브로커 목록을 제공해야 합니다. 그러면 도구는 주어진 토픽 목록의 모든 파티션을 새 브로커 집합에 고르게 분산합니다. 이 이동 중에 토픽의 복제 팩터는 일정하게 유지됩니다. 사실상 입력 토픽 목록의 모든 파티션의 복제본이 이전 브로커 집합에서 새로 추가된 브로커로 이동됩니다.
예를 들어 다음 예제는 토픽 foo1, foo2의 모든 파티션을 새 브로커 집합 5,6으로 이동합니다. 이 이동이 끝나면 토픽 foo1과 foo2의 모든 파티션은 브로커 5,6에만 존재합니다.
도구가 입력 토픽 목록을 json 파일로 받으므로 먼저 이동할 토픽을 식별하고 다음과 같이 json 파일을 만들어야 합니다.
$ cat topics-to-move.json
{
"topics": [
{ "topic": "foo1" },
{ "topic": "foo2" }
],
"version": 1
}
json 파일이 준비되면 파티션 재배치 도구를 사용해 후보 할당을 생성합니다.
$ bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --topics-to-move-json-file topics-to-move.json --broker-list "5,6" --generate
Current partition replica assignment
{"version":1,
"partitions":[{"topic":"foo1","partition":0,"replicas":[2,1],"log_dirs":["any"]},
{"topic":"foo1","partition":1,"replicas":[1,3],"log_dirs":["any"]},
{"topic":"foo1","partition":2,"replicas":[3,4],"log_dirs":["any"]},
{"topic":"foo2","partition":0,"replicas":[4,2],"log_dirs":["any"]},
{"topic":"foo2","partition":1,"replicas":[2,1],"log_dirs":["any"]},
{"topic":"foo2","partition":2,"replicas":[1,3],"log_dirs":["any"]}]
}
Proposed partition reassignment configuration
{"version":1,
"partitions":[{"topic":"foo1","partition":0,"replicas":[6,5],"log_dirs":["any"]},
{"topic":"foo1","partition":1,"replicas":[5,6],"log_dirs":["any"]},
{"topic":"foo1","partition":2,"replicas":[6,5],"log_dirs":["any"]},
{"topic":"foo2","partition":0,"replicas":[5,6],"log_dirs":["any"]},
{"topic":"foo2","partition":1,"replicas":[6,5],"log_dirs":["any"]},
{"topic":"foo2","partition":2,"replicas":[5,6],"log_dirs":["any"]}]
}
도구는 토픽 foo1, foo2의 모든 파티션을 브로커 5,6으로 이동할 후보 할당을 생성합니다. 그러나 이 시점에서 파티션 이동이 시작된 것이 아니라 현재 할당과 제안된 새 할당만 알려준다는 점에 유의하세요. 롤백하고 싶은 경우를 대비해 현재 할당을 저장해야 합니다. 새 할당은 json 파일(예: expand-cluster-reassignment.json)에 저장해 다음과 같이 --execute 옵션으로 도구에 입력해야 합니다.
$ bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --reassignment-json-file expand-cluster-reassignment.json --execute
Current partition replica assignment
{"version":1,
"partitions":[{"topic":"foo1","partition":0,"replicas":[2,1],"log_dirs":["any"]},
{"topic":"foo1","partition":1,"replicas":[1,3],"log_dirs":["any"]},
{"topic":"foo1","partition":2,"replicas":[3,4],"log_dirs":["any"]},
{"topic":"foo2","partition":0,"replicas":[4,2],"log_dirs":["any"]},
{"topic":"foo2","partition":1,"replicas":[2,1],"log_dirs":["any"]},
{"topic":"foo2","partition":2,"replicas":[1,3],"log_dirs":["any"]}]
}
Save this to use as the --reassignment-json-file option during rollback
Successfully started partition reassignments for foo1-0,foo1-1,foo1-2,foo2-0,foo2-1,foo2-2
마지막으로 --verify 옵션을 도구와 함께 사용해 파티션 재배치 상태를 확인할 수 있습니다. --execute 옵션과 함께 사용된 것과 동일한 expand-cluster-reassignment.json을 --verify 옵션과 함께 사용해야 합니다.
$ bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --reassignment-json-file expand-cluster-reassignment.json --verify
Status of partition reassignment:
Reassignment of partition [foo1,0] is completed
Reassignment of partition [foo1,1] is still in progress
Reassignment of partition [foo1,2] is still in progress
Reassignment of partition [foo2,0] is completed
Reassignment of partition [foo2,1] is completed
Reassignment of partition [foo2,2] is completed
커스텀 파티션 할당과 마이그레이션 (Custom partition assignment and migration)
파티션 재배치 도구는 파티션의 복제본을 특정 브로커 집합으로 선택적으로 이동하는 데도 사용할 수 있습니다. 이 방식으로 사용할 때는 사용자가 재배치 계획을 알고 있으며 도구가 후보 재배치를 생성할 필요가 없다고 가정합니다. 사실상 --generate 단계를 건너뛰고 바로 --execute 단계로 가는 것입니다.
예를 들어 다음 예제는 토픽 foo1의 파티션 0을 브로커 5,6으로, 토픽 foo2의 파티션 1을 브로커 2,3으로 이동합니다.
첫 단계는 json 파일에 커스텀 재배치 계획을 직접 만드는 것입니다.
$ cat custom-reassignment.json
{"version":1,"partitions":[{"topic":"foo1","partition":0,"replicas":[5,6]},{"topic":"foo2","partition":1,"replicas":[2,3]}]}
그런 다음 --execute 옵션과 함께 json 파일을 사용해 재배치 프로세스를 시작합니다.
$ bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --reassignment-json-file custom-reassignment.json --execute
Current partition replica assignment
{"version":1,
"partitions":[{"topic":"foo1","partition":0,"replicas":[1,2],"log_dirs":["any"]},
{"topic":"foo2","partition":1,"replicas":[3,4],"log_dirs":["any"]}]
}
Save this to use as the --reassignment-json-file option during rollback
Successfully started partition reassignments for foo1-0,foo2-1
--verify 옵션을 도구와 함께 사용해 파티션 재배치 상태를 확인할 수 있습니다. --execute 옵션과 함께 사용된 것과 동일한 custom-reassignment.json을 --verify 옵션과 함께 사용해야 합니다.
$ bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --reassignment-json-file custom-reassignment.json --verify
Status of partition reassignment:
Reassignment of partition [foo1,0] is completed
Reassignment of partition [foo2,1] is completed
브로커와 로그 디렉터리 폐기 (Decommissioning brokers and log directories)
브로커 폐기 (Decommissioning brokers)
브로커를 폐기하는 첫 단계는 Admin API를 통해 브로커를 cordoned(봉쇄)로 표시하는 것입니다.
예를 들어 브로커 1을 cordon하려면:
$ bin/kafka-configs.sh --bootstrap-server localhost:9092 --alter --add-config cordoned.log.dirs="*" --entity-type brokers --entity-name 1
Completed updating config for broker 1.
그런 다음 그 브로커의 모든 파티션을 클러스터의 다른 브로커로 재배치합니다. 파티션 재배치 도구는 아직 브로커 폐기를 위한 재배치 계획을 자동으로 생성하는 기능이 없습니다. 따라서 관리자는 폐기할 브로커가 호스팅하는 모든 파티션의 복제본을 나머지 브로커로 이동하는 재배치 계획을 만들어야 합니다.
모든 재배치가 완료되면 브로커를 종료하고 등록을 해제(unregister)해 클러스터에서 제거합니다.
예를 들어 브로커 1을 등록 해제하려면:
$ bin/kafka-cluster.sh unregister --bootstrap-server localhost:9092 --id 1
로그 디렉터리 폐기 (Decommissioning log directories)
로그 디렉터리를 폐기하는 첫 단계는 Admin API를 통해 브로커를 cordoned로 표시하는 것입니다.
예를 들어 브로커 1에서 /data/dir1을 cordon하려면:
$ bin/kafka-configs.sh --bootstrap-server localhost:9092 --alter --add-config cordoned.log.dirs=/data/dir1 --entity-type brokers --entity-name 1
Completed updating config for broker 1.
그런 다음 폐기할 로그 디렉터리의 모든 파티션을 클러스터의 다른 로그 디렉터리나 브로커로 재배치합니다. 파티션 재배치 도구는 아직 로그 디렉터리 폐기를 위한 재배치 계획을 자동으로 생성하는 기능이 없습니다. 따라서 관리자는 폐기할 로그 디렉터리가 호스팅하는 모든 파티션의 복제본을 이동하는 재배치 계획을 만들어야 합니다.
모든 재배치가 완료되면 브로커를 종료합니다.
그런 다음 로그 디렉터리를 uncordon합니다. 그 디렉터리를 호스팅하는 브로커가 오프라인이므로 --bootstrap-controller를 사용해 수행하세요.
예를 들어:
$ bin/kafka-configs.sh --bootstrap-controller localhost:9093 --alter --delete-config cordoned.log.dirs --entity-type brokers --entity-name 1
Completed updating config for broker 1.
브로커 구성을 업데이트하고 폐기할 로그 디렉터리를 log.dir 또는 log.dirs에서 제거합니다.
예를 들어 브로커 구성이 다음과 같았다면:
log.dirs=/data/dir1,/data/dir2
다음과 같이 업데이트합니다.
log.dirs=/data/dir2
마지막으로 브로커를 재시작합니다.
복제 팩터 증가 (Increasing replication factor)
기존 파티션의 복제 팩터를 높이는 것은 쉽습니다. 커스텀 재배치 json 파일에 추가 복제본을 지정하고 --execute 옵션과 함께 사용해 지정된 파티션의 복제 팩터를 높이면 됩니다.
예를 들어 다음 예제는 토픽 foo의 파티션 0의 복제 팩터를 1에서 3으로 높입니다. 복제 팩터를 높이기 전에 파티션의 유일한 복제본은 브로커 5에 존재했습니다. 복제 팩터를 높이는 과정의 일부로 브로커 6과 7에 복제본을 추가하겠습니다.
첫 단계는 json 파일에 커스텀 재배치 계획을 직접 만드는 것입니다.
$ cat increase-replication-factor.json
{"version":1,
"partitions":[{"topic":"foo","partition":0,"replicas":[5,6,7]}]}
그런 다음 --execute 옵션과 함께 json 파일을 사용해 재배치 프로세스를 시작합니다.
$ bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --reassignment-json-file increase-replication-factor.json --execute
Current partition replica assignment
{"version":1,
"partitions":[{"topic":"foo","partition":0,"replicas":[5],"log_dirs":["any"]}]}
Save this to use as the --reassignment-json-file option during rollback
Successfully started partition reassignment for foo-0
--verify 옵션을 도구와 함께 사용해 파티션 재배치 상태를 확인할 수 있습니다. --execute 옵션과 함께 사용된 것과 동일한 increase-replication-factor.json을 --verify 옵션과 함께 사용해야 합니다.
$ bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --reassignment-json-file increase-replication-factor.json --verify
Status of partition reassignment:
Reassignment of partition [foo,0] is completed
kafka-topics.sh 도구로 복제 팩터 증가를 확인할 수도 있습니다.
$ bin/kafka-topics.sh --bootstrap-server localhost:9092 --topic foo --describe
Topic:foo PartitionCount:1 ReplicationFactor:3 Configs:
Topic: foo Partition: 0 Leader: 5 Replicas: 5,6,7 Isr: 5,6,7
데이터 마이그레이션 중 대역폭 사용 제한 (Limiting bandwidth usage during data migration)
Kafka를 사용하면 복제 트래픽에 스로틀(throttle)을 적용해 머신 간, 디스크 간에 복제본을 이동하는 데 사용되는 대역폭의 상한을 설정할 수 있습니다. 이는 클러스터 리밸런싱, 브로커 추가·제거, 디스크 추가·제거 시 유용하며, 이러한 데이터 집약적 작업이 사용자에게 미치는 영향을 제한합니다.
스로틀을 적용하는 데 사용할 수 있는 두 가지 인터페이스가 있습니다. 가장 간단하고 안전한 것은 kafka-reassign-partitions.sh를 호출할 때 스로틀을 적용하는 것이지만, kafka-configs.sh를 사용해 스로틀 값을 직접 보고 변경할 수도 있습니다.
예를 들어 리밸런스를 실행한다면 아래 명령으로 브로커 간에는 50MB/s 이하, 브로커의 디스크 간에는 100MB/s 이하로 파티션을 이동합니다.
$ bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --execute --reassignment-json-file bigger-cluster.json --throttle 50000000 --replica-alter-log-dirs-throttle 100000000
이 스크립트를 실행하면 스로틀이 적용되는 것을 볼 수 있습니다.
The inter-broker throttle limit was set to 50000000 B/s
The replica-alter-dir throttle limit was set to 100000000 B/s
Successfully started partition reassignment for foo1-0
리밸런스 중에 스로틀을 변경하고 싶다면, 예를 들어 더 빨리 완료되도록 브로커 간 처리량을 높이려면, 동일한 reassignment-json-file을 전달하며 --additional 옵션과 함께 execute 명령을 다시 실행하면 됩니다.
$ bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --additional --execute --reassignment-json-file bigger-cluster.json --throttle 700000000
The inter-broker throttle limit was set to 700000000 B/s
리밸런스가 완료되면 관리자는 --verify 옵션으로 리밸런스 상태를 확인할 수 있습니다. 리밸런스가 완료되면 --verify 명령으로 스로틀이 제거됩니다. 관리자는 --verify 옵션으로 명령을 실행해 리밸런싱 완료 후 적시에 스로틀을 제거하는 것이 중요합니다. 그렇게 하지 않으면 일반 복제 트래픽까지 스로틀될 수 있습니다.
--verify 옵션이 실행되고 재배치가 완료되면 스크립트는 스로틀이 제거되었음을 확인합니다.
$ bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --verify --reassignment-json-file bigger-cluster.json
Status of partition reassignment:
Reassignment of partition [my-topic,1] is completed
Reassignment of partition [my-topic,0] is completed
Clearing broker-level throttles on brokers 1,2,3
Clearing topic-level throttles on topic my-topic
관리자는 kafka-configs.sh로 할당된 구성을 검증할 수도 있습니다. 스로틀링 과정을 관리하는 데 사용되는 두 가지 스로틀 구성 집합이 있습니다. 첫 번째 집합은 스로틀 값 자체를 나타냅니다. 이는 브로커 수준에서 동적 속성으로 구성됩니다.
leader.replication.throttled.rate
follower.replication.throttled.rate
replica.alter.log.dirs.io.max.bytes.per.second
그다음 스로틀된 복제본의 열거된 집합의 구성 쌍이 있습니다.
leader.replication.throttled.replicas
follower.replication.throttled.replicas
이것들은 토픽별로 구성됩니다.
다섯 구성 값 모두 kafka-reassign-partitions.sh가 자동으로 할당합니다(아래에서 자세히 설명).
스로틀 한도 구성을 보려면:
$ bin/kafka-configs.sh --describe --bootstrap-server localhost:9092 --entity-type brokers
Configs for brokers '2' are leader.replication.throttled.rate=700000000,follower.replication.throttled.rate=700000000,replica.alter.log.dirs.io.max.bytes.per.second=1000000000
Configs for brokers '1' are leader.replication.throttled.rate=700000000,follower.replication.throttled.rate=700000000,replica.alter.log.dirs.io.max.bytes.per.second=1000000000
이는 복제 프로토콜의 리더와 팔로워 양쪽(기본적으로 양쪽에 동일한 스로틀 처리량 값이 할당됨)에 적용되는 스로틀과 디스크 스로틀을 보여줍니다.
스로틀된 복제본 목록을 보려면:
$ bin/kafka-configs.sh --describe --bootstrap-server localhost:9092 --entity-type topics
Configs for topic 'my-topic' are leader.replication.throttled.replicas=1:102,0:101,
follower.replication.throttled.replicas=1:101,0:102
여기서 리더 스로틀이 브로커 102의 파티션 1과 브로커 101의 파티션 0에 적용되는 것을 볼 수 있습니다. 마찬가지로 팔로워 스로틀이 브로커 101의 파티션 1과 브로커 102의 파티션 0에 적용됩니다.
기본적으로 kafka-reassign-partitions.sh는 리밸런스 전에 존재하는 모든 복제본(어느 것이든 리더가 될 수 있음)에 리더 스로틀을 적용합니다. 모든 이동 대상에는 팔로워 스로틀을 적용합니다. 브로커 101,102에 복제본이 있는 파티션이 102,103으로 재배치되면 그 파티션에 대한 리더 스로틀은 101,102에, 팔로워 스로틀은 103에만 적용됩니다.
필요하다면 kafka-configs.sh의 --alter 스위치를 사용해 스로틀 구성을 수동으로 변경할 수도 있습니다.
스로틀 복제의 안전한 사용 (Safe usage of throttled replication)
스로틀 복제를 사용할 때는 몇 가지 주의가 필요합니다. 특히:
(1) 스로틀 제거:
재배치가 완료되면 적시에 스로틀을 제거해야 합니다(bin/kafka-reassign-partitions.sh --verify 실행).
(2) 진행 보장:
들어오는 쓰기 속도와 비교해 스로틀이 너무 낮게 설정되면 복제가 진행되지 않을 수 있습니다. 이는 다음 경우에 발생합니다.
max(BytesInPerSec) > throttle
여기서 BytesInPerSec은 각 브로커에 프로듀서의 쓰기 처리량을 모니터링하는 지표입니다.
관리자는 리밸런스 중 다음 지표를 사용해 복제가 진행되는지 모니터링할 수 있습니다.
kafka.server:type=FetcherLagMetrics,name=ConsumerLag,clientId=([-.\w]+),topic=([-.\w]+),partition=([0-9]+)
복제 중 랙이 계속 감소해야 합니다. 지표가 감소하지 않으면 위에서 설명한 대로 관리자가 스로틀 처리량을 증가시켜야 합니다.
쿼터 설정 (Setting quotas)
쿼터 오버라이드와 기본값은 여기에 설명된 대로 (user, client-id), user 또는 client-id 수준에서 구성할 수 있습니다. 기본적으로 클라이언트는 무제한 쿼터를 받습니다. 각 (user, client-id), user 또는 client-id 그룹에 커스텀 쿼터를 설정하는 것이 가능합니다.
(user=user1, client-id=clientA)에 커스텀 쿼터 구성:
$ bin/kafka-configs.sh --bootstrap-server localhost:9092 --alter --add-config 'producer_byte_rate=1024,consumer_byte_rate=2048,request_percentage=200' --entity-type users --entity-name user1 --entity-type clients --entity-name clientA
Updated config for entity: user-principal 'user1', client-id 'clientA'.
user=user1에 커스텀 쿼터 구성:
$ bin/kafka-configs.sh --bootstrap-server localhost:9092 --alter --add-config 'producer_byte_rate=1024,consumer_byte_rate=2048,request_percentage=200' --entity-type users --entity-name user1
Updated config for entity: user-principal 'user1'.
client-id=clientA에 커스텀 쿼터 구성:
$ bin/kafka-configs.sh --bootstrap-server localhost:9092 --alter --add-config 'producer_byte_rate=1024,consumer_byte_rate=2048,request_percentage=200' --entity-type clients --entity-name clientA
Updated config for entity: client-id 'clientA'.
--entity-name 대신 --entity-default 옵션을 지정해 각 (user, client-id), user 또는 client-id 그룹에 기본 쿼터를 설정할 수 있습니다.
user=user1에 대한 기본 client-id 쿼터 구성:
$ bin/kafka-configs.sh --bootstrap-server localhost:9092 --alter --add-config 'producer_byte_rate=1024,consumer_byte_rate=2048,request_percentage=200' --entity-type users --entity-name user1 --entity-type clients --entity-default
Updated config for entity: user-principal 'user1', default client-id.
user에 대한 기본 쿼터 구성:
$ bin/kafka-configs.sh --bootstrap-server localhost:9092 --alter --add-config 'producer_byte_rate=1024,consumer_byte_rate=2048,request_percentage=200' --entity-type users --entity-default
Updated config for entity: default user-principal.
client-id에 대한 기본 쿼터 구성:
$ bin/kafka-configs.sh --bootstrap-server localhost:9092 --alter --add-config 'producer_byte_rate=1024,consumer_byte_rate=2048,request_percentage=200' --entity-type clients --entity-default
Updated config for entity: default client-id.
주어진 (user, client-id)의 쿼터를 describe하는 방법:
$ bin/kafka-configs.sh --bootstrap-server localhost:9092 --describe --entity-type users --entity-name user1 --entity-type clients --entity-name clientA
Configs for user-principal 'user1', client-id 'clientA' are producer_byte_rate=1024,consumer_byte_rate=2048,request_percentage=200
주어진 user의 쿼터 describe:
$ bin/kafka-configs.sh --bootstrap-server localhost:9092 --describe --entity-type users --entity-name user1
Configs for user-principal 'user1' are producer_byte_rate=1024,consumer_byte_rate=2048,request_percentage=200
주어진 client-id의 쿼터 describe:
$ bin/kafka-configs.sh --bootstrap-server localhost:9092 --describe --entity-type clients --entity-name clientA
Configs for client-id 'clientA' are producer_byte_rate=1024,consumer_byte_rate=2048,request_percentage=200
user의 기본 쿼터 describe:
$ bin/kafka-configs.sh --bootstrap-server localhost:9092 --describe --entity-type users --entity-default
Quota configs for the default user-principal are consumer_byte_rate=2048.0, request_percentage=200.0, producer_byte_rate=1024.0
client-id의 기본 쿼터 describe:
$ bin/kafka-configs.sh --bootstrap-server localhost:9092 --describe --entity-type clients --entity-default
Quota configs for the default client-id are consumer_byte_rate=2048.0, request_percentage=200.0, producer_byte_rate=1024.0
엔티티 이름이 지정되지 않으면 지정된 타입의 모든 엔티티가 describe됩니다. 예를 들어 모든 user를 describe:
$ bin/kafka-configs.sh --bootstrap-server localhost:9092 --describe --entity-type users
Configs for user-principal 'user1' are producer_byte_rate=1024,consumer_byte_rate=2048,request_percentage=200
Configs for default user-principal are producer_byte_rate=1024,consumer_byte_rate=2048,request_percentage=200
마찬가지로 (user, client)에 대해:
$ bin/kafka-configs.sh --bootstrap-server localhost:9092 --describe --entity-type users --entity-type clients
Configs for user-principal 'user1', default client-id are producer_byte_rate=1024,consumer_byte_rate=2048,request_percentage=200
Configs for user-principal 'user1', client-id 'clientA' are producer_byte_rate=1024,consumer_byte_rate=2048,request_percentage=200
더 알아보기 (Learn more)
- 파티션·복제 팩터는 토픽 도구로, 동적 리밸런싱은
kafka-reassign-partitions.sh로 다뤄요. - 컨슈머 위치 확인은
kafka-consumer-groups.sh --describe로 해요. - 데이터 마이그레이션 중엔 스로틀을 걸고, 완료 후
--verify로 제거해야 해요.