Kafka

동적 설정 변경 — 토픽 오버라이드

클러스터가 돌아가는 중에 토픽과 클라이언트 쿼터에 대한 설정을 오버라이드할 수 있다. 앞으로 더 많은 동적 설정이 추가될 예정이었기 때문에 이 변경들은 별도의 CLI 도구인 kafka-configs.sh로 분리되었다. 이 도구로 특정 토픽과 클라이언트 ID에 대한 설정을 지정할 수 있다. 한 번 설정하면 이 설정들은 클러스터에 영구적이다. Zookeeper에 저장되고, 각 브로커가 시작할 때 읽는다. 도구와 문서에서는 이런 동적 설정을 per-topic 또는 per-client 설정, 혹은 오버라이드라고 부른다.

kafka-configs.sh --zookeeper zoo1.example.com:2181/kafka-cluster \
  --alter --entity-type topics --entity-name <topic name> \
  --add-config <key>=<value>[,<key>=<value>...]

토픽에 쓸 수 있는 주요 키들:

- cleanup.policy: compact로 설정하면 이 토픽의 메시지는 같은 키에 대해 가장 최근 메시지만 남도록 버려진다(로그 압축) - compression.type: 브로커가 이 토픽의 메시지 배치를 디스크에 쓸 때 사용하는 압축 타입. gzip, snappy, lz4 - delete.retention.ms: 삭제 tombstone을 얼마나 보관할지(ms). 로그 압축 토픽에만 유효 - file.delete.delay.ms: 로그 세그먼트와 인덱스를 디스크에서 지우기 전 대기 시간(ms) - flush.messages: 몇 개의 메시지를 받으면 디스크로 강제 flush할지 - flush.ms: 몇 ms마다 디스크로 강제 flush할지 - index.interval.bytes: 로그 세그먼트 인덱스 엔트리 사이에 생산될 수 있는 메시지 바이트 수 - max.message.bytes: 이 토픽의 단일 메시지 최대 크기(바이트) - message.format.version: 브로커가 메시지를 디스크에 쓸 때 사용할 메시지 포맷 버전. 유효한 API 버전 번호여야 한다(예: "0.10.0") - message.timestamp.difference.max.ms: 메시지를 받았을 때 메시지 타임스탬프와 브로커 타임스탬프의 최대 허용 차이(ms). message.timestamp.type이 CreateTime일 때만 유효 - message.timestamp.type: 디스크에 쓸 때 어느 타임스탬프를 쓸지. 클라이언트가 지정한 CreateTime 또는 브로커가 파티션에 쓴 시각인 LogAppendTime - min.cleanable.dirty.ratio: 로그 컴팩터가 이 토픽의 파티션을 얼마나 자주 압축할지를, 압축되지 않은 세그먼트 수 대 전체 세그먼트 수의 비율로 표현. 로그 압축 토픽에만 유효 - min.insync.replicas: 이 토픽의 파티션이 가용하다고 간주되기 위해 in-sync여야 하는 최소 복제본 수 - preallocate: true면 새 세그먼트가 롤될 때 로그 세그먼트를 미리 할당 - retention.bytes: 이 토픽에 보관할 메시지 양(바이트) - retention.ms: 이 토픽의 메시지를 얼마나 오래 보관할지(ms) - segment.bytes: 파티션의 단일 로그 세그먼트에 쓸 메시지 양(바이트) - segment.index.bytes: 단일 로그 세그먼트 인덱스의 최대 크기(바이트) - segment.jitter.ms: 로그 세그먼트를 롤할 때 segment.ms에 무작위로 더해지는 최대 ms - segment.ms: 각 파티션의 로그 세그먼트를 몇 ms마다 회전시킬지 - unclean.leader.election.enable: false면 이 토픽에 대해 unclean 리더 선출이 허용되지 않는다

# my-topic의 보관 기간을 1시간(3,600,000ms)으로 설정
kafka-configs.sh --zookeeper zoo1.example.com:2181/kafka-cluster \
  --alter --entity-type topics --entity-name my-topic \
  --add-config retention.ms=3600000

클라이언트 쿼터는 브로커 '한 대당' 적용된다

