Kafka는 Kafka 클러스터의 메타데이터를 관리하는 Controller와 파티션 복제, 리더-팔로워 간 메세지를 동기화는 Broker, 메세지를 송/수신 하는 Client와 통신한다.

어떻게 통신하는 걸까?
참고로 현 상황은 EC2 인스턴스를 3개 실행하고 각 인스턴스에 Kafka를 Binary로 실행해서 클러스터를 만들었다. broker1의 server.properties의 설정을 보면서 통신의 흐름을 알아보고자 한다.

server.properties

process.roles=broker,controller
node.id=1

controller.quorum.voters=1@<broker1프라이빗IP>:19093,2@<broker2프라이빗IP>:29093,3@<broker3프라이빗IP>:39093
controller.listener.names=CONTROLLER

# Listeners
listeners=PLAINTEXT://0.0.0.0:9092,INTERNAL://<broker1프라이빗IP>:19092,CONTROLLER://0.0.0.0:19093
advertised.listeners=PLAINTEXT://<broker1퍼블릭IP>:9092,INTERNAL://<broker1프라이빗IP>:19092

# Listener roles
inter.broker.listener.name=INTERNAL
listener.security.protocol.map=PLAINTEXT:PLAINTEXT,INTERNAL:PLAINTEXT,CONTROLLER:PLAINTEXT

log.dirs=/opt/kafka/kraft-logs/kafka-logs1

auto.create.topics.enable=True

통신 흐름

컨트롤러 통신(Controller Quorum)

컨트롤러는 토픽, 파티션, 리더 선출 등 Kafka 클러스터의 메타데이터를 관리하며 브로커들과 통신한다.
포트는 브로커마다 다르고, 원하는대로 설정할 수 있다. 위의 server.properties에서는 19093,29093,39093등으로 설정되어 있다.

컨트롤러 통신을 좀 더 자세히 알아보자면 아래와 같다.
KRaft 모드에서는 여러 컨트롤러 중 1개만 리더 컨트롤러로 선출되어, Kafka 클러스터의 메타데이터(토픽, 파티션, ISR 등)를 관리하고, 이 리더는 controller.quorum.voters로 구성된 쿼럼 안에서 Raft 방식으로 선출된다.
여기서는 컨트롤러 후보가 3개가 되고 컨트롤러 쿼럼을 통해 3개의 노드가 서로 통신하면서 리더를 선출한다. 선출된 리더는 active controller가 되며 클러스터 메타데이터를 관리하고 나머지 2개는 passive 상태로 리더 장애 시 대체가 가능하도록 대기하게 된다.

listeners에서 CONTROLLER://0.0.0.0:19093를 선언하였지만 실제로는 kafka-brokers라는 EC2의 보안 그룹 인바운드 규칙에서 19093-39093 등의 포트를 프라이빗 IP 대역에 대해서만 허용하여, 외부 접근 없이 클러스터 내부에서만 CONTROLLER 포트 통신이 가능하도록 설정하였다.

브로커 간 통신(Replication, Fetching)

브로커 간의 통신은 파티션을 복제하고, 리더-팔로워 간 메세지를 동기화하기 위한 것이다. 위에서는 INTERNAL에 관한 통신이 이에 해당한다.

클라이언트 통신(Producer/Consumer)

클라이언트 통신은 메세지를 전송하거나 수신하기 위한 것이다. 따라서 Kafka 브로커는 외부시스템, 여기서는 Airflow가 포함된 docker compose와 통신한다.
포트는 통상 9092로 사용되며 Producer와 Consumer에 bootstrap_servers에 지정된 IP로 연결하고 브로커는 다른 클라이언트에게 "다른 브로커는 이 주소야"라고 알려준다.

• listeners=PLAINTEXT://0.0.0.0:9092: Kafka 브로커가 수신을 열어두는 포트와 인터페이스, 모든 네트워크 인터페이스(IP 주소)에 대해 9092 포트를 열어두고 서버가 9092 포트로 수신을 받을 준비를 했다는 의미이다.

• advertised.listeners=PLAINTEXT://<퍼블릭IP>:9092: Kafka가 클라이언트에게 알려주는 내 주소, 클라이언트는 이 주소를 보고 접속을 시도한다. bootstrap_servers에 연결하는 주소가 곧 advertised.listeners에 설정한 값이 된다.

기타 설정

log.dirs=/opt/kafka/kraft-logs/kafka-logs1

Kafka의 기본 설정은 log.dirs=/tmp/kafka-logs로, 임시 디렉토리인 /tmp에 로그 파일을 저장하도록 되어 있다.
하지만 /tmp는 인스턴스가 재시작되면 내용이 사라질 수 있기 때문에, Kafka가 재시작 후에도 데이터를 유지하려면 영속적인 디스크 경로를 설정해야 한다.

KRaft 모드에서는 클러스터를 처음 생성할 때, log.dirs로 지정된 경로에 메타데이터를 포맷하는 과정이 반드시 필요하다. 이 경로에 저장되는 메타데이터는 KRaft 모드의 핵심 구성 요소이므로, /tmp처럼 휘발성 디렉토리를 사용할 경우 클러스터를 정상적으로 운영할 수 없다.

내 경우 Kafka를 /opt/ 하위 디렉토리에 설치했기 때문에, 로그 디렉토리도 이와 유사한 경로로 설정했다.
이때 해당 디렉토리는 루트 권한으로 생성해야 하며, Kafka가 해당 경로에 접근할 수 있도록 권한도 설정해주어야 한다.

$ sudo mkdir -p /opt/kafka/kraft-logs/kafka-logs1
$ sudo chown -R ubuntu:ubuntu /opt/kafka/kraft-logs

auto.create.topics.enable=True

테스트 단계에서는 auto.create.topics.enable=true 설정을 통해 토픽이 자동으로 생성되도록 설정해 두었다. 이는 Kafka의 기본 동작이기도 하다.
하지만 운영 환경에서는 자동 토픽 생성을 비활성화하는 것이 권장된다.
토픽의 수, 파티션 개수, 복제본(replica) 개수 등은 Kafka의 성능과 직접적으로 관련되며, 클러스터 전체에 생성된 파티션 수가 많아질수록 성능에 악영향을 줄 수 있다.
따라서 실제 운영 환경에서는 필요한 토픽을 미리 명시적으로 생성하고, 파티션 수와 복제본 개수도 사전에 적절히 설정하는 것이 안정적인 운영에 도움이 된다.

아키텍처로 Kafka 통신 이해하기

참고
PLAINTEXT: 외부 클라이언트 접속용 (9092)
INTERNAL: 브로커간 통신용 (19092 - 39092)
CONTROLLER: 브로커와 컨트롤러 간, 컨트롤러 간 통신 (19093 - 39093)

리더는 동적인 역할이다. 위 이미지에서는 리더가 중간에 고정된 것처럼 보이지만 이는 클라이언트가 브로커 중에 리더와 통신한다는 것을 시각적으로 나타내기 위함이고, 실제 Kafka에서는 파티션 단위로 리더 브로커가 다를 수 있다. 따라서 클라이언트는 각 파티션마다 리더 브로커가 다를 수 있으므로 여러 브로커에 연결할 수 있다.

# 클라이언트는 이 둘 모두에 연결 가능
topic-A partition 0 → 브로커1 리더

topic-A partition 1 → 브로커3 리더

클라이언트는 컨트롤러와 직접 통신하지 않는다. 브로커에게 메타데이터 요청만 할뿐이다. 클라리언트는 접속한 브로커에게 리더가 누군지 묻고 해당 브로커는 컨트롤러로부터 받은 메타데이터를 기반으로 응답을 한다.