Bài 3: Kafka Producer Nâng Cao — Throughput, Batching, Acks, Idempotent và Exactly-Once

Lời Mở Đầu: Kafka Producer — Không Đơn Giản Như Bạn Nghĩ

Gọi producer.send(record) trông có vẻ đơn giản. Tuy nhiên, bên dưới lớp API đó là một cỗ máy tinh vi với hàng chục tham số cấu hình, mỗi tham số ảnh hưởng trực tiếp đến hiệu năng hệ thống. Cụ thể, throughput xác định bao nhiêu message/giây bạn có thể đẩy, latency quyết định mất bao lâu từ lúc gọi send() đến khi Broker confirm, durability đảm bảo data không bị mất khi mạng đứt hoặc Broker crash, và ordering kiểm soát thứ tự message có được đảm bảo không. Đây chính là bài toán tam giác bất khả thi của distributed systems — và bài này sẽ giúp bạn chọn đúng góc tam giác đó cho từng use case.

Nếu bạn chưa nắm kiến trúc Partition và Segment, hãy đọc qua Bài 2: Giải Phẫu Kiến Trúc Kafka trước để hiểu tại sao các config bên dưới lại hoạt động theo cách chúng hoạt động.

1. Kiến Trúc Nội Tại Của Kafka Producer

Trước khi nói về config, hãy hiểu Producer hoạt động như thế nào bên trong:

Application Thread                    Network Thread
      │                                     │
      │ producer.send(record)               │
      ▼                                     │
┌──────────────────────────┐                │
│    Serializer            │                │
│  (Key + Value → bytes)   │                │
└──────────┬───────────────┘                │
           ▼                                │
┌──────────────────────────┐                │
│    Partitioner           │                │
│  (Decide partition)      │                │
└──────────┬───────────────┘                │
           ▼                                │
┌──────────────────────────┐                │
│   RecordAccumulator      │────── Batch ──►│ Sender Thread
│   (In-memory buffer)     │    (khi đủ     │     │
│                          │     điều kiện) │     ▼
│  Partition 0: [M1,M2,M3] │                │  TCP to Broker
│  Partition 1: [M4,M5]    │                │
│  Partition 2: [M6]       │                │
└──────────────────────────┘                │

Application Thread và Network/Sender Thread chạy hoàn toàn song song. Application Thread ghi vào buffer, trong khi Sender Thread tự động gom batch và gửi đi — nhờ đó Application không bao giờ bị block chờ network.

2. Batching — Nghệ Thuật Gom Lô

2.1 Vì Sao Cần Batch?

Không có batching (mỗi message = 1 TCP request):
[M1] → [TCP overhead 40B header] → Broker
[M2] → [TCP overhead 40B header] → Broker
[M3] → [TCP overhead 40B header] → Broker
→ 3 round-trips, 3x overhead

Với batching:
[M1+M2+M3] → [TCP overhead 40B header] → Broker
→ 1 round-trip, 1x overhead + messages nén lại

Kết quả là throughput tăng đột biến — không phải vì Kafka xử lý nhanh hơn mà vì chi phí mạng giảm đi đáng kể theo tỷ lệ batch size.

2.2 Cấu Hình Batch Tinh Chỉnh

# Kích thước tối đa của một batch cho một Partition (bytes)
batch.size=65536          # 64KB (mặc định 16KB, tăng lên cho high-throughput)

# Thời gian đợi thêm records vào batch trước khi gửi
linger.ms=10              # Đợi 10ms (mặc định 0 = gửi ngay)

# Tổng kích thước buffer RAM của Producer (tất cả partitions)
buffer.memory=67108864    # 64MB (mặc định 32MB)

# Thuật toán nén batch
compression.type=lz4      # Tùy chọn: none, gzip, snappy, lz4, zstd

2.3 Trade-off Latency vs Throughput

linger.ms là tham số quan trọng nhất khi tinh chỉnh Producer. Với linger.ms=0 (mặc định), message được gửi ngay lập tức — latency thấp nhất nhưng throughput thấp vì batch nhỏ và nhiều round-trip. Đây là lựa chọn phù hợp cho hệ thống trading hay real-time alert.