Kafka 클라이언트에 대해 오버라이드할 수 있는 설정은 프로듀서·컨슈머 쿼터뿐이다. 둘 다 초당 바이트 단위의 속도이며, 지정된 클라이언트 ID를 가진 모든 클라이언트가 브로커 '한 대당' 생산하거나 소비할 수 있는 양이다.

이 '브로커 단위'가 실무에서 자주 놓치는 지점이다. 클러스터에 브로커가 5대 있고 어떤 클라이언트에 프로듀서 쿼터 10MB/sec를 지정하면, 그 클라이언트는 각 브로커에서 동시에 10MB/sec를 생산할 수 있으므로 총합은 50MB/sec가 된다. 쿼터를 정할 때 이 곱셈을 계산에 넣지 않으면 의도한 것보다 훨씬 큰 트래픽을 허용하게 된다.

kafka-configs.sh --zookeeper zoo1.example.com:2181/kafka-cluster \
  --alter --entity-type clients --entity-name <client ID> \
  --add-config <key>=<value>[,<key>=<value>...]

클라이언트용 설정 키는 두 개다. - producer_bytes_rate: 하나의 클라이언트 ID가 1초에 브로커 한 대로 생산할 수 있는 메시지 양(바이트) - consumer_bytes_rate: 하나의 클라이언트 ID가 1초에 브로커 한 대에서 소비할 수 있는 메시지 양(바이트)

클라이언트 ID는 컨슈머 그룹 이름과 같지 않다는 점도 주의해야 한다. 컨슈머는 자기 클라이언트 ID를 스스로 설정할 수 있고, 서로 다른 그룹의 여러 컨슈머가 같은 클라이언트 ID를 지정할 수도 있다. 각 컨슈머 그룹의 클라이언트 ID를 그 그룹을 식별하는 고유한 값으로 설정하는 것이 모범 사례다. 그러면 하나의 컨슈머 그룹이 쿼터를 공유하게 되고, 로그에서 어떤 그룹이 요청을 일으켰는지 파악하기도 쉬워진다.

설정 오버라이드는 --describe로 조회한다.

# kafka-configs.sh --zookeeper zoo1.example.com:2181/kafka-cluster \
  --describe --entity-type topics --entity-name my-topic
Configs for topics:my-topic are retention.ms=3600000,segment.ms=3600000

중요한 제약이 있다. 설정 describe는 '오버라이드만' 보여 주며 클러스터 기본 설정은 포함하지 않는다. 그리고 (이 책 시점에는) Zookeeper로도 Kafka 프로토콜로도 브로커 자체의 설정을 동적으로 알아낼 방법이 없다. 즉 이 도구로 토픽·클라이언트 설정을 자동화에서 조회할 때, 도구가 클러스터 기본 설정을 별도로 알고 있어야 한다.

오버라이드를 완전히 제거하면 해당 엔티티는 클러스터 기본값으로 되돌아간다. --alter와 --delete-config 파라미터를 함께 쓴다.

kafka-configs.sh --zookeeper zoo1.example.com:2181/kafka-cluster \
  --alter --entity-type topics --entity-name my-topic \
  --delete-config retention.ms

preferred replica election — 재시작한 브로커는 리더를 자동으로 되찾지 않는다

파티션은 신뢰성을 위해 여러 복제본을 가질 수 있지만, 그중 하나만 리더가 되고 모든 produce·consume 연산이 그 브로커에서 일어난다. Kafka 내부에서 리더는 '복제본 목록의 첫 번째 in-sync 복제본'으로 정의된다. 그런데 브로커가 정지되었다가 재시작되면 어떤 파티션의 리더십도 자동으로 되찾지 않는다.

브로커가 리더십을 되찾게 하는 방법 중 하나가 preferred replica election을 트리거하는 것이다. 이는 클러스터 컨트롤러에게 파티션들의 이상적인 리더를 선택하라고 지시한다. 클라이언트가 리더십 변경을 자동으로 추적하므로 이 작업은 대체로 영향이 없다. kafka-preferred-replica-election.sh 유틸리티로 수동으로 지시할 수 있다.

브로커 설정에 자동 리더 리밸런싱 옵션이 있지만 프로덕션 사용은 권장되지 않는다. 자동 밸런싱 모듈은 성능에 상당한 영향을 주고, 큰 클러스터에서는 클라이언트 트래픽을 오래 멈추게 할 수 있다.

