Kafka 4 — Lưu trữ, Retention & Độ tin cậy

Ở ba phần trước của series, chúng ta đã nhìn Kafka từ góc độ producer/consumer và mô hình partition. Phần này đi sâu vào phần "cứng" nhất: Kafka lưu dữ liệu như thế nào trên đĩa, giữ dữ liệu bao lâu, và làm sao để không mất dữ liệu khi broker chết. Đây là những cấu hình quyết định giữa một hệ thống "chạy được" và một hệ thống "không mất một xu nào" — điều tối quan trọng khi topic của bạn đang mang số dư tài khoản khách hàng.

Chúng ta sẽ dùng xuyên suốt một ví dụ thực tế: một topic changelog tên account-balance-changelog lưu số dư mới nhất của từng tài khoản, với key là account_id. Đây là ứng dụng kinh điển của log compaction.

Kafka lưu log trên đĩa như thế nào

Kafka không lưu message trong một cấu trúc dữ liệu phức tạp như B-tree của một RDBMS. Bản chất mỗi partition là một commit log chỉ ghi thêm (append-only) — một chuỗi bytes tuần tự trên đĩa. Đây là lý do Kafka đạt throughput cực cao: ghi tuần tự (sequential write) nhanh hơn ghi ngẫu nhiên (random write) hàng chục lần, kể cả trên SSD, và tận dụng tối đa page cache của OS.

Partition → Segment files

Một partition không phải là một file khổng lồ duy nhất. Nếu vậy, việc xoá dữ liệu cũ sẽ rất tốn kém (phải truncate đầu file). Thay vào đó, mỗi partition được chia thành nhiều segment — mỗi segment là một tập file trên đĩa.

Trong thư mục dữ liệu của broker (log.dirs), bạn sẽ thấy một thư mục cho mỗi partition, ví dụ account-balance-changelog-0/, chứa các bộ ba file:

account-balance-changelog-0/
├── 00000000000000000000.log      # dữ liệu message thật
├── 00000000000000000000.index    # ánh xạ offset → vị trí byte trong .log
├── 00000000000000000000.timeindex # ánh xạ timestamp → offset
├── 00000000000000368214.log      # segment tiếp theo
├── 00000000000000368214.index
├── 00000000000000368214.timeindex
└── leader-epoch-checkpoint

• File .log: chứa các message thật (record batch). Tên file chính là base offset — offset của message đầu tiên trong segment.
• File .index: index thưa (sparse) ánh xạ offset tương đối → vị trí byte trong file .log. Không index từng message mà index cách nhau vài KB (điều khiển bởi log.index.interval.bytes, mặc định 4096). Khi consumer muốn đọc từ offset X, broker tìm nhị phân trong .index để nhảy gần đúng, rồi quét tuyến tính phần còn lại.
• File .timeindex: ánh xạ timestamp → offset, phục vụ truy vấn theo thời gian và retention theo thời gian.

Active segment

Trong mỗi partition, chỉ có một segment duy nhất đang được ghi — gọi là active segment. Mọi message mới đều append vào đây. Các segment còn lại là "đóng" (closed) — chỉ đọc.

Kafka cuộn (roll) sang segment mới khi một trong các điều kiện xảy ra:

Điểm cực kỳ quan trọng: retention và compaction chỉ tác động lên các segment đã đóng, không bao giờ chạm vào active segment. Vì vậy dữ liệu mới nhất luôn an toàn.

Retention — Giữ dữ liệu bao lâu

Với cleanup.policy=delete (mặc định), Kafka xoá toàn bộ segment cũ khi vượt ngưỡng. Có hai chiều điều khiển:

Retention theo thời gian

# Giữ 7 ngày (đơn vị ms)
retention.ms=604800000

Kafka kiểm tra timestamp của message mới nhất trong một segment đã đóng. Nếu segment đó chỉ chứa message cũ hơn retention.ms so với hiện tại, cả segment bị đánh dấu xoá. Lưu ý: xoá theo cả segment, không theo từng message — nên dữ liệu có thể tồn tại lâu hơn retention.ms một chút cho đến khi segment cuối cùng chứa nó bị xoá.

Các tham số broker liên quan (bị override bởi cấu hình topic):

• log.retention.ms / .minutes / .hours (mặc định 168 giờ = 7 ngày)
• log.retention.check.interval.ms (mặc định 5 phút): tần suất thread LogCleaner/retention quét.
• log.segment.delete.delay.ms (mặc định 60s): độ trễ trước khi xoá file thật, để consumer đang đọc kịp hoàn tất.

Retention theo dung lượng

# Giữ tối đa 50 GB mỗi partition
retention.bytes=53687091200

Rất quan trọng: retention.bytes áp dụng cho mỗi partition, không phải cho cả topic. Một topic 10 partition với retention.bytes=50GB có thể chiếm tối đa ~500 GB (chưa tính replication).

