raft

1. Raft Consensus Algorithm

클러스터가 여러개의 노드로 구성되어 있을때, 노드들이 서로 동일한 상태를 유지하기 위해 사용하는 알고리즘이다.

이론빼고 실전

zookeeper

vi docker-compose.yml

version: '3.8'
services:
  zk1:
    image: confluentinc/cp-zookeeper:7.4.1
    hostname: zk1
    container_name: zk1
    environment:
      ZOOKEEPER_SERVER_ID: 1
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
      ZOOKEEPER_INIT_LIMIT: 5
      ZOOKEEPER_SYNC_LIMIT: 2
      ZOOKEEPER_SERVERS: zk1:2888:3888;zk2:2888:3888;zk3:2888:3888
  zk2:
    image: confluentinc/cp-zookeeper:7.4.1
    hostname: zk2
    container_name: zk2
    environment:
      ZOOKEEPER_SERVER_ID: 2
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
      ZOOKEEPER_INIT_LIMIT: 5
      ZOOKEEPER_SYNC_LIMIT: 2
      ZOOKEEPER_SERVERS: zk1:2888:3888;zk2:2888:3888;zk3:2888:3888
  zk3:
    image: confluentinc/cp-zookeeper:7.4.1
    hostname: zk3
    container_name: zk3
    environment:
      ZOOKEEPER_SERVER_ID: 3
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
      ZOOKEEPER_INIT_LIMIT: 5
      ZOOKEEPER_SYNC_LIMIT: 2
      ZOOKEEPER_SERVERS: zk1:2888:3888;zk2:2888:3888;zk3:2888:3888

  kafka1:
    image: confluentinc/cp-kafka:7.4.1
    hostname: kafka1
    container_name: kafka1
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zk1:2181,zk2:2182,zk3:2183
      KAFKA_LISTENERS: INTERNAL://kafka1:9092
      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka1:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 2
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 3
  kafka2:
    image: confluentinc/cp-kafka:7.4.1
    hostname: kafka2
    container_name: kafka2
    environment:
      KAFKA_BROKER_ID: 2
      KAFKA_ZOOKEEPER_CONNECT: zk1:2181,zk2:2182,zk3:2183
      KAFKA_LISTENERS: INTERNAL://kafka2:9092
      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka2:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 2
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 3
  kafka3:
    image: confluentinc/cp-kafka:7.4.1
    hostname: kafka3
    container_name: kafka3
    environment:
      KAFKA_BROKER_ID: 3
      KAFKA_ZOOKEEPER_CONNECT: zk1:2181,zk2:2182,zk3:2183
      KAFKA_LISTENERS: INTERNAL://kafka3:9092
      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka3:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 2
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 3

  kafka-ui:
    image: provectuslabs/kafka-ui:latest
    hostname: kafka-ui
    container_name: kafka-ui
    ports:
      - 8080:8080
    environment:
      KAFKA_CLUSTERS_0_NAME: kafka
      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka1:9092,kafka2:9092,kafka3:9092
    depends_on:
      - kafka1
      - kafka2
      - kafka3

cd 'sample/raft/zookeeper-3-node'
docker-compose up -d

이제 zookeeper를 하나 죽여봅니다.

docker stop zk3

kafka-topics --bootstrap-server kafka1:9092 --create --topic test2

동작

하나 더 죽여봅니다.

docker stop zk2

kafka-topics --bootstrap-server kafka1:9092 --create --topic test3

에러납니다. 생성되지 않습니다.

3개에서는 2개 죽으면 클러스터 폭발..

kraft

3개 node kraft kafka cluster를 구성한다.

vi docker-compose.yml

