MQTT 브로커(EMQX)의 Kafka 브리지를 만들 때 연결 시험에서 다음이 나왔다.
Create action failed, Please check and retry:
{disconnected,#{error => no_connected_partition_leader,
kafka_client => <<"_probe_:kafka_producer:probedryrun">>,
kafka_topic => "order-events",
partitions_limit => 3}}
프로듀서는 메시지를 보내기 전에 메타데이터를 받아 각 파티션의 리더 브로커에 직접 연결한다. 이 오류는 메타데이터는 받았으나 리더 브로커와 연결을 맺지 못했다는 뜻이다. 부트스트랩 접속은 성공했는데 그다음 단계에서 막힌 것이므로, 원인은 거의 항상 부트스트랩 주소가 아니라 advertised.listeners 쪽에 있다.
grep -E '^(listeners|advertised.listeners)' /opt/kafka/config/server.properties
advertised.listeners=PLAINTEXT://0.0.0.0:9092 는 잘못된 설정이다. 클라이언트가 메타데이터로 0.0.0.0 을 받게 되어 연결할 수 없다. listeners 에는 바인딩 주소로 0.0.0.0 을 쓸 수 있지만, advertised.listeners 에는 클라이언트가 실제로 접속할 수 있는 이름을 적어야 한다.
listeners=PLAINTEXT://0.0.0.0:9092
advertised.listeners=PLAINTEXT://broker01.example.com:9092
클라이언트 쪽에서 그 이름이 해석되고 접속되는지 확인한다.
kafka-broker-api-versions.sh --bootstrap-server broker01.example.com:9092 | head
getent hosts broker01.example.com
nc -vz broker01.example.com 9092
컨테이너·Kubernetes·NAT 를 거치는 구성에서는 내부용과 외부용 리스너를 따로 두고 listener.security.protocol.map 으로 이름을 붙인다.
브리지가 지정한 토픽이 없고 자동 생성도 꺼져 있으면 리더가 있을 수 없다.
kafka-topics.sh --bootstrap-server broker01.example.com:9092 --describe --topic order-events
partitions_limit => 3 은 브리지가 파티션 3 개까지 다루겠다는 자체 설정값이며, 실제 파티션 수와 다를 수 있다.
kafka-topics.sh --bootstrap-server broker01.example.com:9092 --describe --unavailable-partitions
kafka-topics.sh --bootstrap-server broker01.example.com:9092 --describe --under-replicated-partitions
Leader: none 또는 Leader: -1 이면 그 파티션의 복제본이 전부 죽은 것이다. 해당 브로커를 살려야 한다.
보안 리스너를 쓰는데 브리지가 PLAINTEXT 로 접속하면 부트스트랩부터 실패하지만, 인증은 통과하고 ACL 에서 막히면 이런 형태로 나타나기도 한다. 프로듀서에게는 대상 토픽의 WRITE 와 DESCRIBE 권한이 필요하다.
kafka-acls.sh --bootstrap-server broker01.example.com:9092 --list --topic order-events
부트스트랩 포트만 열고 나머지 브로커 포트를 막아 둔 경우가 흔하다. 모든 브로커의 리스너 포트가 클라이언트에서 열려 있어야 한다.
같은 대화에서 나온 물음이라 함께 적는다. Kafka 헤더는 키가 문자열, 값이 바이트 배열인 목록이며 같은 키를 여러 번 넣을 수 있다. 무엇을 넣어야 한다는 규칙은 없고, 본문을 건드리지 않고 넘기고 싶은 메타데이터를 담는다. 흔히 쓰는 것은 발신 시스템 이름, 스키마 버전, 내용 형식, 추적 ID 다.
source = emqx-bridge
content-type = application/json
schema.version = 3
trace.id = 8f2b...
컨슈머가 헤더를 읽지 않도록 만들어져 있으면 헤더에 담은 정보는 그대로 버려지므로, 소비 측과 먼저 합의한다.