# kafka-preferred-replica-election.sh --zookeeper zoo1.example.com:2181/kafka-cluster
Successfully started preferred replica election for partitions
Set([my-topic,5], [my-topic,0], [my-topic,7], [my-topic,4],
[my-topic,6], [my-topic,2], [my-topic,3], [my-topic,1])

파티션이 아주 많은 클러스터에서는 한 번의 preferred replica election이 실행되지 못할 수 있다. 요청이 클러스터 메타데이터 안의 Zookeeper znode에 기록되어야 하는데, 요청이 znode 크기 한계(기본 1MB)보다 크면 실패한다. 이 경우에는 선출 대상 파티션을 나열한 JSON 객체 파일을 만들어 요청을 여러 단계로 쪼개야 한다.

{
    "partitions": [
        { "partition": 1, "topic": "foo" },
        { "partition": 2, "topic": "foobar" }
    ]
}

kafka-preferred-replica-election.sh --zookeeper zoo1.example.com:2181/kafka-cluster \
  --path-to-json-file partitions.json

파티션 재할당 — generate / execute / verify 3단계

파티션의 복제본 할당을 바꿔야 할 때가 있다. 토픽의 파티션이 클러스터에 고르게 분포하지 않아 브로커 부하가 불균형할 때, 브로커가 오프라인이 되어 파티션이 under-replicated일 때, 새 브로커를 추가해 클러스터 부하를 나눠 받아야 할 때다.

kafka-reassign-partitions.sh가 이 작업을 한다. 이 도구는 최소 두 단계로 사용해야 한다. 첫 단계는 브로커 목록과 토픽 목록으로 이동 계획(set of moves)을 생성한다. 두 번째 단계는 생성된 이동을 실행한다. 선택적인 세 번째 단계는 생성된 목록으로 재할당의 진행 상황이나 완료 여부를 검증한다.

이동 계획을 생성하려면 토픽을 나열한 JSON 객체 파일이 필요하다(version 번호는 현재 항상 1이다).

{
    "topics": [ { "topic": "foo" }, { "topic": "foo1" } ],
    "version": 1
}

kafka-reassign-partitions.sh --zookeeper zoo1.example.com:2181/kafka-cluster \
  --generate --topics-to-move-json-file topics.json --broker-list 0,1

브로커 목록은 명령행에 콤마로 구분된 브로커 ID 목록으로 준다. 도구는 표준 출력으로 두 개의 JSON 객체를 낸다. 하나는 토픽들의 현재 파티션 할당, 다른 하나는 제안된 파티션 할당이다. 첫 번째 JSON 객체는 재할당을 되돌려야 할 경우를 대비해 저장해 둔다. 두 번째(제안된 할당)를 새 파일로 저장해, 이것을 2단계에서 도구에 다시 넘긴다.

kafka-reassign-partitions.sh --zookeeper zoo1.example.com:2181/kafka-cluster \
  --execute --reassignment-json-file reassign.json

실행하면 지정한 파티션 복제본들의 새 브로커로의 재할당이 시작된다. 클러스터 컨트롤러는 각 파티션의 복제본 목록에 새 복제본을 '추가'하는 방식으로 이 작업을 수행한다(리플리케이션 팩터가 일시적으로 늘어난다). 새 복제본들은 현재 리더에게서 각 파티션의 기존 메시지를 모두 복사한다. 디스크상의 파티션 크기에 따라 네트워크를 통한 복사에 상당한 시간이 걸릴 수 있다. 복제가 끝나면 컨트롤러가 복제본 목록에서 옛 복제본을 제거한다(리플리케이션 팩터가 원래 크기로 돌아간다).

네트워크 활용을 개선하는 요령이 있다. 한 브로커에서 많은 파티션을 제거할 때(예: 그 브로커를 클러스터에서 빼는 경우), 재할당을 시작하기 전에 그 브로커를 셧다운했다가 재시작하는 것이 모범 사례다. 그러면 (자동 리더 선출이 켜져 있지 않은 한) 그 브로커의 파티션 리더십이 다른 브로커로 옮겨 간다. 이렇게 하면 복제 트래픽이 여러 브로커로 분산되어 재할당 성능이 크게 오르고 클러스터에 미치는 영향이 줄어든다.