version: '3'
services:
  controller1:
    image: confluentinc/cp-kafka:7.4.0
    hostname: controller
    container_name: controller
    # ports:
    #   - '9093:9093'
    environment:
      KAFKA_CONTROLLER_LISTENER_NAMES: 'CONTROLLER'
      KAFKA_LISTENERS: 'CONTROLLER://controller1:9093'
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_PROCESS_ROLES: 'controller'
      KAFKA_NODE_ID: 11
      KAFKA_CONTROLLER_QUORUM_VOTERS: '11@controller1:9093,12@controller2:9093,13@controller3:9093'
      # Replace CLUSTER_ID with a unique base64 UUID using "bin/kafka-storage.sh random-uuid"
      # See https://docs.confluent.io/kafka/operations-tools/kafka-tools.html#kafka-storage-sh
      CLUSTER_ID: 'ciWo7IWazngRchmPES6q5A=='
  controller2:
    image: confluentinc/cp-kafka:7.4.0
    hostname: controller2
    container_name: controller2
    environment:
      KAFKA_CONTROLLER_LISTENER_NAMES: 'CONTROLLER'
      KAFKA_LISTENERS: 'CONTROLLER://controller2:9093'
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_PROCESS_ROLES: 'controller'
      KAFKA_NODE_ID: 12
      KAFKA_CONTROLLER_QUORUM_VOTERS: '11@controller1:9093,12@controller2:9093,13@controller3:9093'
      # Replace CLUSTER_ID with a unique base64 UUID using "bin/kafka-storage.sh random-uuid"
      # See https://docs.confluent.io/kafka/operations-tools/kafka-tools.html#kafka-storage-sh
      CLUSTER_ID: 'ciWo7IWazngRchmPES6q5A=='
  controller3:
    image: confluentinc/cp-kafka:7.4.0
    hostname: controller3
    container_name: controller3
    environment:
      KAFKA_CONTROLLER_LISTENER_NAMES: 'CONTROLLER'
      KAFKA_LISTENERS: 'CONTROLLER://controller3:9093'
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_PROCESS_ROLES: 'controller'
      KAFKA_NODE_ID: 13
      KAFKA_CONTROLLER_QUORUM_VOTERS: '11@controller1:9093,12@controller2:9093,13@controller3:9093'
      # Replace CLUSTER_ID with a unique base64 UUID using "bin/kafka-storage.sh random-uuid"
      # See https://docs.confluent.io/kafka/operations-tools/kafka-tools.html#kafka-storage-sh
      CLUSTER_ID: 'ciWo7IWazngRchmPES6q5A=='

  kafka1:
    image: confluentinc/cp-kafka:7.4.0
    container_name: kafka1
    hostname: kafka1
    ports:
      - '9092:9092'
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_CONTROLLER_QUORUM_VOTERS: '11@controller1:9093,12@controller2:9093,13@controller3:9093'
      KAFKA_LISTENERS: 'INTERNAL://kafka1:9092'
      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka1:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: 'CONTROLLER:PLAINTEXT,INTERNAL:PLAINTEXT'
      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
      KAFKA_PROCESS_ROLES: 'broker'
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 2
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 3
      KAFKA_JMX_PORT: 9999
      KAFKA_JMX_HOSTNAME: kafka1
      KAFKA_LOG_DIRS: '/tmp/kraft-combined-logs'
      # Replace CLUSTER_ID with a unique base64 UUID using "bin/kafka-storage.sh random-uuid"
      # See https://docs.confluent.io/kafka/operations-tools/kafka-tools.html#kafka-storage-sh
      CLUSTER_ID: 'ciWo7IWazngRchmPES6q5A=='
    depends_on:
      - controller1
      - controller2
      - controller3

  kafka2:
    image: confluentinc/cp-kafka:7.4.0
    container_name: kafka2
    hostname: kafka2
    environment:
      KAFKA_NODE_ID: 2
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_CONTROLLER_QUORUM_VOTERS: '11@controller1:9093,12@controller2:9093,13@controller3:9093'
      KAFKA_LISTENERS: 'INTERNAL://kafka2:9092'
      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka2:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: 'CONTROLLER:PLAINTEXT,INTERNAL:PLAINTEXT'
      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
      KAFKA_PROCESS_ROLES: 'broker'
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 2
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 3
      KAFKA_JMX_PORT: 9999
      KAFKA_JMX_HOSTNAME: kafka2
      KAFKA_LOG_DIRS: '/tmp/kraft-combined-logs'
      # Replace CLUSTER_ID with a unique base64 UUID using "bin/kafka-storage.sh random-uuid"
      # See https://docs.confluent.io/kafka/operations-tools/kafka-tools.html#kafka-storage-sh
      CLUSTER_ID: 'ciWo7IWazngRchmPES6q5A=='
    depends_on:
      - controller1
      - controller2
      - controller3

  kafka3:
    image: confluentinc/cp-kafka:7.4.0
    container_name: kafka3
    hostname: kafka3
    environment:
      KAFKA_NODE_ID: 3
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_CONTROLLER_QUORUM_VOTERS: '11@controller1:9093,12@controller2:9093,13@controller3:9093'
      KAFKA_LISTENERS: 'INTERNAL://kafka3:9092'
      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka3:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: 'CONTROLLER:PLAINTEXT,INTERNAL:PLAINTEXT'
      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
      KAFKA_PROCESS_ROLES: 'broker'
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 2
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 3
      KAFKA_JMX_PORT: 9999
      KAFKA_JMX_HOSTNAME: kafka3
      KAFKA_LOG_DIRS: '/tmp/kraft-combined-logs'
      # Replace CLUSTER_ID with a unique base64 UUID using "bin/kafka-storage.sh random-uuid"
      # See https://docs.confluent.io/kafka/operations-tools/kafka-tools.html#kafka-storage-sh
      CLUSTER_ID: 'ciWo7IWazngRchmPES6q5A=='
    depends_on:
      - controller1
      - controller2
      - controller3

  kafka-ui:
    image: provectuslabs/kafka-ui:latest
    restart: always
    hostname: kafka-ui
    container_name: kafka-ui
    ports:
      - 8080:8080
    environment:
      KAFKA_CLUSTERS_0_NAME: kafka
      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka1:9092,kafka2:9092,kafka3:9092
      # KAFKA_CLUSTERS_0_KAFKACONNECT_0_ADDRESS: http://kafka1:8083
      # KAFKA_CLUSTERS_0_KAFKACONNECT_0_NAME: kafka1-connect
    depends_on:
      - kafka1
      - kafka2
      - kafka3
      - controller1
      - controller2
      - controller3