Ngược lại, khi tăng lên linger.ms=5~20ms, Producer đợi tối đa 20ms để gom đủ batch. Latency tăng nhẹ nhưng throughput có thể tăng 5–10x — mức trade-off thường chấp nhận được cho event streaming và log aggregation. Hơn nữa, nếu bạn đang chạy batch processing hay ETL pipeline, linger.ms=100ms+ sẽ gom được batch cực lớn và đạt throughput tối đa, dù latency cao hơn nhiều.

2.4 Compression — Vũ Khí Bí Mật

Benchmark thực tế với JSON payload (1000 records):

Không nén:  100MB → Network: 100MB,  CPU: thấp,   Latency: thấp
snappy:     100MB → Network: 45MB,   CPU: thấp,   Latency: +1ms
lz4:        100MB → Network: 40MB,   CPU: thấp,   Latency: +0.5ms
gzip:       100MB → Network: 25MB,   CPU: cao,    Latency: +5ms
zstd:       100MB → Network: 22MB,   CPU: trung,  Latency: +2ms

Nhìn chung, lz4 là khuyến nghị cho production vì cân bằng tốt nhất giữa tỷ lệ nén, CPU overhead và latency. Tuy nhiên, nếu network đang là bottleneck chính, zstd cho tỷ lệ nén tốt hơn với CPU overhead chấp nhận được.

3. Acks — Sự Đánh Đổi Giữa Tốc Độ và Sự Toàn Vẹn

Acks=0: Fire and Forget

Producer ──SEND──► Broker (Không chờ response)
             │
             └─► Producer tiếp tục ngay lập tức

Throughput: ████████████ Tối đa
Durability: █           Không đảm bảo gì
Use case:   Metrics tracking, clickstream (mất vài event không sao)

Acks=1: Leader Confirm

Producer ──SEND──► Leader Broker
                       │
                       │ (Ghi vào local disk)
                       │
Producer ◄──ACK──────── Leader

Throughput: █████████   Cao
Durability: ████        Trung bình
Risk: Leader crash TRƯỚC khi sync sang Follower → MẤT DATA
Use case: Non-critical event stream

Acks=all (hoặc acks=-1): Full ISR Confirm

Producer ──SEND──► Leader Broker
                       │
                       ├──REPLICATE──► Follower 1 (ghi xong → ACK Leader)
                       ├──REPLICATE──► Follower 2 (ghi xong → ACK Leader)
                       │
Producer ◄──ACK──────── Leader (khi tất cả ISR confirm)

Throughput: █████       Thấp hơn (phụ thuộc ISR lag)
Durability: ████████████ Tối đa
Use case: Giao dịch tài chính, order processing, dữ liệu quan trọng

Phương Trình Acks=all Kết Hợp min.insync.replicas

Cấu hình:
  replication.factor = 3  (3 bản sao)
  min.insync.replicas = 2 (cần ít nhất 2 bản ghi thành công)
  acks = all

Kịch bản:
  ✅ 3/3 ISR sống: Hoạt động bình thường
  ✅ 2/3 ISR sống: Hoạt động bình thường (đủ min.insync.replicas)
  ❌ 1/3 ISR sống: NotEnoughReplicasException → Producer retry
  ❌ 0/3 ISR sống: Cluster down, không ghi được

4. Retry và Idempotent — Chống Trùng Lặp Không Cần Code

4.1 Vấn Đề: Tại Sao Retry Gây Duplicate?

Bước 1: Producer gửi M1 → Broker nhận, ghi disk
Bước 2: Broker gửi ACK → ACK bị mất giữa đường (network hiccup)
Bước 3: Producer timeout, nghĩ M1 chưa tới → Retry: gửi M1 lần 2
Bước 4: Broker nhận M1 lần 2 → Ghi thêm một bản nữa → DUPLICATE!

Kết quả là thanh toán bị charge 2 lần, inventory bị trừ 2 lần — một lỗi nghiêm trọng mà không thể phát hiện từ phía Producer vì từ góc nhìn của nó, cả hai lần gửi đều “thành công”.