재할당이 진행되는 동안과 완료 후에는 --verify로 상태를 확인할 수 있다. 진행 중인 것, 완료된 것, 오류가 있었다면 실패한 것을 보여 준다. 이를 위해서는 execute 단계에서 쓴 JSON 파일이 있어야 한다.

kafka-reassign-partitions.sh --zookeeper zoo1.example.com:2181/kafka-cluster \
  --verify --reassignment-json-file reassign.json

파티션 재할당은 메모리 페이지 캐시의 일관성을 바꾸고 네트워크·디스크 I/O를 쓰기 때문에 클러스터 성능에 큰 영향을 준다. 재할당을 여러 개의 작은 단계로 쪼개는 것이 영향을 최소화하는 좋은 방법이다.

리플리케이션 팩터 변경. 재할당 도구에는 파티션의 리플리케이션 팩터를 늘리거나 줄일 수 있는 문서화되지 않은 기능이 있다. 토픽 생성 시 브로커가 충분하지 않아 잘못된 리플리케이션 팩터로 만들어진 경우 같은 상황에서 필요하다. execute 단계에서 쓰는 형식의 JSON 객체를 만들되, replicas 목록에 복제본을 추가하거나 제거해 리플리케이션 팩터를 맞추면 된다. 예를 들어 replicas가 [1]인 파티션에 대해 [1,2]를 주면 리플리케이션 팩터가 2로 올라간다. 클러스터는 재할당을 완료하고 새 크기의 리플리케이션 팩터를 유지한다.

로그 세그먼트 덤프와 레플리카 검증

컨슈머가 처리하지 못하는 'poison pill' 메시지가 토픽에 들어갔을 때처럼 특정 메시지의 내용을 직접 봐야 할 때가 있다. DumpLogSegments 도구가 파티션의 로그 세그먼트를 디코드해, 메시지를 소비·디코드하지 않고도 개별 메시지를 볼 수 있게 해 준다. 도구는 콤마로 구분된 로그 세그먼트 파일 목록을 인자로 받고, 메시지 요약 정보 또는 상세 메시지 데이터를 출력한다.

# 요약만
kafka-run-class.sh kafka.tools.DumpLogSegments --files 00000000000052368601.log

# 페이로드까지
kafka-run-class.sh kafka.tools.DumpLogSegments --files 00000000000052368601.log \
  --print-data-log

출력에는 offset, position, 타임스탬프 타입, isvalid, payloadsize, magic, compresscodec, crc가 나오고, --print-data-log를 주면 payload까지 함께 출력된다.

이 도구로 로그 세그먼트에 딸린 인덱스 파일을 검증할 수도 있다. 인덱스는 로그 세그먼트 안에서 메시지를 찾는 데 쓰이며, 손상되면 소비에서 오류를 일으킨다. 검증은 브로커가 unclean 상태로(정상적으로 종료되지 않은 채) 시작할 때마다 자동 수행되지만 수동으로도 할 수 있다. 확인 옵션은 검사 강도에 따라 두 가지다. --index-sanity-check는 인덱스가 사용 가능한 상태인지만 확인하고, --verify-index-only는 인덱스 엔트리를 모두 출력하지 않으면서 불일치를 검사한다.

kafka-run-class.sh kafka.tools.DumpLogSegments \
  --files 00000000000052368601.index,00000000000052368601.log --index-sanity-check

레플리카 검증. 파티션 복제는 일반 Kafka 컨슈머 클라이언트와 비슷하게 동작한다. 팔로워 브로커는 가장 오래된 오프셋부터 복제를 시작하고 현재 오프셋을 주기적으로 디스크에 체크포인트한다. 복제가 멈췄다 재개되면 마지막 체크포인트에서 이어 간다. 이전에 복제된 로그 세그먼트가 브로커에서 삭제될 수 있는데, 이 경우 팔로워는 그 빈틈을 메우지 않는다.

토픽 파티션들의 복제본이 클러스터 전체에서 동일한지 검증하려면 kafka-replica-verification.sh를 쓴다. 이 도구는 주어진 토픽 파티션 집합의 모든 복제본에서 메시지를 가져와 모든 메시지가 모든 복제본에 존재하는지 확인한다. 검증할 토픽을 매치하는 정규식을 제공해야 하며, 제공하지 않으면 모든 토픽이 검증된다. 접속할 브로커 목록도 명시적으로 제공해야 한다.

