Kafka LISTENER 속성

카프카 속성중 KAFKA_LISTENERS,KAFKA_ADVERTISE_LISTENERS에 대해 알아 보자.

카프카에서는 “LINSTNERS”를 제공하고 있다.

보통 Spring boot와 같은 client(kafka client)를 이용하게 되며, 클라이언트안의 “bootstrapservers” 항목에 적혀있는 서버 정보를 보고 kafka 서버에 최초 접근한다.

이후 클라이언트는 *kafka 측으로 메타정보 요청을 하고 응답을 수신 받았을때 *“KAFKA_ADVERTISER_LISTENER” 에 적혀 있는 주소를 보고 어떤 브로커에 접속해야 할지 알게되며 이것을 기반으로 클라이언트는 접근 요청을 보낸다.

다이어그램으로 보면 다음과 같다.

 

1. 클라이언트는 bootstraps 정보를 보고 브로커에 접속하여 메타데이터를 요청한다.
2. 수신된 클라언트에게 메타데이터 정보를 응답한다.
3. 응답된 메타정보중 Advertise Listeners 정보로 브로커에 재접속 한다.
   1. 기존에 사용한 application.yaml의 bootstraps 정보는 사용하지 않는다.
4. 실제 데이터 송수신을 시작한다.

리스너 정보

구분	내용	설명
KAFKA_LISTENER	브로커가 어떤 IP/PORT로 리스닝 할지에 대한 설정	브로커가 실제 열고 있는 리스닝 정보
KAFKA_ADVERTISE_LISTENERS	클라이언트가 접속해야할 IP/PORT 정보	클라이언트에게 어디로 접속하라고 알려주는 경로

---

외부 접근 설정 (Listener)

# 리스너 설정
KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:29092,PLAINTEXT_HOST://0.0.0.0:9092,CONTROLLER://0.0.0.0:29093,**PLAINTEXT_LOCAL://0.0.0.0:39092**
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka-1:29092,PLAINTEXT_HOST://host.docker.internal:9092,**PLAINTEXT_LOCAL://localhost:39092**
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,**PLAINTEXT_LOCAL:PLAINTEXT**,PLAINTEXT_HOST:PLAINTEXT,CONTROLLER:PLAINTEXT

외부 접근을 하기 위해서는 리스너 설정을 해주면 된다. 여기서 예제는 어느곳에서든 접속 가능하게 하고
광고 리스너는 localhost로 해준다. 보안프로토콜 매핑은 개발임으로 PLANTEXT(비암호화)로 해준다.
추가된 리스너 명은 PLAINTEXT_LOCAL 이다

이처럼 필요한 프로토콜을 리스너에 추가 하여 사용 할 수있다.

그림으로 보면 다음과 같다.

 

 

(1)Kafka Client는 PLANTEXT_LOCAL(localhost) 명칭 프로토콜 39092,39093,39094 포트로 Docker container 안의 kafka와 통신한다.

(2)kubernates에 있는 SpringBoot Pod 들은 PLANTEXT_HOST(host.docker.internal) 9092,9093,9094의 포트로 통신한다.

(3) 카프카 간의 내부 통신은 ‘29092’ 포트를 사용하며 메타데이터를 주고 받으며 데이터를 복제한다.

(4) KRaft 모드를 사용하고 ‘29093’ 포트를 사용한다,

전체 Docker-compose.yaml 파일

####################################################
# compose-kafka-cluster.yaml
#
# ----------------------------------------
#
# docker-compose up -f ./compose-kafka-cluster.yaml
#
####################################################