Khi cả hai ngưỡng được đặt, segment bị xoá nếu vi phạm bất kỳ ngưỡng nào (điều kiện OR). Đặt retention.ms=-1 hoặc retention.bytes=-1 để tắt ngưỡng tương ứng (giữ vô hạn theo chiều đó).

Log Compaction — Giữ bản ghi mới nhất theo key

Retention delete phù hợp cho dữ liệu dạng event/sự kiện ("giao dịch lúc 10:05"). Nhưng topic account-balance-changelog mang ngữ nghĩa trạng thái: "số dư hiện tại của account 123 là 5.000.000đ". Với dữ liệu trạng thái, ta không muốn xoá theo thời gian — vì nếu account 123 không thay đổi trong 8 ngày, retention 7 ngày sẽ xoá mất số dư của nó, và một consumer mới khôi phục state sẽ tưởng account đó không tồn tại.

Log compaction giải quyết điều này: với mỗi key, Kafka đảm bảo giữ lại ít nhất bản ghi (value) mới nhất. Các bản cũ hơn cùng key sẽ bị dọn dần trong nền.

cleanup.policy=compact

Kết quả: log sau compaction đảm bảo bất kỳ consumer nào đọc từ đầu đến hết đều thấy giá trị cuối cùng cho mọi key — chính là một snapshot của bảng trạng thái. Đây là nền tảng của KTable trong Kafka Streams, changelog topic của state store, và nhiều pipeline CDC (Change Data Capture).

Cấu trúc log sau compaction

Compaction chia log thành hai vùng:

• Tail (đuôi): phần đã được compact — mỗi key xuất hiện tối đa một lần (bản mới nhất tại thời điểm compact).
• Head (đầu): phần chưa compact (bao gồm active segment) — có thể có nhiều bản trùng key.

Quan trọng: offset không bị đánh lại. Sau khi xoá bản cũ, các offset trở nên "thưa" (có lỗ hổng), nhưng offset của các bản còn lại không đổi. Consumer vẫn seek theo offset bình thường.

Tombstone — Xoá một key

Muốn "xoá" một account khỏi changelog (ví dụ tài khoản đóng), producer gửi một message với key đó và value = null. Đây gọi là tombstone. Compaction sẽ:

1. Giữ tombstone lại một thời gian để consumer kịp nhận tín hiệu "key này đã bị xoá".
2. Sau delete.retention.ms (mặc định 24 giờ), xoá luôn cả tombstone.

Nếu tombstone bị xoá quá sớm, một consumer khởi động muộn sẽ không bao giờ biết key đó đã bị xoá — nên delete.retention.ms cần đủ dài so với thời gian tối đa một consumer có thể offline.

Điều khiển thời điểm compaction

Compaction không chạy tức thì — nó do các thread nền (log.cleaner.threads) xử lý dựa trên dirty ratio:

# Bật cleaner (mặc định bật)
log.cleaner.enable=true
# Tỉ lệ "bẩn" tối thiểu để kích hoạt compact một partition (50%)
min.cleanable.dirty.ratio=0.5
# Message phải "ở" ít nhất bao lâu trước khi được phép compact
min.compaction.lag.ms=0
# Ép compact tối đa sau khoảng này kể cả chưa đủ dirty ratio
max.compaction.lag.ms=9223372036854775807

Vì min.cleanable.dirty.ratio mặc định 0.5, một key vừa cập nhật có thể vẫn còn nhiều bản trùng trong head cho đến khi cleaner chạy. Consumer đọc từ đầu vẫn sẽ nhận đúng bản mới nhất (vì nó ở cuối), nhưng có thể nhận cả bản cũ trước đó — code state store phải luôn "ghi đè theo key", điều mà KTable làm sẵn.

Kết hợp compact + delete

Có thể đặt cả hai:

cleanup.policy=compact,delete

Khi đó Kafka vừa giữ bản mới nhất theo key (compact) vừa xoá segment quá hạn theo retention.ms/retention.bytes. Hữu ích khi bạn muốn changelog nhưng không muốn giữ vô hạn (ví dụ chấp nhận mất key rất cũ, không hoạt động).

So sánh delete vs compact

Độ bền & Replication chuyên sâu

Lưu trên đĩa một broker vẫn chưa đủ bền — broker có thể chết, đĩa có thể hỏng. Kafka nhân bản (replicate) mỗi partition sang nhiều broker.

Leader, Follower và ISR

Mỗi partition có một leader (nhận mọi write/read) và các follower sao chép dữ liệu từ leader. Tập các replica đang bắt kịp leader gọi là ISR (In-Sync Replicas).