주의할 점은 이 도구가 파티션 재할당과 비슷한 수준의 클러스터 영향을 준다는 것이다. 복제본을 검증하려면 가장 오래된 오프셋부터 모든 메시지를 읽어야 하고, 게다가 한 파티션의 모든 복제본을 병렬로 읽는다. 따라서 신중하게 사용해야 한다.

kafka-replica-verification.sh --broker-list \
  kafka1.example.com:9092,kafka2.example.com:9092 --topic-white-list 'my-.*'

콘솔 컨슈머와 콘솔 프로듀서

애플리케이션 상태를 확인하기 위해 메시지를 직접 소비하거나 샘플 메시지를 생산해야 할 때가 있다. kafka-console-consumer.sh와 kafka-console-producer.sh가 그 용도다. 이들은 Java 클라이언트 라이브러리를 감싼 래퍼로, 애플리케이션을 작성하지 않고도 Kafka 토픽과 상호작용하게 해 준다.

다만 콘솔 컨슈머·프로듀서를 다른 애플리케이션에 파이프하는 식으로 감싸는 것은 피해야 한다. 이런 애플리케이션은 매우 취약하다. 콘솔 컨슈머와 메시지를 잃지 않는 방식으로 상호작용하기가 어렵고, 콘솔 프로듀서는 모든 기능을 쓸 수 없으며 바이트를 제대로 보내는 것도 까다롭다. Java 클라이언트 라이브러리를 직접 쓰거나, Kafka 프로토콜을 직접 사용하는 서드파티 클라이언트를 쓰는 편이 낫다. 또한 콘솔 컨슈머는 클러스터와 같은 버전을 써야 한다. 구버전 콘솔 컨슈머는 Zookeeper와 잘못된 방식으로 상호작용해 클러스터를 손상시킬 수 있다.

콘솔 컨슈머. 메시지는 개행으로 구분되어 표준 출력에 찍히며, 기본적으로 포맷 없이 메시지의 원시 바이트를 출력한다(DefaultFormatter). 먼저 새 컨슈머를 쓸지 지정하고 클러스터를 가리켜야 한다. 구버전 컨슈머는 --zookeeper 뒤에 연결 문자열만 주면 되고, 새 컨슈머는 --new-consumer 플래그와 --broker-list를 함께 줘야 한다.

소비할 토픽을 지정하는 옵션은 --topic, --whitelist, --blacklist 세 가지이며 이 중 하나만 쓸 수 있다. --topic은 단일 토픽을 지정한다. --whitelist와 --blacklist는 각각 정규식을 받는다(셸이 정규식을 바꾸지 않도록 제대로 이스케이프해야 한다). whitelist는 정규식에 매치되는 모든 토픽을 소비하고, blacklist는 매치되는 것을 '제외한' 모든 토픽을 소비한다.

일반 컨슈머 설정도 넘길 수 있다. --consumer.config CONFIGFILE로 설정 파일을 주거나, --consumer-property KEY=VALUE 형태로 명령행에 직접 준다. 컨슈머 그룹 ID 설정 같은 데 유용하다.

혼동하기 쉬운 지점이 있다. 콘솔 컨슈머와 콘솔 프로듀서 모두 --property 옵션이 있지만, 이것은 --consumer-property / --producer-property와 다르다. --property는 '메시지 포맷터'에 설정을 넘기는 용도이며 클라이언트 자체에 넘기는 것이 아니다.

자주 쓰는 콘솔 컨슈머 옵션: - --formatter CLASSNAME: 메시지를 디코드할 포맷터 클래스. 기본값 kafka.tools.DefaultFormatter - --from-beginning: 가장 오래된 오프셋부터 소비. 지정하지 않으면 최신 오프셋부터 시작 - --max-messages NUM: 최대 NUM개를 소비하고 종료 - --partition NUM: ID가 NUM인 파티션에서만 소비(새 컨슈머 필요)