cd kafka/sample/raft/3-node
docker-compose up -d
docker exec -it kafka1 bash

# create topic
kafka-topics --bootstrap-server kafka1:9092,kafka2:9092,kafka3:9092 --create --topic test

동작합니다.

이제 controler를 하나 죽여봅니다.

docker stop controller3

kafka-topics --bootstrap-server kafka1:9092,kafka2:9092,kafka3:9092 --create --topic test2

여전히 동작합니다.

하나 더 죽여봅니다.

docker stop controller2
kafka-topics --bootstrap-server kafka1:9092,kafka2:9092,kafka3:9092 --create --topic test2

Error while executing topic command : Call(callName=createTopics, deadlineMs=1705497739741, tries=1, nextAllowedTryMs=1705497739889) timed out at 1705497739789 after 1 attempt(s)
[2024-01-17 13:22:19,792] ERROR org.apache.kafka.common.errors.TimeoutException: Call(callName=createTopics, deadlineMs=1705497739741, tries=1, nextAllowedTryMs=1705497739889) timed out at 1705497739789 after 1 attempt(s)
Caused by: org.apache.kafka.common.errors.DisconnectException: Cancelled createTopics request with correlation id 3 due to node 2 being disconnected
 (kafka.admin.TopicCommand$)

생성되지 않습니다. 이유는 controller가 정족수(2개)를 넘지 못해서 입니다.

결론

kraft도 여전히 Raft Consensus Algorithm 을 사용하기때문에 정족수를 넘지 못하면 동작하지 않습니다.

참고

https://thesecretlivesofdata.com/raft/ 이 사이트 신기하네요

내결함성을 위한 최적의 노드 수

합의 알고리즘(Consensus Algorithm)을 채택한 분산 시스템에서는 전체 노드 수를 가급적 3개 이상의 홀수로 유지하는 것이 권장된다. 이유는 아래와 같다.

전체 노드 수	필요 정족수 허용	가능 장애 노드 수
1	1	0
2	2	0
3	2	1
4	3	1
5	3	2
6	4	2
7	4	3
8	5	3

... ... ... 2k (짝수) k+1 k-1 2k+1 (홀수) k+1 k

전체 노드 수가 3개 이상이어야 허용 가능한 장애 노드가 생긴다. 그 미만으로는 내결함성을 갖출 수 없다.

2개의 노드로 클러스터를 구성하는 것은 자원 관리의 측면에서나 내결함성의 측면에서나 모두 비효율적인 방식임을 알 수 있다.

zookeeper나 controller 클러스터를 이야기하는것입니다. kafka 클러스터는 다른 이야기입니다.