Một follower được coi là "in-sync" nếu nó fetch được đến cuối log của leader trong vòng replica.lag.time.max.ms (mặc định 30s). Nếu follower tụt hậu quá lâu (mạng chậm, GC, đĩa chậm), nó bị loại khỏi ISR — ISR "co lại". Khi nó bắt kịp lại, nó được thêm vào — ISR "giãn ra". ISR co giãn liên tục là cơ chế tự bảo vệ của Kafka.

High Watermark (HW)

High watermark là offset cao nhất mà mọi replica trong ISR đã sao chép. Consumer chỉ đọc được message có offset < HW — tức chỉ những message đã được nhân bản an toàn. Message đã ghi vào leader nhưng chưa qua HW gọi là chưa "committed" và không hiển thị cho consumer. Điều này ngăn consumer đọc dữ liệu có thể bị mất nếu leader chết trước khi replicate.

Leader Epoch

Trước đây, dựa hoàn toàn vào HW để truncate dữ liệu khi đổi leader có thể gây mất/phân kỳ dữ liệu trong một số kịch bản đổi leader liên tiếp. Kafka giải quyết bằng leader epoch: mỗi lần bầu leader mới, epoch tăng lên. Mỗi message được gắn với epoch của leader ghi nó. Khi một follower phục hồi, nó hỏi leader "tại epoch X, offset nào là biên?" để truncate chính xác thay vì mù quáng theo HW. File leader-epoch-checkpoint bạn thấy trong thư mục partition chính là để lưu lịch sử này. Đây là cơ chế nền tảng đảm bảo tính nhất quán khi failover.

Unclean Leader Election — Nguy cơ mất dữ liệu

Giả sử tất cả replica trong ISR đều chết, chỉ còn một replica ngoài ISR (đang tụt hậu, thiếu dữ liệu mới nhất) còn sống. Có hai lựa chọn:

• unclean.leader.election.enable=false (mặc định, khuyến nghị): partition offline cho đến khi một replica trong ISR sống lại. Ưu tiên độ bền — không mất dữ liệu, nhưng mất tính sẵn sàng.
• unclean.leader.election.enable=true: bầu replica ngoài ISR làm leader. Partition online ngay, nhưng những message chỉ có trên các replica đã chết sẽ mất vĩnh viễn. Ưu tiên sẵn sàng, hy sinh độ bền.

Với topic số dư tài khoản, tuyệt đối để false. Mất một cập nhật số dư là không thể chấp nhận.

acks + min.insync.replicas — Bộ đôi chống mất dữ liệu

Đây là điểm cốt lõi. Hai tham số phải phối hợp:

• acks (phía producer): 0 (bắn và quên), 1 (chỉ leader ghi xong), all/-1 (mọi replica trong ISR ghi xong).
• min.insync.replicas (phía topic/broker): số replica in-sync tối thiểu để một write với acks=all được chấp nhận. Nếu ISR nhỏ hơn ngưỡng này, producer nhận lỗi NotEnoughReplicasException — write bị từ chối thay vì âm thầm mất.

Chỉ acks=all kết hợp min.insync.replicas >= 2 mới thực sự chống mất dữ liệu. acks=all mà min.insync.replicas=1 thì khi ISR co xuống còn mỗi leader, "all" chỉ còn nghĩa là "một mình leader" — leader chết là mất.

Công thức chuẩn cho dữ liệu quan trọng:

replication.factor=3
min.insync.replicas=2
# producer: acks=all

Với cấu hình này: cần ít nhất 2 replica ghi mới ack. Cụm chịu được 1 broker chết mà vẫn ghi được (còn 2 in-sync). Nếu 2 broker chết cùng lúc, ISR còn 1 < 2 → write bị từ chối (fail fast, không mất dữ liệu), consumer vẫn đọc được dữ liệu đã committed. Tam giác RF=3, min.insync=2, acks=all là tiêu chuẩn công nghiệp.

Durability vs Availability — Đánh đổi

Không có bữa trưa miễn phí. Cấu hình càng bền, càng dễ mất tính sẵn sàng cho write:

Ngoài ra, để tránh trùng lặp khi retry, bật idempotent producer (enable.idempotence=true) — từ Kafka mới nó là mặc định và tự đặt acks=all.

Tiered Storage — Tách lưu trữ nóng/lạnh

Từ Kafka 3.6 (KIP-405) trở đi, tiered storage trở nên khả dụng cho sản xuất. Ý tưởng: tách phần "nóng" (dữ liệu gần đây, đọc thường xuyên) khỏi phần "lạnh" (dữ liệu cũ).

• Local tier: các segment gần đây nằm trên đĩa broker (SSD nhanh, đắt).
• Remote tier: segment cũ được đẩy sang lưu trữ đối tượng rẻ (S3, GCS, HDFS...).