4.2 Giải Pháp: enable.idempotence=true

Producer nhận PID (Producer ID) = 7834 từ Broker khi khởi động

Gửi lần 1:  Message { PID=7834, SeqNum=1, data=M1 } → Broker ghi OK
Network glitch → ACK mất
Gửi lần 2:  Message { PID=7834, SeqNum=1, data=M1 } → Broker kiểm tra:
            "PID=7834, SeqNum=1 đã tồn tại!" → DROP, gửi ACK lại
            
→ Chỉ có 1 bản M1 trong Partition. Exactly-Once ở mức Broker!

# Kể từ Kafka 3.0, đây là DEFAULT:
enable.idempotence=true
max.in.flight.requests.per.connection=5  # Tối đa 5 request chưa ACK
retries=2147483647                        # Retry vô hạn
acks=all                                  # Bắt buộc khi idempotence=true

Idempotent chỉ đảm bảo exactly-once cho một Producer → một Partition. Tuy nhiên, nếu bạn cần ghi vào nhiều Partition — hay đọc từ Partition này và ghi sang Partition khác — là atomic, bạn cần Kafka Transactions:

4.3 Kafka Transactions — Exactly-Once Across Topics

Properties props = new Properties();
props.put("transactional.id", "order-processor-1");

KafkaProducer<String, String> producer = new KafkaProducer<>(props);
producer.initTransactions();

try {
    producer.beginTransaction();
    
    producer.send(new ProducerRecord<>("order-events", key, orderJson));
    producer.send(new ProducerRecord<>("audit-log", key, auditJson));
    producer.send(new ProducerRecord<>("billing-queue", key, billingJson));
    
    producer.commitTransaction(); // Atomic commit
} catch (Exception e) {
    producer.abortTransaction(); // Atomic rollback
}

Thêm vào đó, Kafka Transactions còn hỗ trợ read-process-write pattern (consume → transform → produce) với exactly-once semantics khi kết hợp với isolation.level=read_committed ở phía Consumer. Đây là nền tảng của Kafka Streams exactly-once processing.

5. Cấu Hình Producer Theo Use Case

Use Case 1: High-Throughput Event Streaming

# Tối ưu throughput, chấp nhận latency vài chục ms
batch.size=131072          # 128KB
linger.ms=20               # Đợi 20ms gom batch
compression.type=lz4       # Nén tốt, CPU thấp
acks=1                     # Leader confirm đủ rồi
buffer.memory=134217728    # 128MB buffer
max.in.flight.requests.per.connection=5

Use Case 2: Financial Transaction (Không Được Mất Data)

# Tối ưu durability
batch.size=16384           # 16KB (nhỏ, gửi nhanh)
linger.ms=0                # Gửi ngay
compression.type=none      # Không nén (giảm CPU)
acks=all                   # Phải có tất cả ISR confirm
enable.idempotence=true    # Chống duplicate
retries=2147483647         # Retry vô hạn
delivery.timeout.ms=120000 # 2 phút timeout tổng
transactional.id=tx-producer-1

Use Case 3: Metrics/Logging (Chịu Mất Một Ít)

# Tối ưu latency, chấp nhận loss
batch.size=65536
linger.ms=5
compression.type=snappy
acks=0                     # Fire and forget

Để biết cách Consumer đọc và xử lý những message này theo từng chiến lược, hãy xem tiếp Bài 4: Kafka Consumer — Group, Rebalance và Offset

Kết Luận & Takeaway

         Throughput
            ▲
           /│\
          / │ \
         /  │  \
        ────┼────
       /    │    \
   Latency──┘──Durability

→ Không thể tối ưu cả 3 cùng lúc.
→ Chọn dựa trên business requirement!

💬 Bài toán thực tế: Bạn đang xây dựng tính năng “Lưu vết hành vi người dùng” (clickstream) với 50,000 event/giây và chấp nhận mất tối đa 0.01% event. Bạn sẽ chọn acks bao nhiêu? linger.ms bao nhiêu? Và có bật enable.idempotence không? Lý giải bên dưới nhé!