Kafka replication: broker, leader, follower, ISR

Ở đây cần tách hai tầng: nhiều instance Kafka server (broker) khác với nhiều instance consumer service. Bài này nói về broker. Consumer để bài 8.8.

1. Bài toán: một broker cho cả shop-app

Topic orders có 6 partition, đang chạy trên một Kafka server:

kafka-1:  P0 P1 P2 P3 P4 P5
             ▲
order-service ghi, inventory/notification/analytics đọc

2. Vấn đề

• kafka-1 chết (đĩa hỏng, VM restart): order-service không publish được, đặt hàng lỗi.
• Đĩa hỏng hẳn: mất toàn bộ event chưa được consume, analytics không replay được.
• Một máy gánh hết network/disk của 6 partition: khó scale throughput.

3. Giải pháp: cluster nhiều broker + replication

4. Code từng bước

Bước 1: Cluster 3 broker (docker-compose rút gọn, KRaft mode, không cần ZooKeeper):

services:
  kafka-1:
    image: apache/kafka:3.9.0
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka-1:9093,2@kafka-2:9093,3@kafka-3:9093
      KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka-1:9092
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_DEFAULT_REPLICATION_FACTOR: 3
      KAFKA_MIN_INSYNC_REPLICAS: 2
      KAFKA_UNCLEAN_LEADER_ELECTION_ENABLE: "false"
  kafka-2: { ... KAFKA_NODE_ID: 2 ... }
  kafka-3: { ... KAFKA_NODE_ID: 3 ... }

Bước 2: Topic với replication 3, min ISR 2:

@Bean
public NewTopic ordersTopic() {
    return TopicBuilder.name("orders")
            .partitions(6)
            .replicas(3)
            .config(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, "2")
            .build();
}

Bước 3: Producer bền (order-service application.yml):

spring:
  kafka:
    bootstrap-servers: kafka-1:9092,kafka-2:9092,kafka-3:9092
    producer:
      acks: all
      properties:
        enable.idempotence: true
        delivery.timeout.ms: 120000

bootstrap-servers chỉ để lấy metadata ban đầu; liệt kê vài broker để một con chết vẫn kết nối được. Sau đó client tự biết leader của từng partition nằm ở broker nào.

Bước 4: Test failover

kafka-topics.sh --bootstrap-server kafka-1:9092 --describe --topic orders
# Partition: 4  Leader: 2  Replicas: 2,3,1  Isr: 2,3,1

docker stop kafka-2
kafka-topics.sh --bootstrap-server kafka-1:9092 --describe --topic orders
# Partition: 4  Leader: 3  Replicas: 2,3,1  Isr: 3,1      ← leader mới

curl -X POST localhost:8081/orders -d '{"productId":"P1","quantity":1}' -H "Content-Type: application/json"
# vẫn 201: ISR còn 2 ≥ min.insync.replicas=2

docker stop kafka-3
# ISR chỉ còn 1 < 2 → producer nhận NotEnoughReplicasException, đặt hàng báo lỗi/retry

5. Cách hoạt động

             Broker 1        Broker 2        Broker 3
P0           Leader ★        follower        follower
P1           follower        Leader ★        follower
P4           follower        Leader ★        follower
...

Producer (acks=all) gửi record key=101 → P4
   │
   ▼
Broker 2 (leader P4) ghi vào log
   │  follower ở Broker 3, Broker 1 fetch
   ▼
Khi mọi replica trong ISR có record → high watermark tăng → trả ACK cho producer
Consumer chỉ đọc được tới high watermark (record đã "committed")

Broker 2 chết:
Controller chọn leader mới cho P4 trong ISR {3,1} → Broker 3
Client refresh metadata → ghi/đọc P4 ở Broker 3

Hai kiểu "chia" rất dễ nhầm:

Broker chia nhau LƯU partition:      P0→B1, P1→B2, P2→B3 ...
Consumer chia nhau ĐỌC partition:    P0,P1→C1, P2,P3→C2 ...   (bài 8.8)
Consumer KHÔNG gắn cố định với broker.

6. So sánh cấu hình durability

Từ Kafka 3.0, producer mặc định acks=all và enable.idempotence=true. Nhưng min.insync.replicas mặc định của broker vẫn là 1, phải tự đặt.