remote.storage.enable=true
# giữ bao lâu trên đĩa local trước khi đẩy lên remote
local.retention.ms=86400000
# tổng retention (cả local + remote)
retention.ms=2592000000

Lợi ích: giữ dữ liệu rất lâu (nhiều tháng) với chi phí thấp, giảm áp lực đĩa broker, và giúp thêm/mở rộng broker nhanh hơn (ít dữ liệu cần rebalance vì phần lạnh không nằm trên broker). Consumer đọc dữ liệu cũ được phục vụ trong suốt từ remote tier, dù độ trễ cao hơn đọc từ đĩa local. Đây là hướng đi lớn giúp Kafka đóng vai trò lưu trữ dài hạn, không chỉ là bộ đệm.

Dung lượng & Sizing

Ước lượng dung lượng đĩa cần thiết:

Dung lượng ≈ (throughput byte/s) × (retention giây) × (replication.factor) × (1 + hệ số overhead)

Ví dụ: 10 MB/s × 604800s (7 ngày) × 3 (RF) ≈ 17,4 TB thô cho một topic, chia trên các broker. Cộng thêm ~10% cho index và overhead. Với compaction, dung lượng phụ thuộc số key duy nhất và kích thước value trung bình, không phụ thuộc số lần cập nhật — thường nhỏ hơn nhiều so với delete.

Nguyên tắc: dành headroom (thường giữ đĩa dưới 60-70%) để có chỗ cho segment mới, log cleaner cần không gian tạm khi compact, và burst traffic.

Ví dụ cấu hình topic thực tế

Topic changelog số dư — cần compaction, độ bền cao:

kafka-topics --bootstrap-server broker:9092 \
  --create \
  --topic account-balance-changelog \
  --partitions 12 \
  --replication-factor 3 \
  --config cleanup.policy=compact \
  --config min.insync.replicas=2 \
  --config delete.retention.ms=86400000 \
  --config min.cleanable.dirty.ratio=0.5 \
  --config segment.bytes=536870912 \
  --config unclean.leader.election.enable=false

Topic log giao dịch — dạng event, giữ 30 ngày, độ bền cao:

kafka-topics --bootstrap-server broker:9092 \
  --create \
  --topic transactions-log \
  --partitions 24 \
  --replication-factor 3 \
  --config cleanup.policy=delete \
  --config retention.ms=2592000000 \
  --config retention.bytes=-1 \
  --config min.insync.replicas=2 \
  --config unclean.leader.election.enable=false

Xem và sửa cấu hình sau khi tạo:

# Xem cấu hình hiện tại
kafka-configs --bootstrap-server broker:9092 \
  --entity-type topics --entity-name account-balance-changelog --describe

# Đổi retention một topic đang chạy
kafka-configs --bootstrap-server broker:9092 \
  --entity-type topics --entity-name transactions-log \
  --alter --add-config retention.ms=1209600000

Tóm tắt

• Mỗi partition là commit log append-only, chia thành segment (.log, .index, .timeindex); chỉ active segment được ghi, retention/compaction chỉ chạm segment đã đóng.
• Retention delete xoá cả segment theo retention.ms (thời gian) hoặc retention.bytes (dung lượng, per-partition) — điều kiện OR.
• Log compaction (cleanup.policy=compact) giữ bản mới nhất theo key, dùng cho KTable/CDC/changelog trạng thái; tombstone (value=null) để xoá key, giữ theo delete.retention.ms.
• Độ bền dựa trên replication: leader/follower/ISR co giãn, high watermark che dữ liệu chưa nhân bản, leader epoch đảm bảo truncate đúng khi failover.
• Chống mất dữ liệu = acks=all + min.insync.replicas>=2 + RF=3 + unclean.leader.election=false; đó là đánh đổi thiên về độ bền, hy sinh chút sẵn sàng.
• Tiered storage tách nóng/lạnh để lưu dài hạn với chi phí thấp.

Tự kiểm tra

1. Tại sao retention xoá theo cả segment thay vì từng message, và điều đó ảnh hưởng thế nào đến việc dữ liệu có thể "sống" lâu hơn retention.ms?
2. Với topic account-balance-changelog, giải thích tại sao cleanup.policy=compact phù hợp hơn delete, và điều gì xảy ra nếu bạn quên đặt key cho message.
3. Một tombstone là gì, và điều gì hỏng nếu delete.retention.ms quá ngắn so với thời gian consumer offline?
4. Cụm có RF=3, min.insync.replicas=2, acks=all. Ghi/đọc còn hoạt động thế nào khi 1 broker chết? Khi 2 broker chết?
5. Giải thích vì sao acks=all mà không đặt min.insync.replicas>=2 vẫn có thể mất dữ liệu.
6. High watermark và leader epoch mỗi cái giải quyết vấn đề gì trong quá trình failover?

Đọc tiếp

Kafka 5 — Kafka Connect