🏘️ 카프카 클러스터 구축

1. vi docker-compose.yml

로컬에 다운받은 카프카 폴더에서 yml 파일 편집

version: '2.1'

services:
  zoo1:
    image: confluentinc/cp-zookeeper:latest
    hostname: zoo1
    container_name: zoo1
    ports:
      - "2182:2181"
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_SERVER_ID: 1
      ZOOKEEPER_SERVERS: zoo1:2888:3888

  kafka1:
    image: confluentinc/cp-kafka:latest
    hostname: kafka1
    container_name: kafka1
    ports:
      - "9093:9093"
      - "29093:29093"
    environment:
      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka1:19093,EXTERNAL://localhost:9093,DOCKER://host.docker.internal:29093
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT,DOCKER:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
      KAFKA_ZOOKEEPER_CONNECT: "zoo1:2181"
      KAFKA_BROKER_ID: 1
      KAFKA_LOG4J_LOGGERS: "kafka.controller=INFO,kafka.producer.async.DefaultEventHandler=INFO,state.change.logger=INFO"
      KAFKA_AUTHORIZER_CLASS_NAME: kafka.security.authorizer.AclAuthorizer
      KAFKA_ALLOW_EVERYONE_IF_NO_ACL_FOUND: "true"
    depends_on:
      - zoo1

  kafka2:
    image: confluentinc/cp-kafka:latest
    hostname: kafka2
    container_name: kafka2
    ports:
      - "9094:9094"
      - "29094:29094"
    environment:
      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka2:19094,EXTERNAL://localhost:9094,DOCKER://host.docker.internal:29094
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT,DOCKER:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
      KAFKA_ZOOKEEPER_CONNECT: "zoo1:2181"
      KAFKA_BROKER_ID: 2
      KAFKA_LOG4J_LOGGERS: "kafka.controller=INFO,kafka.producer.async.DefaultEventHandler=INFO,state.change.logger=INFO"
      KAFKA_AUTHORIZER_CLASS_NAME: kafka.security.authorizer.AclAuthorizer
      KAFKA_ALLOW_EVERYONE_IF_NO_ACL_FOUND: "true"
    depends_on:
      - zoo1

  kafka3:
    image: confluentinc/cp-kafka:latest
    hostname: kafka3
    container_name: kafka3
    ports:
      - "9095:9095"
      - "29095:29095"
    environment:
      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka3:19095,EXTERNAL://localhost:9095,DOCKER://host.docker.internal:29095
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT,DOCKER:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
      KAFKA_ZOOKEEPER_CONNECT: "zoo1:2181"
      KAFKA_BROKER_ID: 3
      KAFKA_LOG4J_LOGGERS: "kafka.controller=INFO,kafka.producer.async.DefaultEventHandler=INFO,state.change.logger=INFO"
      KAFKA_AUTHORIZER_CLASS_NAME: kafka.security.authorizer.AclAuthorizer
      KAFKA_ALLOW_EVERYONE_IF_NO_ACL_FOUND: "true"
    depends_on:
      - zoo1

로컬 2182 포트로 주키퍼 접속 (내부적으론 2181)
카프카 1 : 9093 포트
카프카 2 : 9094 포트
카프카 3 : 9095 포트

    ports:
      - "9095:9095"
      - "29095:29095"

9095:9095: 이 부분은 호스트의 9095 포트를 컨테이너의 9095 포트로 매핑
-> 호스트에서 Kafka 브로커에 접속할 때 9095 포트를 사용한다는 의미
29095:29095: Docker 내부 네트워크에서의 포트 매핑을 정의
-> 컨테이너 내부에서 Kafka 브로커에 접속할 때 사용할 포트 지정.
컨테이너 내부에서 29095 포트를 사용하여 Kafka 브로커에 접속할 수 있음.

1. 실행

$docker-compose -f docker-compose.yml up -d # yml에 포함된 컨테이너들 올리는 명령어
$docker ps # docker ps 명령어로 잘 돌아가나 확인

2. 토픽생성

# 로컬 카프카 폴더내에서
./bin/kafka-topics.sh --create \
    --replication-factor 3 \
    --partitions 3 \
    --topic app-push-topic \
    --bootstrap-server localhost:9093,localhost:9094,localhost:9095

• "app-push-topic"이라는 토픽을 생성하며, 각 파티션에는 3개의 복제본이 있고, 파티션 수는 3개
• 실행하면 Kafka 클러스터 내에서 해당 토픽이 생성되고, 관련된 브로커 간에 파티션 복제가 설정됨

# 토픽 상세 조회
./bin/kafka-topics.sh --describe \
    --topic app-push-topic \
    --bootstrap-server localhost:9093,localhost:9094,localhost:9095

