목적

단일 Kafka 브로커에서 1개 브로커, 3개 파티션으로 구성된 클러스터를 3개의 브로커로 확장한 목적은 주로 확장성과 고가용성을 향상시키기 위함입니다.

대표적으로 장애 대응에 있어 하나의 브로커에 장애가 발생하더라도 다른 브로커가 해당 데이터를 처리할 수 있도록 특히 파티션 복제(replication)를 설정하여 브로커 중 하나가 다운되더라도 다른 브로커가 파티션을 유지할 수 있도록 하였습니다.

또한 끊임없이 들어오는 실시간 데이터를 병렬로 처리하는 Kafka 특징을 활용하여 더많은 데이터를 동시에 처리하도록 구성하고 싶었습니다. 그로 인해 부하가 분산되고 시스템의 안정성과 성능의 향상을 기대하였습니다.

Docker - Compose 작성

로컬 환경에서는 Zookeeper 1대, Broker 3대로 구성하여 진행하였습니다.

  coin-zookeeper:
    image: confluentinc/cp-zookeeper:latest
    container_name: coin-zookeeper
    ports:
      - "2181:2181"
    networks:
      - kafka_network
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_SERVER_ID: 1

  coin-kafka1:
    image: confluentinc/cp-kafka:latest
    container_name: coin-kafka1
    ports:
      - "9092:9092"
      - "29092:29092"
      - "7071:7071"  # JMX Exporter 포트
    healthcheck:
      test: [ "CMD", "bash", "-c", "printf \\"\\" > /dev/tcp/localhost/9092" ]
      interval: 5s
      timeout: 10s
      start_period: 40s
    networks:
      - kafka_network
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: "coin-zookeeper:2181"
#      KAFKA_LISTENERS: INTERNAL://0.0.0.0:19091,EXTERNAL://0.0.0.0:9091,DOCKER://0.0.0.0:29091
      KAFKA_ADVERTISED_LISTENERS: INTERNAL://coin-kafka1:19092,EXTERNAL://${DOCKER_HOST_IP:-127.0.0.1}:9092,DOCKER://host.docker.internal:29092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT,DOCKER:PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3
      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
      JMX_PORT: 9876  # JMX 포트
      KAFKA_OPTS: -javaagent:/usr/share/jmx_exporter/jmx_prometheus_javaagent.jar=7071:/usr/share/jmx_exporter/jmx_exporter_config.yml -Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.port=9876 -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false

    volumes:
      - ./jmx_exporter/jmx_prometheus_javaagent-1.0.1.jar:/usr/share/jmx_exporter/jmx_prometheus_javaagent.jar
      - ./jmx_exporter/jmx_exporter_config.yml:/usr/share/jmx_exporter/jmx_exporter_config.yml

    depends_on:
      - coin-zookeeper

  coin-kafka2:
    image: confluentinc/cp-kafka:latest
    container_name: coin-kafka2
    ports:
      - "9093:9093"
      - "29093:29093"
      - "7072:7072"  # JMX Exporter 포트
    healthcheck:
      test: [ "CMD", "bash", "-c", "printf \\"\\" > /dev/tcp/localhost/9093" ]
      interval: 5s
      timeout: 10s
      start_period: 40s
    networks:
      - kafka_network
    environment:
      KAFKA_BROKER_ID: 2
      KAFKA_ZOOKEEPER_CONNECT: "coin-zookeeper:2181"
      KAFKA_ADVERTISED_LISTENERS: INTERNAL://coin-kafka2:19093,EXTERNAL://${DOCKER_HOST_IP:-127.0.0.1}:9093,DOCKER://host.docker.internal:29093
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT,DOCKER:PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3
      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
      JMX_PORT: 9877  # JMX 포트
      KAFKA_OPTS: -javaagent:/usr/share/jmx_exporter/jmx_prometheus_javaagent.jar=7072:/usr/share/jmx_exporter/jmx_exporter_config.yml -Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.port=9877 -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false

    volumes:
      - ./jmx_exporter/jmx_prometheus_javaagent-1.0.1.jar:/usr/share/jmx_exporter/jmx_prometheus_javaagent.jar
      - ./jmx_exporter/jmx_exporter_config.yml:/usr/share/jmx_exporter/jmx_exporter_config.yml
    depends_on:
      - coin-zookeeper

  coin-kafka3:
    image: confluentinc/cp-kafka:latest
    container_name: coin-kafka3
    ports:
      - "9094:9094"
      - "29094:29094"
      - "7073:7073"  # JMX Exporter 포트
    healthcheck:
      test: [ "CMD", "bash", "-c", "printf \\"\\" > /dev/tcp/localhost/9094" ]
      interval: 5s
      timeout: 10s
      start_period: 40s
    networks:
      - kafka_network
    environment:
      KAFKA_BROKER_ID: 3
      KAFKA_ZOOKEEPER_CONNECT: "coin-zookeeper:2181"
      KAFKA_ADVERTISED_LISTENERS: INTERNAL://coin-kafka3:19094,EXTERNAL://${DOCKER_HOST_IP:-127.0.0.1}:9094,DOCKER://host.docker.internal:29094
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT,DOCKER:PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3
      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
      JMX_PORT: 9878  # JMX 포트
      KAFKA_OPTS: -javaagent:/usr/share/jmx_exporter/jmx_prometheus_javaagent.jar=7073:/usr/share/jmx_exporter/jmx_exporter_config.yml -Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.port=9878 -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false

    volumes:
      - ./jmx_exporter/jmx_prometheus_javaagent-1.0.1.jar:/usr/share/jmx_exporter/jmx_prometheus_javaagent.jar
      - ./jmx_exporter/jmx_exporter_config.yml:/usr/share/jmx_exporter/jmx_exporter_config.yml
    depends_on:
      - coin-zookeeper

spring application 설정

Producer Spring kafka Configuration

application.yml

KafkaCleanUpService
- 테스트를 진행하는 과정에서 계속 들어나는 Partition 의 Offset 을 초기화 하기 위해 서버가 종료되면 자동으로 Topic 이 삭제되어 Offset 이 초기화 되도록 구현
KafkaProducerConfig
KafkaTopicConfig
- Topic 의 이름과 파티션의 개수 그리고 replication 을 정해주었다. 여기서 3은 자기 자신을 포함한 숫자이다.

Consumer Spring kafka Configuration

KafkaConsumerConfig
- 해당코드만 변경되었고 나머지 코드는 동일하다.
- 파티션 개수와 Cosumer Group 안의 Consumer 의 개수 현재 브로커의 개수를 고려하여 1 : 1 로 처리하여 처리속도를 높혔다.

Kafka cluster EC2 환경 구성

ec2 서버는 주키퍼 앙상블을 활용하여 3대의 주키퍼와 3대의 브로커를 구성하였다
- 3대의 주키퍼 노드는 한 노드가 실패해도 클러스터가 정상적으로 동작할 수 있도록 리더 팔로우 구조로 구축하였다.
설정 과정에서는 큰 어려움 이 없었다.
주의할점은 주키퍼 앙상블, 주키퍼 노드, 브로커 노드의 포트번호를 잘 열어주어야 한다.