기본 외의 포맷터는 세 가지다. kafka.tools.LoggingMessageFormatter는 표준 출력 대신 로거로 출력하며 INFO 레벨로 타임스탬프·키·값을 찍는다. kafka.tools.ChecksumMessageFormatter는 메시지 체크섬만 출력한다. kafka.tools.NoOpMessageFormatter는 소비하지만 아무것도 출력하지 않는다.

kafka.tools.DefaultMessageFormatter는 --property로 넘길 수 있는 옵션이 있다: print.timestamp, print.key, key.separator, line.separator, key.deserializer, value.deserializer. deserializer 클래스는 org.apache.kafka.common.serialization.Deserializer를 구현해야 하고, 콘솔 컨슈머는 그 결과에 toString을 호출해 출력한다. 보통 Java 클래스로 구현한 뒤 콘솔 컨슈머 실행 전에 CLASSPATH 환경 변수를 설정해 클래스패스에 넣는다.

오프셋 토픽 소비. 클러스터의 컨슈머 그룹들이 어떤 오프셋을 커밋하는지 보고 싶을 때가 있다. 특정 그룹이 오프셋을 커밋하고 있는지, 얼마나 자주 커밋하는지 확인하는 경우다. 콘솔 컨슈머로 내부 토픽 __consumer_offsets를 소비하면 된다. 모든 컨슈머 오프셋이 이 토픽에 메시지로 기록된다. 이 토픽의 메시지를 디코드하려면 포맷터 클래스 kafka.coordinator.GroupMetadataManager$OffsetsMessageFormatter를 써야 한다.

콘솔 프로듀서. 기본적으로 한 줄에 하나씩 메시지를 읽으며, 탭 문자로 키와 값을 구분한다(탭이 없으면 키는 null이다). 최소 두 인자가 필요하다. --broker-list는 hostname:port 항목을 콤마로 구분한 목록이고, --topic은 생산할 토픽이다. 다 끝내면 EOF 문자를 보내 클라이언트를 닫는다.

프로듀서 설정도 --producer.config CONFIGFILE 또는 --producer-property KEY=VALUE로 넘길 수 있다. linger.ms나 batch.size 같은 배치 설정에 유용하다.

유용한 콘솔 프로듀서 옵션: - --key-serializer CLASSNAME / --value-serializer CLASSNAME: 키·값 직렬화 클래스. 기본값 kafka.serializer.DefaultEncoder - --compression-codec STRING: none, gzip, snappy, lz4 중 하나. 기본값은 gzip - --sync: 메시지를 동기로 생산하며, 각 메시지가 확인될 때까지 기다렸다 다음을 보낸다

표준 입력을 읽어 프로듀서 레코드를 만드는 kafka.tools.LineMessageReader에도 --property로 넘길 옵션이 있다. - ignore.error: parse.key가 true인데 키 구분자가 없을 때 예외를 던지려면 false로. 기본값 true - parse.key: 항상 키를 null로 두려면 false로. 기본값 true - key.separator: 읽을 때 키와 값 사이의 구분 문자. 기본값은 탭

LineMessageReader는 입력을 key.separator가 '처음' 나오는 지점에서 자른다. 그 뒤에 남은 문자가 없으면 메시지 값은 비게 된다. 줄에 키 구분자가 없거나 parse.key가 false면 키는 null이 된다.

위험한 작업 — Zookeeper를 직접 건드리는 최후 수단

기술적으로는 가능하지만 극단적인 상황이 아니면 시도해서는 안 되는 관리 작업들이 있다. 문제를 진단하다 다른 선택지가 없거나, 특정 버그를 임시로 우회해야 할 때다. 이 작업들은 대개 문서화되어 있지 않고 지원되지 않으며 애플리케이션에 어느 정도 위험을 준다. 이 절의 작업들은 Zookeeper에 저장된 클러스터 메타데이터를 직접 다루므로 매우 위험하다.

클러스터 컨트롤러 옮기기. 모든 Kafka 클러스터에는 컨트롤러가 있고, 이는 브로커 중 하나 안에서 도는 스레드다. 컨트롤러는 클러스터 연산을 총괄하며, 때때로 컨트롤러를 다른 브로커로 강제로 옮기는 것이 바람직할 때가 있다. 컨트롤러가 예외 등의 문제를 겪어 살아 있지만 제대로 동작하지 않는 경우가 그 예다. 이 상황에서 컨트롤러를 옮기는 것은 위험이 높지 않지만 일상적인 작업은 아니다.