#결과
Topic: app-push-topic	TopicId: C__MUpdpTv-Ej1s8Empw8A	PartitionCount: 3	ReplicationFactor: 3	Configs:
	Topic: app-push-topic	Partition: 0	Leader: 1	Replicas: 1,2,3	Isr: 2,3,1
	Topic: app-push-topic	Partition: 1	Leader: 2	Replicas: 2,3,1	Isr: 2,3,1
	Topic: app-push-topic	Partition: 2	Leader: 3	Replicas: 3,1,2	Isr: 3,2,1

• Partition: 0, 1, 2: 각 파티션의 번호
• Leader: 1, 2, 3: 각 파티션의 리더 브로커의 ID
• Replicas: 1,2,3, 2,3,1, 3,1,2: 각 파티션에 대한 복제본(Replica)의 브로커 ID 목록. 순서는 중요하지 않음!
• Isr: 2,3,1, 2,3,1, 3,2,1: 현재 인-시크(In-Sync Replica) 목록. 각 파티션의 리더 브로커를 포함한 현재 복제본의 목록

📌 인-시크란?
Kafka에서 각 파티션은 여러 개의 복제본을 가질 수 있음, 이 중에서 하나는 리더이며, 나머지는 팔로워 복제본. 리더 파티션은 메시지를 쓰고 소비하는 역할을 수행하며, 팔로워 복제본들은 리더와 동기화되어 있어 리더에 장애가 발생했을 때 대체 가능.
→ 인-시크는 현재 메시지를 따라가는 동기화된 상태에 있는 복제본들을 나타냄. 즉, 인-시크에 속한 복제본들은 리더와 동일한 메시지를 가지고 있으며, 데이터의 일관성을 보장하는것
→ 만약 리더 파티션에서 메시지를 송신하고 있는데 팔로워 중 하나가 동기화되지 못한 상태라면 해당 복제본은 인-시크에서 제외됨.

쉘 4개 이제 열어서 , 1개 프로듀서 3개 컨슈머 실행

# consumer jar 파일 있는 디렉토리에서
java -jar consumer-2.3.0.RELEASE.jar

# producer jar 파일 있는 디렉토리에서
java -jar producer-2.3.0.RELEASE.jar

postman으로 TEST

분산처리 확인완료

---

# 컨슈머 그룹 리스트 조회
./bin/kafka-consumer-groups.sh --list \
    --bootstrap-server localhost:9093,localhost:9094,localhost:9095

app-push-group

# 컨슈머 그룹 상세조회
./bin/kafka-consumer-groups.sh --describe \
    --group app-push-group \
    --bootstrap-server localhost:9093,localhost:9094,localhost:9095

GROUP           TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID                                                    HOST            CLIENT-ID
app-push-group  app-push-topic  0          264             264             0               consumer-app-push-group-1-a9d6852b-1f81-400b-8337-6466a1f0aab2 /192.168.65.1   consumer-app-push-group-1
app-push-group  app-push-topic  2          248             248             0               consumer-app-push-group-1-dc22572f-eddf-4ed2-8299-d6da8db4edfa /192.168.65.1   consumer-app-push-group-1
app-push-group  app-push-topic  1          270             270             0               consumer-app-push-group-1-c17f0f35-d5a2-48d4-8952-9c9f9cc820fd /192.168.65.1   consumer-app-push-group-1

• GROUP | Topic | PARTITION : 조회한 컨슈머 그룹이 마지막으로 커밋한 토픽과 파티션을 나타냄
• CURRENT-OFFSET : 컨슈머 그룹이 가져간 토픽의 파티션에 가장 최신 오프셋(offset)이 몇번 인지 나타냄
  • 오프셋 : 파티션의 각 레코드에 할당된 번호 , 데이터가 파티션에 들어올때마다 1씩 증가
• LOG-END-OFFSET : 해당 컨슈머 그룹의 컨슈머가 어느 오프셋까지 커밋했는지 알 수 있다
  • CURRENT-OFFSET ≤ LOG-END-OFFSET
• LAG : 컨슈머 그룹이 토픽의 파티션에 있는 데이터를 가져가는데 얼마나 지연이 발생하나 나타내는 지표
  • LOG-END-OFFSET(컨슈머 그룹이 커밋한 오프셋) - CURRENT-OFFSET(해당 파티션의 가장 최신 오프셋) = LAG
• CONSUMER-ID : 컨슈머의 할당을 내부적으로 구분하기 위한 id
• HOST : 컨슈머가 동작하는 host 명 , 이 값을 통해 컨슈머의 호스트명 또는 ip를 알아 낼 수 있다.
• CLIENT-ID : 컨슈머에 할당된 id , 사용자가 지정가능 , 지정하지 않으면 자동생성됨