services:
  kafka-1:
    user: root
    image: confluentinc/cp-kafka:7.4.0
    hostname: kafka-1
    container_name: kafka-1
    ports:
      - "9092:9092"
      - "9101:9101"
      - "39092:39092"
    environment:
      # KRaft 클러스터 설정 - 노드 1 (브로커 + 컨트롤러)
      CLUSTER_ID: "5tU1lP-kS9uH_fT8pA2v-1"
      KAFKA_PROCESS_ROLES: 'broker,controller'
      KAFKA_NODE_ID: 1
      KAFKA_CONTROLLER_QUORUM_VOTERS: '1@kafka-1:29093,2@kafka-2:29093,3@kafka-3:29093'

      # 리스너 설정
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:29092,PLAINTEXT_HOST://0.0.0.0:9092,CONTROLLER://0.0.0.0:29093,PLAINTEXT_LOCAL://0.0.0.0:39092
      #KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka-1:29092,PLAINTEXT_HOST://localhost:9092    #
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka-1:29092,PLAINTEXT_HOST://host.docker.internal:9092,PLAINTEXT_LOCAL://localhost:39092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_LOCAL:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT,CONTROLLER:PLAINTEXT

      # 브로커 간 통신 설정
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER

      # 로그 디렉토리
      KAFKA_LOG_DIRS: '/tmp/kraft-combined-logs'

      # 클러스터 설정 (복제 팩터 3)
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 2
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 3
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_DEFAULT_REPLICATION_FACTOR: 3
      KAFKA_MIN_INSYNC_REPLICAS: 2

      # JMX 모니터링
      KAFKA_JMX_PORT: 9101
      KAFKA_JMX_HOSTNAME: localhost
    volumes:
      - kafka-1-logs:/tmp/kraft-combined-logs
    command: >
      sh -c "
        echo '🔧 Fixing permissions on /tmp/kraft-combined-logs...' &&
        chown -R 1001:1001 /tmp/kraft-combined-logs &&
        exec /etc/confluent/docker/run
      "

  kafka-2:
    user: root
    image: confluentinc/cp-kafka:7.4.0
    hostname: kafka-2
    container_name: kafka-2
    ports:
      - "9093:9093"
      - "9102:9102"
      - "39093:39093"
    environment:
      # KRaft 클러스터 설정 - 노드 2 (브로커 + 컨트롤러)
      CLUSTER_ID: "5tU1lP-kS9uH_fT8pA2v-1"
      KAFKA_PROCESS_ROLES: 'broker,controller'
      KAFKA_NODE_ID: 2
      KAFKA_CONTROLLER_QUORUM_VOTERS: '1@kafka-1:29093,2@kafka-2:29093,3@kafka-3:29093'

      # 리스너 설정
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:29092,PLAINTEXT_HOST://0.0.0.0:9093,CONTROLLER://0.0.0.0:29093,PLAINTEXT_LOCAL://0.0.0.0:39093
      #KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka-2:29092,PLAINTEXT_HOST://localhost:9093
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka-2:29092,PLAINTEXT_HOST://host.docker.internal:9093,PLAINTEXT_LOCAL://localhost:39093
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT,CONTROLLER:PLAINTEXT,PLAINTEXT_LOCAL:PLAINTEXT

      # 브로커 간 통신 설정
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER

      # 로그 디렉토리
      KAFKA_LOG_DIRS: '/tmp/kraft-combined-logs'

      # 클러스터 설정 (복제 팩터 3)
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 2
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 3
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_DEFAULT_REPLICATION_FACTOR: 3
      KAFKA_MIN_INSYNC_REPLICAS: 2

      # JMX 모니터링
      KAFKA_JMX_PORT: 9102
      KAFKA_JMX_HOSTNAME: localhost
    volumes:
      - kafka-2-logs:/tmp/kraft-combined-logs

  kafka-3:
    user: root
    image: confluentinc/cp-kafka:7.4.0
    hostname: kafka-3
    container_name: kafka-3
    ports:
      - "9094:9094"
      - "9103:9103"
      - "39094:39094"
    environment:
      # KRaft 클러스터 설정 - 노드 3 (브로커 + 컨트롤러)
      CLUSTER_ID: "5tU1lP-kS9uH_fT8pA2v-1"
      KAFKA_PROCESS_ROLES: 'broker,controller'
      KAFKA_NODE_ID: 3
      KAFKA_CONTROLLER_QUORUM_VOTERS: '1@kafka-1:29093,2@kafka-2:29093,3@kafka-3:29093'

      # 리스너 설정
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:29092,PLAINTEXT_HOST://0.0.0.0:9094,CONTROLLER://0.0.0.0:29093,PLAINTEXT_LOCAL://0.0.0.0:39094
      #KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka-3:29092,PLAINTEXT_HOST://localhost:9094
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka-3:29092,PLAINTEXT_HOST://host.docker.internal:9094,PLAINTEXT_LOCAL://localhost:39094
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT,CONTROLLER:PLAINTEXT,PLAINTEXT_LOCAL:PLAINTEXT

      # 브로커 간 통신 설정
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER

      # 로그 디렉토리
      KAFKA_LOG_DIRS: '/tmp/kraft-combined-logs'

      # 클러스터 설정 (복제 팩터 3)
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 2
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 3
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_DEFAULT_REPLICATION_FACTOR: 3
      KAFKA_MIN_INSYNC_REPLICAS: 2

      # JMX 모니터링
      KAFKA_JMX_PORT: 9103
      KAFKA_JMX_HOSTNAME: localhost
    volumes:
      - kafka-3-logs:/tmp/kraft-combined-logs

  kafka-ui:
    image: provectuslabs/kafka-ui:latest
    container_name: kafka-ui-cluster
    depends_on:
      - kafka-1
      - kafka-2
      - kafka-3
    ports:
      - "8081:8080"
    environment:
      KAFKA_CLUSTERS_0_NAME: kafka-cluster
      # 클러스터 모든 브로커 주소
      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka-1:29092,kafka-2:29092,kafka-3:29092

volumes:
  kafka-1-logs:
  kafka-2-logs:
  kafka-3-logs: