Giao diện
Kafka
Kafka không phải một hàng đợi theo nghĩa thông thường. Nó là một log phân tán: message được ghi nối vào cuối, giữ lại theo thời hạn cấu hình chứ không bị xoá khi có người đọc, và mỗi bên đọc tự nhớ mình đã đọc tới đâu. Nhờ vậy nhiều hệ thống đọc cùng một luồng sự kiện độc lập với nhau, đọc lại từ quá khứ được, và throughput rất cao. Trang này giải thích partition và thứ tự, consumer group, replication, các mức đảm bảo giao nhận, cách giữ message, và khi nào nên chọn Kafka thay vì một hàng đợi truyền thống (Hàng đợi message).
Topic, partition, offset
Topic là một luồng message cùng loại, ví dụ orders. Mỗi topic chia thành nhiều partition. Mỗi partition là một log riêng: message mới ghi nối vào cuối và nhận một số thứ tự tăng dần gọi là offset. Consumer đọc theo offset và tự nhớ vị trí của mình, nên đọc xong không làm message biến mất.
Partition là đơn vị để Kafka chia tải. Các partition của một topic nằm rải trên nhiều broker (máy chủ Kafka), nên thêm broker là thêm chỗ ghi và chỗ đọc. Partition cũng là giới hạn của việc đọc song song, như mục consumer group dưới đây sẽ cho thấy.
Kafka nhanh dù ghi xuống đĩa, vì nó chỉ ghi nối tiếp vào cuối file, và để đọc thì dựa vào page cache của hệ điều hành. Tài liệu thiết kế của Kafka giải thích rằng message được gửi thẳng từ page cache ra socket bằng system call sendfile, không chép qua bộ nhớ của process, nên khi consumer đọc theo kịp thì gần như không có lần đọc đĩa nào. Cách này không dùng được khi bật TLS, vì mã hoá phải làm ở user space.
Key, partition và thứ tự
Producer chọn partition cho mỗi message theo key: băm key rồi chia lấy dư cho số partition. Cùng key thì luôn vào cùng partition, và trong một partition thì message được giữ đúng thứ tự ghi. Vì vậy:
- Thứ tự chỉ được đảm bảo trong một partition, không có thứ tự chung cho cả topic. Muốn các sự kiện của một đơn hàng được xử lý đúng thứ tự thì dùng mã đơn làm key. Trên lab, 4 đơn mỗi đơn 4 sự kiện gửi xen kẽ: đọc ra thì sự kiện của từng đơn đúng thứ tự, nhưng thứ tự chung trên topic thì không.
- Ít key thì tải lệch. Cũng trong lab đó, 3 trên 4 đơn rơi vào cùng một partition, còn một partition không có message nào. Một key rất nóng (một khách hàng lớn, một sản phẩm đang flash sale) làm một partition quá tải trong khi các partition khác rảnh.
- Message không có key được rải đều cho các partition, và không có đảm bảo thứ tự nào.
Đổi số partition làm key đổi chỗ. Vì phép chia lấy dư dùng số partition, tăng số partition thì phần lớn key chuyển sang partition khác (Hash mod N). Trên lab, tăng từ 3 lên 6 partition thì 53% key đổi partition. Message cũ của một key nằm ở partition cũ, message mới ở partition mới, nên thứ tự của key đó không còn được giữ trong lúc chuyển. Kafka cũng không cho giảm số partition. Vì vậy nên chọn số partition dư ra ngay từ đầu.
Cùng key, khác client, khác partition. Client Java băm key bằng murmur2. librdkafka (thư viện C mà client của Confluent cho Node.js, Python, Go dựng trên đó) mặc định băm bằng CRC32 (consistent_random). Nếu một service Java và một service Python cùng gửi vào một topic với cấu hình mặc định, cùng một key có thể vào hai partition khác nhau, và thứ tự theo key bị phá mà không có lỗi nào. Client @confluentinc/kafka-javascript ở chế độ tương thích KafkaJS thì mặc định dùng murmur2_random, giống Java. Khi trộn nhiều ngôn ngữ, đặt partitioner giống nhau ở mọi nơi.
Retry có thể đảo thứ tự. Nếu producer gửi lô 1, lô 2, lô 1 lỗi rồi được gửi lại sau lô 2, thì thứ tự đổi. Tài liệu librdkafka ghi rõ retry có thể làm đảo thứ tự nếu không bật enable.idempotence.
Consumer group
Nhiều consumer cùng group.id lập thành một consumer group. Kafka chia partition cho các consumer trong group, và mỗi partition chỉ giao cho đúng một consumer trong group. Nhờ vậy message trong một partition được một consumer xử lý theo đúng thứ tự, và group chia tải mà không xử lý trùng.
Hệ quả là số consumer làm việc song song không vượt quá số partition. Trên lab, topic 6 partition, mỗi message xử lý 10 ms, 600 message:
| Consumer | Đọc hết sau | Partition mỗi consumer giữ |
|---|---|---|
| 1 | 6,6 s | 6 |
| 3 | 2,4–3,9 s | 2, 2, 2 |
| 6 | 1,6–3,0 s | 1 mỗi consumer |
| 8 | 1,3–1,6 s | 1 cho 6 consumer, 2 consumer không có gì |
Consumer thứ 7, 8 ngồi không. Muốn đọc song song hơn nữa thì phải tăng số partition, hoặc xử lý nhanh hơn trong mỗi consumer (gom lô, xử lý song song các key khác nhau trong một partition).
Nhiều group đọc độc lập. Mỗi group có vị trí đọc riêng, nên service gửi email và service thống kê cùng đọc topic orders mà không ảnh hưởng nhau. Trên lab, hai group cùng đọc một topic 600 message thì group nào cũng nhận đủ 600. Đây là khác biệt lớn nhất với một hàng đợi, nơi một message đã giao cho người này thì người khác không thấy nữa.
Lag là khoảng cách giữa offset mới nhất của partition và offset group đã đọc tới. Lag tăng đều nghĩa là consumer xử lý chậm hơn tốc độ message đến, và đây là chỉ số quan trọng nhất cần theo dõi và cảnh báo cho một consumer group.
Rebalance và consumer chết
Khi một consumer vào hay rời group, group rebalance: chia lại partition cho các thành viên. Với cách chia cũ (eager), mọi consumer ngừng đọc, trả hết partition, rồi nhận lại phần mới. Cách cooperative-sticky chỉ chuyển những partition cần chuyển. Kafka 4.x có thêm giao thức rebalance mới (KIP-848), bật bằng group.protocol=consumer, do broker điều phối thay vì client. Trên localhost, cả group vào lại rất nhanh nên lab chỉ đo được khoảng dừng 11–101 ms. Ngoài thực tế, khoảng dừng phụ thuộc consumer chậm nhất trong group, và rebalance liên tục (consumer khởi động lại liên tục, deploy từng pod một) làm cả group đọc chậm hẳn.
Phần dễ bị bỏ sót là khi consumer chết đột ngột (process bị kill, máy mất điện). Nó không kịp báo rời group, nên group vẫn coi nó là thành viên và các partition của nó không ai đọc, cho tới khi hết session.timeout.ms (mặc định 45 giây ở librdkafka). Trên lab, đặt session timeout 10 giây và kill -9 một trong ba consumer: message gửi vào hai partition của nó phải chờ 10,3–10,6 giây mới được đọc, trong khi lúc bình thường chỉ khoảng 12 ms. Có một giới hạn khác nữa là max.poll.interval.ms (mặc định 5 phút): consumer xử lý một message quá lâu bị coi là treo và bị đẩy ra khỏi group.
Kéo các thanh để thấy giới hạn của số partition, rồi kill một consumer:
Replication: ISR, acks và min.insync.replicas
Mỗi partition có replication factor bản sao trên các broker khác nhau. Một bản là leader, nhận mọi lệnh ghi và đọc; các bản còn lại là follower, liên tục chép từ leader (Replication). Những replica đang theo kịp leader lập thành ISR (in-sync replicas). Follower chậm quá replica.lag.time.max.ms (mặc định 30 giây) bị rút khỏi ISR.
Theo tài liệu thiết kế của Kafka, một message chỉ được coi là đã commit khi mọi replica trong ISR đã có nó, và consumer chỉ đọc được message đã commit. Producer chọn mức chờ bằng acks:
acks | Producer nhận xác nhận khi | Mất message khi |
|---|---|---|
0 | Không chờ gì | Bất cứ lỗi nào |
1 | Leader ghi xong | Leader chết trước khi follower kịp chép |
all (mặc định của librdkafka) | Mọi replica trong ISR đã có | Mọi replica trong ISR cùng mất |
acks=all một mình chưa đủ. Nếu ISR chỉ còn mỗi leader, thì "mọi replica trong ISR" chỉ còn là leader, và acks=all cũng như acks=1. Vì vậy cần thêm min.insync.replicas: số replica tối thiểu phải có trong ISR thì leader mới nhận lệnh ghi với acks=all, nếu không nó trả lỗi NOT_ENOUGH_REPLICAS. Giá trị mặc định là 1, tức không bảo vệ gì, nên phải tự đặt. Cấu hình hay dùng là replication factor 3, min.insync.replicas = 2, acks=all: chịu được một broker chết mà vẫn ghi được và không mất message đã xác nhận.
Lab chạy cụm 3 broker thật, topic replication factor 3:
- Chi phí của
acks=allnhỏ khi các broker gần nhau: p50 khoảng 6,6–7,7 ms so với 6,4–7,1 ms củaacks=1. - ISR thiếu thì bị từ chối ngay, nhưng ứng dụng không thấy ngay. Đặt
min.insync.replicas = 3rồi tắt một follower: broker trả lỗiNOT_ENOUGH_REPLICASsau khoảng 8 ms. Nhưng với cấu hình mặc định, producer coi lỗi này là tạm thời và tự gửi lại, nên ứng dụng chỉ nhận lỗi timeout khi hếtmessage.timeout.ms(lab đặt 5 giây, mặc định của librdkafka là 5 phút). Trong lúc đóacks=1vẫn ghi thành công, tức là đang chấp nhận rủi ro mất dữ liệu. - Follower chết đột ngột làm
acks=allbị kẹt. Follower tắt có thứ tự (docker stop) tự báo rời ISR, và lần gửi chậm nhất chỉ 12–59 ms. Follower bị kill thì vẫn nằm trong ISR, leader chờ nó, cho tới khi controller phát hiện broker đã chết saubroker.session.timeout.ms(mặc định 9 giây). Trên lab, mọi lần gửiacks=alltrong khoảng đó đều phải chờ, lâu nhất 6,9–9,1 giây.
Khi mọi replica trong ISR đều chết, Kafka mặc định không cho một replica ngoài ISR lên làm leader (unclean.leader.election.enable = false), vì replica đó có thể thiếu message đã commit. Partition ngừng phục vụ cho tới khi một replica trong ISR quay lại. Bật tuỳ chọn này là chọn availability thay vì không mất dữ liệu.
Thử gửi vài message với acks=1, kill leader trước khi follower kịp chép, rồi làm lại với acks=all:
acksmin.insync.replicas
Broker 1 (leader), trong ISR
Broker 2 (follower), trong ISR
Broker 3 (follower), trong ISR
Số in đậm ở leader là phần đã commit (dưới high watermark): consumer chỉ đọc được tới đó.
Producer
Gửi vài message với acks=1, kill leader trước khi follower kịp chép log để thấy message đã báo thành công vẫn mất. Rồi làm lại với acks=all.
Kafka 4.x không còn dùng ZooKeeper. Metadata của cụm (ai là leader, ISR gồm ai) do một nhóm controller giữ bằng giao thức KRaft, dựa trên Raft (Consensus). Nhóm controller cũng cần đa số còn sống. Lab chạy mỗi node vừa là broker vừa là controller, nên khi kill 2 trên 3 node, cụm mất đa số controller: không ai thu hẹp ISR được nữa và acks=all chờ mãi. Production thường tách controller ra các node riêng.
Gửi nhanh: gom lô
Producer không gửi từng message một. Nó gom các message cùng partition thành lô, chờ thêm một chút (linger.ms, mặc định 5 ms ở librdkafka) để lô đầy hơn, rồi gửi cả lô trong một request. Hệ quả là cách gọi send quyết định throughput:
Cách gửi (lab, acks=all, message 200 byte) | Throughput |
|---|---|
await send() từng message, xong mới gửi tiếp | 131–141 message/s |
Gửi 1.000 message trong một lần send() | 75.000–91.000 message/s |
Gọi send() cho cả 20.000 message rồi chờ tất cả | 104.000–174.000 message/s |
Chờ từng message thì mỗi message tốn trọn một round trip cộng thời gian chờ replica (khoảng 7 ms), nên chỉ được hơn 100 message mỗi giây. Đây là lỗi hay gặp khi viết producer trong vòng lặp. Vẫn phải chờ kết quả để biết message nào lỗi, nhưng nên gửi nhiều rồi chờ chung, hoặc gom thành lô.
Ngữ nghĩa giao nhận
Có hai chỗ có thể làm mất hay nhân đôi message: producer gửi lại, và consumer commit offset (Exactly-once thực chất là gì).
Phía producer. Gửi lại sau khi timeout có thể ghi một message hai lần, vì lần gửi đầu có thể đã thành công. Bật idempotent producer (enable.idempotence=true) thì broker nhận ra và bỏ bản trùng do retry, và giữ đúng thứ tự trong partition.
Phía consumer. Consumer báo cho Kafka mình đã đọc tới đâu bằng cách commit offset. Lúc chết, consumer mới đọc tiếp từ offset đã commit gần nhất. Commit trước hay sau khi xử lý quyết định message bị mất hay bị xử lý lại. Trên lab, 100 message, commit mỗi 10 message, consumer bị kill -9 sau khi xử lý 35 message:
| Cách commit | Consumer mới bắt đầu từ | Kết quả |
|---|---|---|
| Commit trước khi xử lý | message 40 | Mất 5 message (35–39): at-most-once |
| Commit sau khi xử lý | message 30 | Xử lý hai lần 5 message (30–34): at-least-once |
| Tự commit mặc định (mỗi 5 giây) | message 0 | Xử lý hai lần cả 35 message, vì chưa kịp commit lần nào |
Tự commit (enable.auto.commit, mặc định bật, mỗi 5 giây) chỉ commit những message đã xử lý xong, nên vẫn là at-least-once, nhưng số message xử lý lại có thể lên tới vài giây dữ liệu. Chọn lúc chết để thấy từng trường hợp:
01234c556789c101011121314c151516171819c20
Hàng trên: việc consumer cũ làm theo thứ tự (ô số là xử lý message, ô cN là commit offset N); viền đậm là phần đã làm trước khi chết.
012345678910111213141516171819
xử lý đúng một lầnxử lý hai lầnmất, không ai xử lý
Consumer chết ngay sau bước "xử lý message 11". Offset đã commit là 10, nên consumer mới bắt đầu từ message 10. Xử lý hai lần 2 message (10–11): đã xử lý nhưng chưa kịp commit. Đây là at-least-once, nên phần xử lý phải idempotent.
Hàng dưới là kết quả sau khi consumer mới cùng group đọc tiếp từ offset đã commit. Lab chạy cùng ba cách với 100 message và cho đúng kết quả tương ứng (mất 5, trùng 5, trùng 35).
Exactly-once. Theo tài liệu Kafka, exactly-once chỉ có trọn vẹn khi đọc từ Kafka, xử lý, rồi ghi kết quả vào Kafka: producer dùng transaction để ghi kết quả và commit offset trong cùng một transaction, còn consumer phía sau đọc với isolation.level=read_committed. Kafka Streams làm sẵn việc này. Khi kết quả ghi ra ngoài Kafka (database, gọi API, gửi email), Kafka không đảm bảo được nữa, và cách thực tế là at-least-once cộng phía xử lý idempotent: lưu id message đã xử lý trong cùng transaction với kết quả, hoặc làm cho thao tác tự nhiên idempotent. Ghi database rồi gửi event ra Kafka thì dùng outbox (Transactional outbox).
Giữ message: retention và compaction
Kafka xoá message theo retention, không theo việc đã có ai đọc chưa. Mặc định giữ 7 ngày (retention.ms), có thể giới hạn thêm theo dung lượng (retention.bytes, mặc định không giới hạn). Log chia thành từng segment (mặc định 1 GB), và Kafka xoá cả segment khi mọi message trong đó đã quá hạn. Nhờ giữ lại message, một consumer mới hay một consumer vừa sửa lỗi có thể đọc lại từ đầu (lùi offset), đây là điều hàng đợi truyền thống không làm được.
Log compaction (cleanup.policy=compact) là cách giữ khác: với mỗi key, Kafka chỉ cần giữ giá trị mới nhất. Topic như vậy giống một bảng key-value có lịch sử thay đổi, hợp để lưu trạng thái mới nhất của mỗi bản ghi, hay làm đích của CDC (MySQL chương 9). Xoá một key bằng cách gửi message có value null, gọi là tombstone. Tombstone cũng bị dọn sau delete.retention.ms, nên consumer đọc chậm hơn khoảng đó có thể không thấy lệnh xoá.
Trên lab, 10 key mỗi key 100 lần cập nhật cộng một tombstone, tổng 1.001 message: sau khi cleaner chạy (17–25 giây, với cấu hình cho chạy sớm), đọc lại từ đầu chỉ còn 10 message, mỗi key một giá trị mới nhất, và key bị xoá chỉ còn tombstone. Offset của message còn lại không đổi (990, 991… 998, rồi 1000), nên consumer đang giữ offset cũ không bị lệch. Segment đang ghi không bao giờ bị dọn.
Số partition
Số partition quyết định ba thứ: số consumer tối đa đọc song song trong một group, cách tải chia cho các broker, và phạm vi giữ thứ tự (theo key, trong partition). Vì không giảm được và tăng thì làm key đổi chỗ, nên chọn từ đầu theo throughput dự kiến cho vài năm: nếu một consumer xử lý được 500 message mỗi giây và cần 10.000 message mỗi giây thì cần ít nhất 20 partition. Nhiều partition quá cũng có giá: mỗi partition là các file trên mỗi replica, và là một leader cần bầu lại khi broker chết. Topic tự tạo mặc định chỉ có 1 partition (num.partitions = 1), nên nên tắt tự tạo topic và tạo bằng tay với cấu hình rõ ràng.
Kafka hay hàng đợi truyền thống
| Cần | Kafka | Hàng đợi (RabbitMQ, SQS) |
|---|---|---|
| Nhiều hệ thống cùng đọc một luồng sự kiện | Mỗi group đọc độc lập | Phải tạo hàng đợi riêng cho mỗi bên (fan-out) |
| Đọc lại dữ liệu cũ, xử lý lại sau khi sửa lỗi | Lùi offset | Message đã xử lý xong thì mất |
| Thứ tự theo key | Trong partition | Thường chỉ gần đúng, hoặc có FIFO với giới hạn |
| Throughput rất lớn | Hàng trăm nghìn message mỗi giây mỗi broker | Thấp hơn |
| Xác nhận, retry, hẹn giờ từng message riêng lẻ | Không có: offset là một con trỏ, một message chậm chặn cả partition phía sau | Có: ack từng message, visibility timeout, dead letter queue |
| Chia việc linh hoạt cho nhiều worker | Bị giới hạn bởi số partition | Thêm worker là xong |
Kafka hợp với luồng sự kiện (event streaming): ghi lại mọi thay đổi để nhiều hệ thống dùng, CDC, log, số liệu. Hàng đợi hợp với chia việc (task queue): gửi email, tạo ảnh, mỗi việc một lần, retry riêng từng việc. Với lượng việc vừa phải, một bảng trong database làm hàng đợi là đủ (Hàng đợi bằng database).
Client Node.js
kafkajs từng là client phổ biến nhất, nhưng bản cuối cùng phát hành tháng 2/2023. Client @confluentinc/kafka-javascript của Confluent dựng trên librdkafka, còn được bảo trì, và có chế độ API giống KafkaJS để chuyển dễ. Lab dùng client này. Hai điểm lab gặp:
- Một số API admin chưa có (như tăng số partition), phải dùng công cụ dòng lệnh của Kafka.
- Chạy nhiều consumer trong cùng một process Node không song song hoàn toàn: 6 consumer chỉ nhanh hơn 1 consumer khoảng 2,5 lần, trong khi mỗi consumer một process thì đúng như mong đợi. Thực tế thường chạy mỗi consumer một process hay một pod, nên điều này ít gặp, nhưng không nên tạo nhiều consumer trong một process để tăng song song.
Ghi nhớ
- Kafka là log: message giữ theo retention, consumer tự nhớ offset, nhiều group đọc độc lập và đọc lại được.
- Thứ tự chỉ có trong partition; cùng key thì cùng partition. Tăng số partition làm khoảng một nửa key đổi chỗ, không giảm được, nên chọn dư từ đầu; client khác nhau có thể băm key khác nhau.
- Mỗi partition chỉ một consumer trong group: số consumer song song không vượt số partition. Consumer chết đột ngột làm partition của nó dừng tới hết session timeout; theo dõi lag.
- Không mất message đã xác nhận cần cả ba: replication factor 3,
acks=all,min.insync.replicas = 2(mặc định là 1). Follower chết đột ngột làmacks=allkẹt tới khi controller phát hiện (khoảng 9 giây). - Commit trước khi xử lý thì có thể mất, sau khi xử lý thì có thể xử lý lại; tự commit là at-least-once với vài giây dữ liệu. Exactly-once chỉ trọn vẹn trong Kafka (transaction); ghi ra ngoài thì cần phía xử lý idempotent.
- Gom lô quyết định throughput:
awaittừngsendchỉ được hơn 100 message mỗi giây.
Nguồn tham khảo
Đã đối chiếu (đọc nguồn ngày 2026-10-28):
| Nguồn | Dùng cho phần |
|---|---|
| Apache Kafka 4.1, Design | sendfile và page cache, không dùng khi bật TLS; message commit khi mọi replica trong ISR đã có; min.insync.replicas; unclean leader election; at-most-once, at-least-once, idempotent producer, exactly-once bằng transaction và read_committed, ghi ra ngoài Kafka cần hệ thống đích phối hợp; log compaction giữ giá trị mới nhất mỗi key, tombstone, delete.retention.ms; broker.session.timeout.ms và replica.lag.time.max.ms |
| Mã nguồn Kafka 4.1: ReplicationConfigs, KRaftConfigs, ServerLogConfigs, LogConfig | Giá trị mặc định: replica.lag.time.max.ms 30 s, broker.session.timeout.ms 9 s, min.insync.replicas 1, num.partitions 1, retention 7 ngày, segment 1 GB, unclean leader election tắt |
| librdkafka, CONFIGURATION.md | partitioner mặc định consistent_random (CRC32), murmur2 tương thích Java; acks mặc định -1; retry có thể đảo thứ tự nếu không bật idempotence; message.timeout.ms 300 s; linger.ms 5 ms; session.timeout.ms 45 s; max.poll.interval.ms 300 s; auto.commit.interval.ms 5 s; group.protocol classic hoặc consumer (KIP-848) |
| Confluent, confluent-kafka-javascript MIGRATION.md | Chế độ KafkaJS mặc định partitioner murmur2_random, assigner roundRobin |
npm, kafkajs | Bản cuối 2.2.4, cập nhật lần cuối tháng 2/2023 |
Kiểm bằng lab: thứ tự theo key, key đổi partition khi tăng số partition, consumer group với 1–8 consumer, rebalance, consumer bị kill, hai group, latency theo acks, throughput theo cách gửi, follower tắt và bị kill, NOT_ENOUGH_REPLICAS, mất đa số controller khi kill 2/3 node, ba cách commit offset, log compaction. Từ kiến thức chung, chưa đối chiếu lần này: cách chia cooperative-sticky; throughput hàng trăm nghìn message mỗi giây mỗi broker; chi phí của quá nhiều partition; so sánh với RabbitMQ và SQS (trang chính là Hàng đợi message); Kafka không cho giảm số partition.
Số đo chi tiết
Kết quả chạy lab/kafka (Kafka 4.1.0 trong Docker, 3 broker KRaft, client @confluentinc/kafka-javascript 1.11 với librdkafka 2.16, Node 24.14, hai đến bốn lượt)
Thứ tự và partition: 4 đơn × 4 sự kiện, key là mã đơn, topic 3 partition: một partition chứa đơn D, một partition chứa A, B, C xen kẽ, một partition rỗng; thứ tự trong từng đơn đúng, thứ tự chung sai. Tăng 3 → 6 partition: 534/1.000 key đổi partition (hai lượt giống nhau, vì băm tất định).
Consumer group (mỗi consumer một process, 6 partition, 600 message, 10 ms mỗi message, đo từ lúc gửi sau khi group ổn định; ba lượt):
| Consumer | Đọc hết sau |
|---|---|
| 1 | 6,6 s |
| 3 | 2,4–3,9 s |
| 6 | 1,6–3,0 s |
| 8 | 1,3–1,6 s (2 consumer không có partition) |
Thời gian dao động vì các partition bắt đầu được đọc lệch nhau tới khoảng 1 giây sau khi message đến. Thêm consumer thứ 3 khi đang đọc: khoảng không xử lý dài nhất 11–13 ms (ba lượt với mỗi consumer một process), 45–101 ms (hai lượt chạy consumer chung một process). Kill -9 một trong ba consumer, session timeout 10 s, gửi đều 50 message/s: message vào partition của nó chờ tối đa 10,3–10,6 s, lúc bình thường trung vị 11–13 ms. Hai group cùng đọc: mỗi group nhận 600/600.
Lượt đầu chạy 6 consumer trong cùng một process: chỉ nhanh hơn 1 consumer khoảng 2,5 lần; tăng UV_THREADPOOL_SIZE lên 16 không đổi gì, nên giới hạn không nằm ở thread pool của libuv. Nguyên nhân thật chưa tìm ra.
Replication (topic 1 partition, replication factor 3, min.insync.replicas = 2; bốn lượt):
| Phép thử | Kết quả |
|---|---|
Latency mỗi lần gửi, acks=1 / acks=all | p50 6,4–7,1 ms / 6,6–7,7 ms; p99 8,3–10,8 ms / 10,4–12,4 ms |
Gửi 200 byte, acks=all: chờ từng message / 1.000 message mỗi lần / 20.000 cùng lúc | 131–141 / 75.014–91.390 / 104.204–173.568 message/s |
Follower docker stop | ISR còn 2 sau 1,06–1,19 s; lần gửi chậm nhất 11,8–58,6 ms; không lỗi |
Follower docker kill | ISR còn 2 sau 8,2–10,3 s; lần gửi chậm nhất 6,9–9,1 s; không lỗi |
min.insync.replicas = 3, một follower tắt | acks=all: timeout sau 5 s (message.timeout.ms của lab); acks=all, retries 0: lỗi 19 NOT_ENOUGH_REPLICAS sau 7,5–8,6 ms; acks=1: thành công sau 6,4–8,0 ms |
Lượt đầu đo thời gian ISR thu hẹp bằng kafka-topics.sh với danh sách bootstrap có cả broker vừa tắt, nên ra 3–11 s cho cả trường hợp tắt có thứ tự, là thời gian công cụ chờ kết nối; script sau đó chỉ hỏi qua broker còn sống. Kill 2 trong 3 node: ISR không thu hẹp trong 90 s vì cụm mất đa số controller.
Commit offset (100 message, commit mỗi 10, kill sau 35; hai lượt giống nhau): commit trước: bắt đầu lại từ 40, mất 35–39; commit sau: bắt đầu lại từ 30, xử lý hai lần 30–34; tự commit: bắt đầu lại từ 0, xử lý hai lần 0–34.
Log compaction (segment.ms 1 s, min.cleanable.dirty.ratio 0,01, delete.retention.ms 1 s; hai lượt): 1.001 message còn 10 sau 17–25 s; offset 990–998 và 1000; key bị xoá chỉ còn tombstone.
Code đã dùng để chạy thử
Cụm Kafka riêng: npm run lab -- kafka:up, chạy bằng npm run lab -- kafka (khoảng 7–8 phút), xoá bằng npm run lab -- kafka:down. Không cần lab:up. Script replication.mjs tắt và bật lại broker, nên không chạy song song với script khác. Xem Lab.
lab/kafka/docker-compose.yml
yml
# Kafka 4.1 cho trang messaging/kafka: 3 broker chạy KRaft (mỗi node vừa là broker vừa là controller, không cần ZooKeeper).
# docker compose -f kafka/docker-compose.yml up -d --wait (npm run kafka:up)
# docker compose -f kafka/docker-compose.yml down -v (npm run kafka:down)
# Client trên máy host nối qua 127.0.0.1:39092, 39093, 39094; các broker nói chuyện với nhau trong mạng lab-kafka.
# Heap mỗi broker giới hạn 512 MB để cả cụm vừa laptop. Log giữ tối đa 1 giờ, segment 16 MB để không đầy đĩa.
x-kafka: &kafka
image: apache/kafka:4.1.0
networks: [kafka]
healthcheck:
test: ['CMD-SHELL', '/opt/kafka/bin/kafka-broker-api-versions.sh --bootstrap-server localhost:19092 > /dev/null 2>&1']
interval: 3s
timeout: 10s
retries: 40
x-env: &env
CLUSTER_ID: 'lab-kafka-cluster-0001'
KAFKA_PROCESS_ROLES: 'broker,controller'
KAFKA_CONTROLLER_QUORUM_VOTERS: '1@lab-kafka1:9093,2@lab-kafka2:9093,3@lab-kafka3:9093'
KAFKA_CONTROLLER_LISTENER_NAMES: 'CONTROLLER'
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: 'CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,HOST:PLAINTEXT'
KAFKA_INTER_BROKER_LISTENER_NAME: 'PLAINTEXT'
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3
KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 3
KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 2
KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'false'
KAFKA_LOG_RETENTION_MS: 3600000
KAFKA_LOG_SEGMENT_BYTES: 16777216
KAFKA_HEAP_OPTS: '-Xmx512m -Xms256m'
services:
kafka1:
<<: *kafka
container_name: lab-kafka1
ports: ['127.0.0.1:39092:9092']
environment:
<<: *env
KAFKA_NODE_ID: 1
KAFKA_LISTENERS: 'PLAINTEXT://:19092,CONTROLLER://:9093,HOST://:9092'
KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT://lab-kafka1:19092,HOST://127.0.0.1:39092'
kafka2:
<<: *kafka
container_name: lab-kafka2
ports: ['127.0.0.1:39093:9092']
environment:
<<: *env
KAFKA_NODE_ID: 2
KAFKA_LISTENERS: 'PLAINTEXT://:19092,CONTROLLER://:9093,HOST://:9092'
KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT://lab-kafka2:19092,HOST://127.0.0.1:39093'
kafka3:
<<: *kafka
container_name: lab-kafka3
ports: ['127.0.0.1:39094:9092']
environment:
<<: *env
KAFKA_NODE_ID: 3
KAFKA_LISTENERS: 'PLAINTEXT://:19092,CONTROLLER://:9093,HOST://:9092'
KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT://lab-kafka3:19092,HOST://127.0.0.1:39094'
networks:
kafka:
name: lab-kafkalab/kafka/common.mjs
js
// Hàm dùng chung cho lab/kafka. Client là @confluentinc/kafka-javascript (dựng trên librdkafka), dùng API giống KafkaJS.
import { KafkaJS } from '@confluentinc/kafka-javascript'
import { execFileSync } from 'node:child_process'
export const BROKERS = ['127.0.0.1:39092', '127.0.0.1:39093', '127.0.0.1:39094']
export const CONTAINERS = ['lab-kafka1', 'lab-kafka2', 'lab-kafka3']
export const sleep = (ms) => new Promise((r) => setTimeout(r, ms))
export const now = () => performance.now()
// logLevel NOTHING: librdkafka in rất nhiều cảnh báo khi cố ý tắt broker; lỗi thật vẫn ném ra ở lời gọi
export const kafka = (extra = {}) => new KafkaJS.Kafka({ kafkaJS: { brokers: BROKERS, logLevel: KafkaJS.logLevel.NOTHING, ...extra } })
export async function recreateTopic(admin, topic, { partitions = 3, replicationFactor = 3, configs = {} } = {}) {
const existing = await admin.listTopics()
if (existing.includes(topic)) {
await admin.deleteTopics({ topics: [topic] })
for (let i = 0; i < 50 && (await admin.listTopics()).includes(topic); i++) await sleep(200)
await sleep(500)
}
await admin.createTopics({
topics: [{ topic, numPartitions: partitions, replicationFactor, configEntries: Object.entries(configs).map(([name, value]) => ({ name, value: String(value) })) }]
})
await sleep(1000)
}
// Chạy lệnh trong container của một broker (công cụ dòng lệnh đi kèm Kafka)
export const kafkaCli = (container, tool, args) =>
execFileSync('docker', ['exec', container, `/opt/kafka/bin/${tool}`, ...args], { encoding: 'utf8', stdio: ['ignore', 'pipe', 'pipe'] })
export const dockerCmd = (...args) => execFileSync('docker', args, { encoding: 'utf8', stdio: ['ignore', 'pipe', 'pipe'] })
export function pct(sorted, p) { return sorted[Math.min(sorted.length - 1, Math.ceil((p / 100) * sorted.length) - 1)] }
export const ms = (v) => (v >= 100 ? Math.round(v).toLocaleString('vi-VN') : v.toFixed(1).replace('.', ',')) + ' ms'lab/kafka/order.mjs: key, partition, thứ tự
js
// Key, partition và thứ tự.
// 1. Message cùng key luôn vào cùng partition, nên giữ đúng thứ tự theo key; giữa các partition thì không có thứ tự chung
// 2. Thêm partition cho topic thì một phần key chuyển sang partition khác
import { kafka, recreateTopic, sleep, kafkaCli } from './common.mjs'
const k = kafka()
const admin = k.admin(); await admin.connect()
const producer = k.producer(); await producer.connect()
const TOPIC = 'lab-order'
await recreateTopic(admin, TOPIC, { partitions: 3 })
// 4 đơn hàng, mỗi đơn 4 sự kiện theo đúng thứ tự, gửi xen kẽ nhau, key là mã đơn
const STEPS = ['tạo', 'thanh toán', 'giao', 'hoàn tất']
const orders = ['A', 'B', 'C', 'D']
let seq = 0
for (const step of STEPS) for (const o of orders) await producer.send({ topic: TOPIC, messages: [{ key: o, value: JSON.stringify({ order: o, step, seq: seq++ }) }] })
const consumer = k.consumer({ kafkaJS: { groupId: 'lab-order-g', fromBeginning: true } })
await consumer.connect(); await consumer.subscribe({ topic: TOPIC })
const got = []
consumer.run({ eachMessage: async ({ partition, message }) => { got.push({ partition, ...JSON.parse(message.value) }) } })
for (let i = 0; i < 80 && got.length < seq; i++) await sleep(250)
await consumer.disconnect()
console.log('\n=== 1. 4 đơn × 4 sự kiện, key là mã đơn, topic 3 partition ===')
const byPart = {}
for (const g of got) (byPart[g.partition] ??= []).push(g)
for (const p of Object.keys(byPart).sort()) console.log(`partition ${p}: ${byPart[p].map((g) => `${g.order}:${g.step}`).join(', ')}`)
const okPerKey = orders.every((o) => { const s = got.filter((g) => g.order === o).map((g) => g.seq); return s.every((v, i) => i === 0 || v > s[i - 1]) })
const globalInOrder = got.every((g, i) => i === 0 || g.seq > got[i - 1].seq)
console.log(`Thứ tự trong từng đơn đúng: ${okPerKey}; thứ tự đọc ra khớp thứ tự gửi trên toàn topic: ${globalInOrder}`)
console.log('\n=== 2. Tăng từ 3 lên 6 partition: key nào đổi partition ===')
const keys = Array.from({ length: 1000 }, (_, i) => `user-${i}`)
// Gửi từng key một, kết quả của send cho biết message vào partition nào
const partOf = async (p) => {
const m = new Map()
for (const key of keys) { const [r] = await p.send({ topic: TOPIC, messages: [{ key, value: 'x' }] }); m.set(key, r.partition) }
return m
}
const before = await partOf(producer)
// client của Confluent chưa có createPartitions trong admin API, nên dùng công cụ dòng lệnh của Kafka
kafkaCli('lab-kafka1', 'kafka-topics.sh', ['--bootstrap-server', 'localhost:19092', '--alter', '--topic', TOPIC, '--partitions', '6'])
await sleep(3000)
// producer mới để chắc chắn có metadata 6 partition
await producer.disconnect()
const p2 = k.producer(); await p2.connect()
const after = await partOf(p2)
const movedKeys = keys.filter((key) => before.get(key) !== after.get(key))
const moved = movedKeys.length
console.log(`${moved}/${keys.length} key (${Math.round((moved / keys.length) * 100)}%) chuyển sang partition khác; ví dụ ${movedKeys.slice(0, 2).map((key) => `${key}: partition ${before.get(key)} → ${after.get(key)}`).join(', ')}`)
await p2.disconnect(); await admin.deleteTopics({ topics: [TOPIC] }); await admin.disconnect()lab/kafka/group.mjs và consumer-child.mjs: consumer group
js
// Consumer group. Mỗi consumer chạy trong một process riêng (consumer-child.mjs).
// 1. Topic 6 partition, mỗi message xử lý mất 10 ms: thời gian đọc hết 600 message với 1, 3, 6, 8 consumer trong một group
// (đo từ lúc gửi, sau khi group đã ổn định); consumer thứ 7, 8 không được giao partition nào
// 2. Đang đọc thì thêm một consumer: cả group ngừng xử lý trong bao lâu
// 3. Một consumer bị kill -9 (không kịp rời group): partition của nó dừng tới khi hết session timeout
// 4. Hai group khác nhau cùng đọc một topic: mỗi group nhận đủ mọi message
import { fork } from 'node:child_process'
import { kafka, recreateTopic, sleep, now } from './common.mjs'
const k = kafka()
const admin = k.admin(); await admin.connect()
const producer = k.producer(); await producer.connect()
const TOPIC = 'lab-group'
const N = 600
const sec = (v) => (v / 1000).toFixed(1).replace('.', ',') + ' s'
const children = new Set()
// Một group gồm `count` consumer, mỗi consumer một process; theo dõi partition mỗi consumer đang giữ
function startGroup(count, group, { session = 10000, work = 10, onMsg = () => {} } = {}) {
const g = { held: new Map(), lastChange: now(), kids: [], times: [] }
g.add = () => {
const child = fork(new URL('./consumer-child.mjs', import.meta.url), [TOPIC, group, String(session), String(work)])
const id = g.kids.length
children.add(child)
child.on('message', (m) => {
if (m.held) { g.held.set(id, m.held); g.lastChange = now() }
else if (m.msg) { g.times.push(now()); onMsg(m.msg) }
})
child.on('exit', () => children.delete(child))
g.kids.push(child)
return child
}
for (let i = 0; i < count; i++) g.add()
g.heldCount = () => [...g.held.values()].reduce((a, b) => a + b.length, 0)
g.stop = () => Promise.all(g.kids.filter((c) => c.exitCode === null && !c.killed).map((c) => new Promise((r) => { c.once('exit', r); c.send('stop') })))
return g
}
// Group ổn định: đủ 6 partition đã được giao và không có rebalance nào trong 2 giây
async function waitStable(g, total = 6) {
for (let i = 0; i < 400; i++) { if (g.heldCount() === total && now() - g.lastChange > 2000) return; await sleep(100) }
throw new Error('group không ổn định sau 40 giây')
}
const send = async (n) => { for (let i = 0; i < n; i += 100) await producer.send({ topic: TOPIC, messages: Array.from({ length: Math.min(100, n - i) }, (_, j) => ({ key: `k${i + j}`, value: JSON.stringify({ i: i + j, ts: Date.now() }) })) }) }
try {
console.log(`\n=== 1. ${N} message trên 6 partition, mỗi message xử lý 10 ms, đo từ lúc gửi sau khi group ổn định ===`)
for (const count of [1, 3, 6, 8]) {
await recreateTopic(admin, TOPIC, { partitions: 6 })
const g = startGroup(count, `lab-group-${count}-${Date.now()}`)
await waitStable(g)
const t0 = now()
await send(N)
while (g.times.length < N && now() - t0 < 60000) await sleep(10)
const held = [...g.held.values()].map((p) => p.length).concat(Array(count - g.held.size).fill(0)).sort((a, b) => b - a)
const idle = held.filter((x) => x === 0).length
console.log(`${count} consumer: đọc hết sau ${sec(now() - t0)}; số partition mỗi consumer giữ: ${held.join(', ')}${idle ? `; ${idle} consumer không có partition nào` : ''}`)
await g.stop()
}
console.log('\n=== 2. Đang đọc bằng 2 consumer thì thêm consumer thứ 3 ===')
{
await recreateTopic(admin, TOPIC, { partitions: 6 })
const g = startGroup(2, `lab-group-reb-${Date.now()}`)
await waitStable(g)
await send(N)
while (g.times.length < 150) await sleep(5)
const tAdd = now()
g.add()
while (g.times.length < N && now() - tAdd < 60000) await sleep(10)
const t = [...g.times].sort((a, b) => a - b)
let gap = 0, before = 0
for (let i = 1; i < t.length; i++) { const d = t[i] - t[i - 1]; if (t[i] > tAdd) gap = Math.max(gap, d); else if (i > 5) before = Math.max(before, d) }
console.log(`Khoảng không xử lý message nào dài nhất sau khi thêm consumer: ${Math.round(gap)} ms; trước đó dài nhất ${Math.round(before)} ms`)
await g.stop()
}
console.log('\n=== 3. 3 consumer (session timeout 10 s), một consumer bị kill -9; gửi đều 50 message/s ===')
{
await recreateTopic(admin, TOPIC, { partitions: 6 })
const delays = []
const g = startGroup(3, `lab-group-kill-${Date.now()}`, { work: 0, onMsg: (m) => delays.push({ sentAt: m.ts, delay: Date.now() - m.ts }) })
await waitStable(g)
let stop = false
const pump = (async () => { let i = 0; while (!stop) { await producer.send({ topic: TOPIC, messages: [{ key: `p${i++}`, value: JSON.stringify({ ts: Date.now() }) }] }); await sleep(20) } })()
await sleep(3000)
const victim = g.kids[0]
const victimParts = g.held.get(0)
const killedAt = Date.now()
victim.kill('SIGKILL')
await sleep(16000)
stop = true; await pump
await sleep(1000)
const worst = Math.max(...delays.filter((d) => d.sentAt >= killedAt - 500).map((d) => d.delay))
const normal = delays.filter((d) => d.sentAt < killedAt - 500).map((d) => d.delay).sort((a, b) => a - b)
console.log(`Consumer bị kill giữ partition ${victimParts.join(', ')}. Message gửi vào các partition đó phải chờ tối đa ${sec(worst)} mới được đọc (lúc bình thường trễ trung vị ${normal[Math.floor(normal.length / 2)]} ms)`)
await g.stop()
}
console.log('\n=== 4. Hai group khác nhau cùng đọc topic ===')
{
await recreateTopic(admin, TOPIC, { partitions: 6 })
await send(N)
const groups = [0, 1].map((x) => startGroup(3, `lab-group-${x ? 'y' : 'x'}-${Date.now()}`, { work: 0 }))
for (let i = 0; i < 400 && groups.some((g) => g.times.length < N); i++) await sleep(100)
console.log(`Group x nhận ${groups[0].times.length}/${N} message, group y nhận ${groups[1].times.length}/${N}: mỗi group đọc toàn bộ topic, độc lập với nhau`)
for (const g of groups) await g.stop()
}
} finally {
for (const c of children) c.kill('SIGKILL')
await producer.disconnect(); await admin.deleteTopics({ topics: [TOPIC] }).catch(() => {}); await admin.disconnect()
}js
// Một consumer chạy trong process riêng, cho lab/kafka/group.mjs. Mỗi consumer một process giống cách chạy thật (mỗi pod
// một consumer), và để có thể kill -9 nó (chết đột ngột, không kịp rời group). Nhiều consumer trong cùng một process
// Node với client này không chạy song song hoàn toàn (6 consumer chỉ nhanh hơn 1 consumer khoảng 2,5 lần).
// Đối số: topic, group, session timeout (ms), thời gian xử lý mỗi message (ms).
// Gửi cho process cha: { held } mỗi khi được giao hoặc bị thu partition, { msg: { ts } } mỗi message xử lý xong.
import { kafka, sleep } from './common.mjs'
const [topic, group, sessionTimeout, workMs] = process.argv.slice(2)
const c = kafka().consumer({
kafkaJS: { groupId: group, fromBeginning: true, sessionTimeout: Number(sessionTimeout) },
rebalance_cb: (err, parts) => {
if (err.code === -175) process.send({ held: parts.map((p) => p.partition) }) // ERR__ASSIGN_PARTITIONS
else if (err.code === -174) process.send({ held: [] }) // ERR__REVOKE_PARTITIONS
}
})
await c.connect()
await c.subscribe({ topic })
c.run({ eachMessage: async ({ message }) => { if (Number(workMs)) await sleep(Number(workMs)); process.send({ msg: JSON.parse(message.value) }) } })
process.on('message', async (m) => { if (m === 'stop') { await c.disconnect(); process.exit(0) } })lab/kafka/replication.mjs: acks, ISR, min.insync.replicas
js
// Replication, acks và min.insync.replicas. Topic 1 partition, replication factor 3, min.insync.replicas = 2.
// 1. Latency mỗi lần gửi (chờ xác nhận rồi mới gửi tiếp) với acks=1 và acks=all
// 2. Gửi 20.000 message: chờ từng cái, gửi cùng lúc, gửi theo lô
// 3. Một follower tắt có thứ tự (docker stop) so với bị kill (docker kill): acks=all bị kẹt bao lâu
// 4. ISR ít hơn min.insync.replicas: acks=all bị từ chối, acks=1 vẫn ghi được
// Script tắt và bật lại broker; khối finally luôn bật lại mọi broker.
import { kafka, recreateTopic, sleep, now, kafkaCli, dockerCmd, pct, ms } from './common.mjs'
const TOPIC = 'lab-repl'
const k = kafka()
const admin = k.admin(); await admin.connect()
await recreateTopic(admin, TOPIC, { partitions: 1, configs: { 'min.insync.replicas': 2 } })
const stopped = new Set()
const container = (id) => `lab-kafka${id}`
// Luôn hỏi qua một broker đang sống: nếu danh sách bootstrap có broker vừa tắt, công cụ có thể mất vài giây chờ kết nối,
// và thời gian đo được thành thời gian chờ của công cụ chứ không phải thời gian ISR thu hẹp
const describe = () => {
const alive = [1, 2, 3].find((id) => !stopped.has(id))
const out = kafkaCli(container(alive), 'kafka-topics.sh', ['--bootstrap-server', 'localhost:19092', '--describe', '--topic', TOPIC])
const m = out.match(/Leader: (\d+)\s+Replicas: ([\d,]+)\s+Isr: ([\d,]*)/)
return { leader: Number(m[1]), replicas: m[2].split(',').map(Number), isr: m[3] ? m[3].split(',').map(Number) : [] }
}
const waitIsr = async (n, timeout = 90000) => {
const t = now()
while (now() - t < timeout) { try { if (describe().isr.length === n) return now() - t } catch {} await sleep(500) }
throw new Error(`ISR không về ${n} replica sau ${timeout} ms`)
}
// Producer với acks cho trước; message.timeout.ms ngắn để lỗi hiện ra nhanh thay vì retry 2 phút (mặc định)
const makeProducer = async (acks, extra = {}) => { const p = k.producer({ kafkaJS: { acks }, 'message.timeout.ms': 5000, ...extra }); await p.connect(); return p }
try {
console.log('\n=== 1. Latency mỗi lần gửi, chờ xác nhận rồi mới gửi tiếp, 500 lần ===')
for (const acks of [1, -1]) {
const p = await makeProducer(acks)
for (let i = 0; i < 50; i++) await p.send({ topic: TOPIC, messages: [{ value: 'warm' }] })
const lat = []
for (let i = 0; i < 500; i++) { const t = now(); await p.send({ topic: TOPIC, messages: [{ value: 'x'.repeat(200) }] }); lat.push(now() - t) }
lat.sort((a, b) => a - b)
console.log(`acks=${acks === -1 ? 'all' : acks}: p50 ${ms(pct(lat, 50))}, p99 ${ms(pct(lat, 99))}`)
await p.disconnect()
}
console.log('\n=== 2. Gửi 20.000 message 200 byte, acks=all ===')
{
const p = await makeProducer(-1)
const M = 20000, msg = { value: 'x'.repeat(200) }
let t = now()
for (let i = 0; i < 2000; i++) await p.send({ topic: TOPIC, messages: [msg] })
const seqRate = 2000 / ((now() - t) / 1000)
t = now()
await Promise.all(Array.from({ length: M }, () => p.send({ topic: TOPIC, messages: [msg] })))
const concRate = M / ((now() - t) / 1000)
t = now()
for (let i = 0; i < M; i += 1000) await p.send({ topic: TOPIC, messages: Array(1000).fill(msg) })
const batchRate = M / ((now() - t) / 1000)
const f = (v) => Math.round(v).toLocaleString('vi-VN')
console.log(`Chờ từng message: ${f(seqRate)} message/s (đo trên 2.000); gửi cả 20.000 cùng lúc: ${f(concRate)} message/s; mỗi lần gửi 1.000 message: ${f(batchRate)} message/s`)
await p.disconnect()
}
const { leader, replicas } = describe()
const [f1, f2] = replicas.filter((r) => r !== leader)
console.log(`\nLeader của partition: broker ${leader}; follower: ${f1}, ${f2}`)
// Gửi acks=all đều mỗi 100 ms trong `seconds` giây, ghi latency hoặc lỗi của từng lần
const probe = async (p, seconds) => {
const res = []
const end = now() + seconds * 1000
const inflight = []
while (now() < end) {
const t = now()
inflight.push(p.send({ topic: TOPIC, messages: [{ value: 'probe' }] }).then(() => res.push({ at: t, ms: now() - t }), (e) => res.push({ at: t, err: e.code ?? e.message })))
await sleep(100)
}
await Promise.all(inflight)
return res
}
for (const [label, cmd] of [['docker stop (tắt có thứ tự)', 'stop'], ['docker kill (chết đột ngột)', 'kill']]) {
console.log(`\n=== 3. Follower ${f1}: ${label}; gửi acks=all mỗi 100 ms trong 15 giây ===`)
const p = await makeProducer(-1, { 'message.timeout.ms': 30000 })
await p.send({ topic: TOPIC, messages: [{ value: 'warm' }] })
const run = probe(p, 15)
await sleep(1000)
const tDown = now()
dockerCmd(cmd, container(f1)); stopped.add(f1)
const isrAfter = await waitIsr(2).catch(() => null)
const res = await run
const after = res.filter((r) => r.at >= tDown - 200)
const worst = Math.max(...after.filter((r) => !r.err).map((r) => r.ms))
const errs = after.filter((r) => r.err).length
console.log(`ISR còn 2 replica sau ${isrAfter === null ? 'hơn 90 s' : ms(isrAfter)}; lần gửi chậm nhất sau khi follower chết: ${ms(worst)}; ${errs} lần lỗi`)
await p.disconnect()
dockerCmd('start', container(f1)); stopped.delete(f1)
const back = await waitIsr(3)
console.log(`Bật lại broker ${f1}: ISR đủ 3 replica sau ${ms(back)}`)
}
// Lab chạy mỗi node vừa là broker vừa là controller (KRaft combined), nên tắt 2/3 node là mất đa số của controller:
// không còn ai thu hẹp ISR, và acks=all chờ mãi. Vì vậy phần này tăng min.insync.replicas lên 3 rồi chỉ tắt một follower.
console.log(`\n=== 4. min.insync.replicas = 3, tắt follower ${f1}: ISR còn 2 replica, ít hơn mức tối thiểu ===`)
kafkaCli('lab-kafka1', 'kafka-configs.sh', ['--bootstrap-server', 'localhost:19092', '--alter', '--entity-type', 'topics', '--entity-name', TOPIC, '--add-config', 'min.insync.replicas=3'])
dockerCmd('stop', container(f1)); stopped.add(f1)
await waitIsr(2)
// librdkafka tự retry lỗi NOT_ENOUGH_REPLICAS tới hết message.timeout.ms; retries 0 để thấy lỗi gốc từ broker
for (const [label, acks, extra] of [['acks=all', -1, {}], ['acks=all, retries 0', -1, { retries: 0 }], ['acks=1', 1, {}]]) {
const p = await makeProducer(acks, extra)
const t = now()
const r = await p.send({ topic: TOPIC, messages: [{ value: 'khi ISR thiếu' }] }).then(() => 'ghi thành công', (e) => `lỗi ${e.code ?? ''} ${e.message}`)
console.log(`${label}: ${r} sau ${ms(now() - t)}`)
await p.disconnect()
}
} finally {
for (const id of stopped) { try { dockerCmd('start', container(id)) } catch {} }
if (stopped.size) { const t = await waitIsr(3).catch(() => null); console.log(`\nĐã bật lại broker ${[...stopped].join(', ')}; ISR đủ 3 replica sau ${t === null ? 'chưa đủ' : ms(t)}`) }
await admin.deleteTopics({ topics: [TOPIC] }).catch(() => {})
await admin.disconnect()
}lab/kafka/delivery.mjs và delivery-child.mjs: commit offset
js
// Commit offset trước hay sau khi xử lý, khi consumer chết giữa chừng.
// Topic 1 partition, 100 message mang số 0..99. Consumer chạy trong process con (delivery-child.mjs), xử lý mỗi message
// 20 ms, commit offset sau mỗi 10 message theo một trong ba cách; process cha kill -9 nó khi đã xử lý xong 35 message.
// Rồi một consumer mới cùng group đọc tiếp tới hết. Đếm message bị mất (không ai xử lý) và bị xử lý hai lần.
import { fork } from 'node:child_process'
import { kafka, recreateTopic, sleep } from './common.mjs'
const TOPIC = 'lab-delivery'
const k = kafka()
const admin = k.admin(); await admin.connect()
const producer = k.producer(); await producer.connect()
const N = 100
async function runChild(group, mode, killAfter) {
const seen = []
const child = fork(new URL('./delivery-child.mjs', import.meta.url), [TOPIC, group, mode])
await new Promise((resolve) => {
let done = false
child.on('message', (m) => {
seen.push(m.i)
if (killAfter !== null && seen.length >= killAfter && !done) { done = true; child.kill('SIGKILL') }
})
child.on('exit', resolve)
if (killAfter === null) {
// consumer thứ hai: phải chờ consumer cũ hết session timeout (6 s) mới được giao partition;
// dừng khi đã nhận message rồi không có message mới trong 4 giây, hoặc sau 60 giây
const start = Date.now()
let last = 0, lastLen = 0
const iv = setInterval(() => {
if (seen.length !== lastLen) { lastLen = seen.length; last = Date.now() }
else if ((seen.length && Date.now() - last > 4000) || Date.now() - start > 60000) { clearInterval(iv); child.send('stop') }
}, 200)
}
})
return seen
}
console.log(`\n=== ${N} message, xử lý mỗi message 20 ms, commit mỗi 10 message; consumer bị kill -9 sau khi xử lý 35 message ===`)
try {
for (const [mode, label] of [['before', 'Commit trước khi xử lý (at-most-once)'], ['after', 'Commit sau khi xử lý (at-least-once)'], ['auto', 'Tự commit mặc định (mỗi 5 giây)']]) {
await recreateTopic(admin, TOPIC, { partitions: 1 })
await producer.send({ topic: TOPIC, messages: Array.from({ length: N }, (_, i) => ({ value: String(i) })) })
const group = `lab-delivery-${mode}-${Date.now()}`
const first = await runChild(group, mode, 35)
await sleep(500)
const second = await runChild(group, mode, null)
const count = new Map()
for (const i of [...first, ...second]) count.set(i, (count.get(i) ?? 0) + 1)
const lost = Array.from({ length: N }, (_, i) => i).filter((i) => !count.has(i))
const dup = [...count].filter(([, c]) => c > 1).map(([i]) => i).sort((a, b) => a - b)
const span = (a) => (a.length ? `${a.length} (số ${a[0]}–${a.at(-1)})` : '0')
console.log(`${label.padEnd(40)} consumer mới bắt đầu từ message số ${second[0] ?? 'không còn gì'}; mất ${span(lost)}; xử lý hai lần ${span(dup)}`)
}
} finally {
await producer.disconnect(); await admin.deleteTopics({ topics: [TOPIC] }).catch(() => {}); await admin.disconnect()
}js
// Consumer cho lab/kafka/delivery.mjs, chạy trong process riêng để bị kill -9 giữa chừng.
// Đối số: topic, group, cách commit:
// before đầu mỗi lô 10 message, commit offset cuối lô rồi mới xử lý (xử lý "đã nhận" trước khi làm)
// after xử lý xong 10 message rồi mới commit
// auto để client tự commit theo chu kỳ (mặc định 5 giây), như cấu hình mặc định
// Mỗi message xử lý xong thì báo số thứ tự cho process cha.
import { kafka, sleep } from './common.mjs'
const [topic, group, mode] = process.argv.slice(2)
// session timeout 6 s: consumer bị kill vẫn là thành viên của group tới khi hết khoảng này, consumer mới phải chờ
const c = kafka().consumer({ kafkaJS: { groupId: group, fromBeginning: true, autoCommit: mode === 'auto', sessionTimeout: 6000 } })
await c.connect()
await c.subscribe({ topic })
const commit = (partition, offset) => c.commitOffsets([{ topic, partition, offset: String(offset) }])
c.run({
eachMessage: async ({ partition, message }) => {
const off = Number(message.offset)
if (mode === 'before' && off % 10 === 0) await commit(partition, off + 10)
await sleep(20)
process.send({ i: Number(message.value) })
if (mode === 'after' && off % 10 === 9) await commit(partition, off + 1)
}
})
process.on('message', async (m) => { if (m === 'stop') { await c.disconnect(); process.exit(0) } })lab/kafka/compaction.mjs
js
// Log compaction: topic chỉ giữ message mới nhất của mỗi key.
// 10 key, mỗi key 100 lần cập nhật (1.000 message), rồi một tombstone (value null) cho key k9. Segment đóng sau 1 giây
// và tỉ lệ "bẩn" tối thiểu rất nhỏ để cleaner chạy sớm; cleaner của broker kiểm tra mỗi 15 giây (log.cleaner.backoff.ms).
// Segment đang ghi (active segment) không bao giờ bị dọn, nên cuối cùng gửi thêm một message để đóng segment cũ.
import { kafka, recreateTopic, sleep, now } from './common.mjs'
const TOPIC = 'lab-compact'
const k = kafka()
const admin = k.admin(); await admin.connect()
const producer = k.producer(); await producer.connect()
await recreateTopic(admin, TOPIC, {
partitions: 1,
configs: { 'cleanup.policy': 'compact', 'segment.ms': 1000, 'min.cleanable.dirty.ratio': 0.01, 'delete.retention.ms': 1000 }
})
async function readAll() {
const got = []
const c = k.consumer({ kafkaJS: { groupId: `lab-compact-${Date.now()}`, fromBeginning: true } })
await c.connect(); await c.subscribe({ topic: TOPIC })
let last = now()
c.run({ eachMessage: async ({ message }) => { got.push({ key: String(message.key), value: message.value === null ? null : String(message.value), offset: Number(message.offset) }); last = now() } })
while (now() - last < 3000) await sleep(200)
await c.disconnect()
return got
}
try {
for (let v = 1; v <= 100; v++) await producer.send({ topic: TOPIC, messages: Array.from({ length: 10 }, (_, i) => ({ key: `k${i}`, value: `k${i}=v${v}` })) })
await producer.send({ topic: TOPIC, messages: [{ key: 'k9', value: null }] })
const before = await readAll()
console.log(`\n=== Trước khi dọn: ${before.length} message, ${new Set(before.map((m) => m.key)).size} key ===`)
// đóng segment hiện tại bằng một message mới sau khi segment.ms trôi qua, rồi chờ cleaner
const t0 = now()
let after = before
for (let round = 0; round < 12; round++) {
await sleep(5000)
await producer.send({ topic: TOPIC, messages: [{ key: 'tick', value: String(round) }] })
after = (await readAll()).filter((m) => m.key !== 'tick')
if (after.length < 50) break
}
console.log(`Sau khoảng ${Math.round((now() - t0) / 1000)} giây: còn ${after.length} message (không tính message 'tick' dùng để đóng segment)`)
for (const m of after.slice(0, 12)) console.log(` offset ${m.offset}: ${m.key} → ${m.value ?? 'null (tombstone)'}`)
console.log(`Key k9 sau tombstone: ${after.some((m) => m.key === 'k9' && m.value !== null) ? 'vẫn còn giá trị' : after.some((m) => m.key === 'k9') ? 'chỉ còn tombstone' : 'không còn message nào'}`)
} finally {
await producer.disconnect(); await admin.deleteTopics({ topics: [TOPIC] }).catch(() => {}); await admin.disconnect()
}