현재 컨트롤러인 브로커는 클러스터 경로 최상위의 /controller라는 Zookeeper 노드에 자신을 등록한다. 이 Zookeeper 노드를 수동으로 삭제하면 현재 컨트롤러가 사임하고 클러스터가 새 컨트롤러를 선출한다.

파티션 이동 중단하기. 파티션 재할당의 정상 흐름은 다음과 같다. 1. 재할당이 요청된다(Zookeeper 노드가 생성됨) 2. 클러스터 컨트롤러가 추가되는 새 브로커에 파티션을 추가한다 3. 새 브로커가 각 파티션을 in-sync가 될 때까지 복제한다 4. 클러스터 컨트롤러가 파티션 복제본 목록에서 옛 브로커를 제거한다

요청 시 모든 재할당이 병렬로 시작되므로, 진행 중인 재할당을 취소하려 할 이유는 보통 없다. 예외는 재할당 도중 브로커가 죽어 즉시 재시작할 수 없는 경우다. 그러면 절대 끝나지 않는 재할당이 남아, 추가 재할당(예: 죽은 브로커에서 파티션을 빼 다른 브로커에 할당하는 작업)을 시작할 수 없게 된다. 이런 경우 클러스터가 기존 재할당을 잊게 만들 수 있다. 1. Kafka 클러스터 경로에서 /admin/reassign_partitions Zookeeper 노드를 제거한다 2. 컨트롤러 이동을 강제한다

주의할 점이 있다. 진행 중인 파티션 이동을 제거하면, 아직 완료되지 않은 파티션들은 '옛 브로커를 복제본 목록에서 제거하는' 단계를 거치지 않는다. 즉 일부 파티션의 리플리케이션 팩터가 의도보다 커질 수 있다. 브로커는 리플리케이션 팩터가 일관되지 않은 파티션을 가진 토픽에 대해 일부 관리 작업(예: 파티션 증가)을 허용하지 않는다. 따라서 진행 중이던 파티션들을 검토해 또 한 번의 재할당으로 리플리케이션 팩터를 바로잡아야 한다.

삭제 대기 토픽 요청 제거하기. 명령행 도구로 토픽을 삭제하면 삭제를 요청하는 Zookeeper 노드가 생성된다. 정상적으로는 클러스터가 즉시 실행한다. 그런데 명령행 도구는 클러스터에서 토픽 삭제가 활성화되어 있는지 알 방법이 없다. 그래서 어쨌든 삭제를 요청하며, 삭제가 비활성화되어 있으면 나중에 놀라게 된다. 이 대기 중인 요청들을 지울 수 있다. 토픽 삭제 요청은 /admin/delete_topic 아래에 토픽 이름을 가진 자식 Zookeeper 노드로 만들어진다. 이 자식 노드들을 삭제하면(부모 /admin/delete_topic 노드는 지우지 않는다) 대기 요청이 제거된다.

토픽 수동 삭제. 토픽 삭제가 비활성화된 클러스터를 운영 중이거나 정상 흐름 밖에서 토픽을 삭제해야 한다면 수동 삭제가 가능하다. 다만 클러스터의 모든 브로커를 완전히 셧다운해야 하며, 브로커가 하나라도 돌고 있으면 할 수 없다. 클러스터가 온라인인 상태에서 Zookeeper의 클러스터 메타데이터를 수정하는 것은 매우 위험하며 클러스터를 불안정한 상태로 만들 수 있다. 1. 클러스터의 모든 브로커를 셧다운한다 2. Kafka 클러스터 경로에서 /brokers/topics/TOPICNAME Zookeeper 경로를 제거한다. 이 노드에는 먼저 지워야 할 자식 노드들이 있다 3. 각 브로커의 로그 디렉터리에서 파티션 디렉터리를 제거한다. 이름은 TOPICNAME-NUM 형식이며 NUM은 파티션 ID다 4. 모든 브로커를 재시작한다

클라이언트 ACL은 kafka-acls.sh 도구로 다룬다. ACL과 보안에 대한 추가 문서는 Apache Kafka 웹사이트에 있다.