7. Khi nào nên dùng / không nên dùng

• Nên: production luôn RF=3, min.insync.replicas=2, acks=all, unclean election tắt cho event quan trọng (đơn hàng, thanh toán).
• Có thể nới: log/metrics chấp nhận mất ít dữ liệu để đổi latency: acks=1.
• Không nên: RF=1 trên production; một broker hỏng là mất partition.

Đây là đúng chỗ đánh đổi giữa availability và consistency.

• ISR còn 1 < min.insync.replicas = 2: producer acks=all bị từ chối (NotEnoughReplicas). Hệ thống "đứng" phần ghi nhưng không mất dữ liệu.
• Nếu leader cuối cùng trong ISR cũng chết, chỉ còn follower out-of-sync:
  • unclean.leader.election.enable=false (mặc định): partition offline, chờ replica ISR quay lại.
  • =true: chọn follower chậm làm leader. Partition sống lại ngay nhưng mất các record follower đó chưa có.

Leader (B2): 0 1 2 3 4 5     ← chết
Follower chậm (B1): 0 1 2    ← nếu unclean election → leader mới
Record 3, 4, 5 biến mất; producer đã nhận ACK cho chúng nếu acks=1

Với topic orders thì giữ false. Mất đơn hàng tệ hơn vài phút không ghi được.

Không cần nữa. Bạn nhớ ngắn gọn:

Ý tưởng không đổi: vẫn có một controller theo dõi broker sống/chết và bầu leader cho partition. Chỉ là nơi lưu metadata đã chuyển vào chính Kafka.

Q1. Kafka đảm bảo không mất dữ liệu khi broker chết như thế nào?

🇻🇳 Tiếng Việt:

Mỗi partition được replicate sang nhiều broker theo replication factor, thường là 3. Một replica là leader nhận ghi, các follower copy từ leader. Producer dùng acks=all để chỉ nhận ACK khi mọi replica trong ISR đã có record, kết hợp min.insync.replicas=2. Leader chết thì controller chọn leader mới trong ISR nên dữ liệu đã ACK không mất.

🇬🇧 English:

Each partition is replicated to several brokers based on the replication factor, usually 3. One replica is the leader that accepts writes and followers copy from it. The producer uses acks=all so it is acknowledged only when all in-sync replicas have the record, together with min.insync.replicas=2. If the leader dies, the controller elects a new leader from the ISR, so acknowledged data is not lost.

Q2. ISR là gì? acks=all có đủ chưa?

🇻🇳 Tiếng Việt:

ISR là các replica đang theo kịp leader. acks=all một mình chưa đủ, vì nếu ISR co lại còn mỗi leader thì acks=all cũng chỉ là một bản. Phải đặt min.insync.replicas, ví dụ 2 với RF=3, để khi không đủ replica Kafka từ chối ghi thay vì âm thầm giảm độ bền.

🇬🇧 English:

The ISR is the set of replicas that are caught up with the leader. acks=all alone is not enough: if the ISR shrinks to only the leader, acks=all means one copy. We set min.insync.replicas, for example 2 with RF=3, so Kafka rejects writes when there are not enough replicas instead of silently lowering durability.

Q3. Unclean leader election là gì?

🇻🇳 Tiếng Việt:

Là cho phép một replica không nằm trong ISR trở thành leader khi mọi replica ISR đều chết. Partition sống lại nhanh nhưng mất các record replica đó chưa có. Mặc định tắt; với dữ liệu quan trọng như đơn hàng em giữ tắt, ưu tiên consistency hơn availability.

🇬🇧 English:

It allows an out-of-sync replica to become leader when all in-sync replicas are gone. The partition comes back quickly but loses records that replica never received. It is disabled by default, and for critical data like orders I keep it disabled, preferring consistency over availability.

Cách nhớ ngắn gọn

• Broker lưu partition; consumer đọc partition. Hai chuyện khác nhau.
• Mỗi partition: 1 leader + follower; ISR = follower theo kịp.
• Bộ ba bền: RF=3 + acks=all + min.insync.replicas=2.
• Unclean election = sống nhanh nhưng mất dữ liệu. KRaft thay ZooKeeper (4.0 bỏ hẳn).