Apache Kafka without Zookeeper

Tài liệu này sẽ nói về Kafka ở chế độ KRAFT. KRAFT có ý nghĩa giờ đây Kafka sẽ tổ chức lại hoạt động của cluster dựa theo concept của thuật toán RAFT, lý do cho việc thay đổi này là do những hạn chế của Zookeeper trong các phiên bản cũ.

Về RAFT, RAFT là thuật toán đồng thuận được sử dụng phổ biến cho các hệ thống phân tán, một số hệ thống phổ biến :

• RabbitMQsử dụng Raft để thay thế loại hang đợi mirror (replicated) sang quorum queue.
• MongoDBsử dụng một biến thể Raft trong replication set.
• SplunkEnterprise sử dụng Raft trong cơ chế Search Head Cluster (SHC)…v.v.

Với RAFT, đây là thuật toán rất thú vị trong tư tưởng về lập trình hệ thống phân tán. Tài liệu này không mô tả về RAFT tuy nhiên để tiếp cận nhanh nhất thuật toán này trong vài phút thì dưới đây là đường dẫn visualization mô phỏng thuật toán :
http://thesecretlivesofdata.com/raft/

1.Vấn đề của Zookeeper

Zookeeper dường như là một phần không thể tách rời khi vận hành một hệ thống phân tán như Kafka nhưng nó có một số hạn chế khiến Kafka không thể phát huy hết khả năng của mình. Quy mô Kafka gặp phải bottleneck hiệu suất với Zookeeper

Hãy nhìn vào kiến trúc sau (hình ảnh từ confluence kafka):

• Kafka metadata : mô tả các borkers hiện có,thông tin về máy chủ và port . Đồng thời cung cấp thông tin về vị trí lưu trữ của borker host nào đang lưu trữ partition nào.
• Kafka controller: node đóng vai trò làm bộ điều khiển còn gọi là bộ điều khiển kafka

• Đối với mô hình Zookeeper trên, mỗi cụm có một node duy nhất đóng vai trò là bộ điều khiển (controller) được bầu chọn bởi những ZooKeeper watchers.
• Nó lưu trữ partition logs, xử lý consume/produce giống như các broker khác và duy trì siêu dữ liệu cụm như ID broker, racks, topic, partition, leader, thông tin ISR(in-sync replicas) và cấu hình toàn cụm theo mỗi topic, cũng như thông tin đăng nhập bảo mật.
• Kết quả là phần lớn lưu lượng đọc và ghi của ZooKeeper được thực hiện bởi node controller.
• Controller đọc trạng thái của zookeeper và cập nhật metadata cho từng brokers một cách lần lượt và tuần tự. Zookeeper là nguồn metadata cho kafka, giữ thông tin trạng thái cụm kafka.
• Giả sử một node trong cụm bị tắt dẫn đến metadata thay đổi, controller sẽ phải ghi lại metadata mới vào Zookeeper đồng thời controller sẽ update lần lượt metadata cho lần lượt từng borker, trong quá trình đó vấn đề ở đây khi client cố gắng tìm kiếm leader bằng cách query thông tin từ broker client có khả năng sẽ không có câu trả lời, vì broker mà client đang tham khảo chưa được cập nhật lại thông tin trạng thái cụm.
• Việc thiết lập cụm kafka sẽ phụ thuộc vào zookeeper, vấn đề lớn nhất ở đây là 2 hệ thống hoàn toàn khác biệt. Zookeeper là hệ thống bên ngoài kafka có nghĩa nó đi kèm với cú pháp tập lệnh điều khiển, cấu hình riêng của nó. Do đó khi bạn triển khai cụm cluster kafka bạn cũng phải quản lý, triển khai và giám sát Zookeeper.
• Kafka & zookeeper triển khai 2 hệ thống sẽ không hiệu quả về mặt tài nguyên.
• Với Zookeeper khả năng mở rộng của Kafka bị hạn chế. Mỗi khi khởi động cluster ,controller node Kafka phải tải trạng thái cluster từ Zookeeper, điều tương tự cũng xảy khi một Kafka broker tham gia vào cluster hoặc rời khỏi cluster , Zookeeper sẽ thực hiện lại việc bầu chọn leader, khi tiến trình này diễn ra Zookeeper sẽ làm chậm cụm (cluster), lượng metadata lớn dần theo thời gian dẫn đến việc tải lại metadata trở nên dần kém hiệu quả và làm hạn chế số lượng phân vùng mà cụm có thể lưu trữ theo confluent số lượng này tối đa là 200.000.
• Để vượt qua các thách thức của Zookeeper ở Kafka , một KIP (Kafka improvement plan) đã được xuất bản đó là KIP-500 (https://cwiki.apache.org/confluence/display/KAFKA/KIP-500%3A+Replace+ZooKeeper+with+a+Self-Managed+Metadata+Quorum). Trong phiên bản phát hành mới nhất , Zookeeper có thể được thay thế bằng một nhóm bộ điều khiển RAFT nội bộ. Bắt đầu từ phiên bản thử nghiệm 2.8.0 . Tuy nhiên cho đến khi phiên bản 4.0.0 phát hành trong tương lai thì các phiên bản hiện tại mới chỉ là bridge release.

2.Các thay đổi về kiến trúc cluster Kafka

Hình ảnh dưới đây mô tả sự khác biệt giữa kiến trúc cũ và kiến trúc mới được khuyến nghị

• Trong kiến trúc được đề xuất, ba nút điều khiển thay thế cho ba nút ZooKeeper.
• Trong kiến trúc cũ zoopkeeper , node controller sẽ phải chủ động update trạng thái cho các node broker trong cụm theo hướng mũi tên.
• Giờ đây thay vì controller đưa ra các bản cập nhật cho các borker node, các borker node sẽ tự lấy metadata từ leader này . Trong các node controller , các node controller đang là follower (màu xanh) sẽ follow metadata từ controller leader (màu cam), tất cả controller sẽ cùng theo dõi trạng thái mới nhất metadata , điều này làm cho quá trình chuyển đổi dự phòng hầu như không bị tải lại metadata khi chuyển sang trạng thái mới. Cũng giống như Zookeeper, Raft yêu cầu phần lớn các node chạy để tiếp tục chạy, do đó cần tối thiểu 3 node cho một mô hình cụm cluster raft.
Mặc dù các quy trình của controller tách biệt về mặt logic với các quy trình của borker, nhưng chúng không cần phải tách biệt về mặt vật lý

2.2 Mô hình triển khai

Một cụm tối thiểu sẽ triển khai như sau :

Các broker A và broker B sẽ lắng nghe event từ metadata topic , broker C leader sẽ là nguồn ghi event vào metadata topic. Khi leader C bị lỗi việc tải lại metadata sẽ không còn mất thời gian vì cả broker A và B đều đã có metadata trong bộ nhớ của nó. Ngoài ra Broker C có 1 một chu kì snapshot để hạn chế kích thước của metadata topic.

Hình bên dưới là kiến trúc nếu ta triển khai tách biệt vai trò giữa các nút trong cụm.

file setup docker-compose :

version: '3'
networks:
  default:
    name: kafka_raft
services:
 kafka-1:
    image: bitnami/kafka:3.2.3
    container_name: kafka1
    hostname: node1
    user: root
    ports:
      - 19092:19092 #controller port
      - 9094:9094  
    environment:
      - KAFKA_BROKER_ID=1
      - KAFKA_CFG_NODE_ID=1
      - KAFKA_ENABLE_KRAFT=yes 
      - KAFKA_CFG_PROCESS_ROLES=controller,broker
      - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT,CLIENT:PLAINTEXT
      - KAFKA_CFG_LISTENERS=CLIENT://0.0.0.0:9092,CONTROLLER://0.0.0.0:19092,EXTERNAL://:9094
      - KAFKA_CFG_ADVERTISED_LISTENERS=CLIENT://127.0.0.1:9092,EXTERNAL://127.0.0.1:9094
      - KAFKA_KRAFT_CLUSTER_ID=kkdhsaJUsdnnxjNNjjddtr  
      - ALLOW_PLAINTEXT_LISTENER=yes
      - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=1@node1:19092,2@node2:19093,3@node3:19094
      - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER
      - KAFKA_CFG_BROKER_LISTENER_NAMES=PLAINTEXT
      - KAFKA_HEAP_OPTS=-Xmx256m -Xms256m
      - BITNAMI_DEBUG=true
      - KAFKA_CFG_INTER_BROKER_LISTENER_NAME=CLIENT
 kafka-2:
    image: bitnami/kafka:3.2.3
    container_name: kafka2
    hostname: node2
    user: root
    ports:
      - 19093:19093 #external port controller
      - 9095:9095    
    environment:
      - KAFKA_BROKER_ID=2
      - KAFKA_CFG_NODE_ID=2
      - KAFKA_ENABLE_KRAFT=yes 
      - KAFKA_CFG_PROCESS_ROLES=controller,broker
      - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT,CLIENT:PLAINTEXT
      - KAFKA_CFG_LISTENERS=CLIENT://0.0.0.0:9092,CONTROLLER://0.0.0.0:19093,EXTERNAL://0.0.0.0:9095
      - KAFKA_CFG_ADVERTISED_LISTENERS=CLIENT://127.0.0.1:9092,EXTERNAL://127.0.0.1:9095
      - KAFKA_KRAFT_CLUSTER_ID=kkdhsaJUsdnnxjNNjjddtr  
      - ALLOW_PLAINTEXT_LISTENER=yes
      - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=1@node1:19092,2@node2:19093,3@node3:19094
      - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER
      - KAFKA_CFG_BROKER_LISTENER_NAMES=PLAINTEXT
      - KAFKA_HEAP_OPTS=-Xmx256m -Xms256m
      - BITNAMI_DEBUG=true
      - KAFKA_CFG_INTER_BROKER_LISTENER_NAME=CLIENT
 kafka-3:
    image: bitnami/kafka:3.2.3
    container_name: kafka3
    hostname: node3
    user: root
    ports:
      - 19094:19094 #external port controller
      - 9096:9096  
    environment:
      - KAFKA_BROKER_ID=3
      - KAFKA_CFG_NODE_ID=3
      - KAFKA_ENABLE_KRAFT=yes 
      - KAFKA_CFG_PROCESS_ROLES=controller,broker
      - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT,CLIENT:PLAINTEXT
      - KAFKA_CFG_LISTENERS=CLIENT://0.0.0.0:9092,CONTROLLER://0.0.0.0:19094,EXTERNAL://0.0.0.0:9096
      - KAFKA_CFG_ADVERTISED_LISTENERS=CLIENT://127.0.0.1:9092,EXTERNAL://127.0.0.1:9096
      - KAFKA_KRAFT_CLUSTER_ID=kkdhsaJUsdnnxjNNjjddtr  
      - ALLOW_PLAINTEXT_LISTENER=yes
      - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=1@node1:19092,2@node2:19093,3@node3:19094
      - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER
      - KAFKA_CFG_BROKER_LISTENER_NAMES=PLAINTEXT
      - KAFKA_HEAP_OPTS=-Xmx256m -Xms256m
      - BITNAMI_DEBUG=true
      - KAFKA_CFG_INTER_BROKER_LISTENER_NAME=CLIENT

Cài đặt

Run docker file với lệnh : docker-compose -f multi-node.yml up
Bật cửa sổ cmd truy cập vào node để tạo topic thep các bước sau :

Step 1. Chạy lệnh :
docker exec -it kafka2 /bin/bash
Step 2.Truy cập vào folder chứa tập lệnh điều khiển của kafka
cd /opt/bitnami/kafka/bin
Step 3.Tạo topic trên 3 node của cụm
./kafka-topics.sh --bootstrap-server localhost:9092 --create --topic worldcapitals --partitions 3 --replication-factor 3
Kiểm tra thông tin topic vừa tạo:
a. ./kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic worldcapitals
Lúc này 3 phân vùng sẽ được tạo ra, mỗi phân vùng sẽ có bản sao ở các phân vùng còn lại