Có một nghịch lý vận hành mà lần đầu gặp tôi phải đọc lại bảng số ba lần: khởi động lại ngay một ứng dụng Kafka Streams lại chậm hơn nhiều so với chờ một lúc rồi mới bật. Hình dung bạn rời phòng họp rồi lập tức quay vào bằng một cửa khác — hệ thống giữ chỗ vẫn thấy ghế cũ của bạn "đang có người", nên nó bắt "bạn mới" đứng đợi tới khi chỗ đặt của "bạn cũ" hết hạn. Cả bài này có nhiều bất ngờ, nhưng đó là cái đáng nhớ nhất, và nó chỉ cần một dòng cấu hình để chữa.
Kafka Streams là thư viện xử lý dòng dữ liệu chạy ngay trong ứng dụng của bạn — không cần cụm riêng. Bài này đo một phép đếm đơn giản, và cả ba con số đều bất ngờ.
Đếm theo khoá
Bảy dòng cho một phép đếm hoàn chỉnh, chịu lỗi, có trạng thái:
StreamsBuilder b = new StreamsBuilder();
b.<String,String>stream("st-in")
.groupByKey()
.count()
.toStream()
.mapValues(String::valueOf)
.to("st-dem");
Chạy trên 200.000 tin với 5.000 khoá phân biệt:
đầu vào 200.000 bản ghi
đầu ra 5.340 bản ghi ít hơn 37 lần
mỗi khoá đếm được 40 đúng (200.000 / 5.000)
Kết quả đúng — mỗi khoá đếm được 40. Nhưng chỉ có 5.340 bản ghi ra, không phải 200.000.
Streams không phát một bản ghi ra cho mỗi bản ghi vào
Đây là điều tôi không mong đợi và nó thay đổi cách suy nghĩ về đầu ra.
Về mặt ngữ nghĩa, mỗi tin vào làm số đếm của một khoá thay đổi, nên "đúng" ra phải có 200.000 cập nhật. Streams có một bộ đệm bản ghi (cache.max.bytes.buffering, mặc định 10 MB) nuốt các cập nhật trung gian và chỉ phát giá trị mới nhất của mỗi khoá ở mỗi chu kỳ commit.
5.000 khoá × khoảng một lần phát mỗi khoá = 5.340. Con số khớp.
Hai hệ quả thực dụng:
Đừng dựa vào việc thấy mọi giá trị trung gian. Nếu hạ nguồn cần thấy từng bước thay đổi, phải tắt bộ đệm (cache.max.bytes.buffering=0) — và khi đó đầu ra sẽ là 200.000 bản ghi, với chi phí tương ứng.
Đây là một tính năng, không phải mất mát. Với bảng theo dõi hay bảng tổng hợp, chỉ giá trị cuối mới quan trọng, và giảm 37 lần lưu lượng ghi là món hời rất lớn.
Trạng thái nằm ở hai nơi
trên đĩa cục bộ RocksDB, 380 KB, 4 tệp SST
trong Kafka app-st-dem-KSTREAM-AGGREGATE-STATE-STORE-0000000001-changelog
cleanup.policy=compact
RocksDB trên đĩa là bản làm việc. Heap của ứng dụng chỉ 35 MB dù đang giữ 5.000 khoá — vì trạng thái nằm ngoài heap, trong RocksDB. Đây là lý do Kafka Streams giữ được trạng thái lớn hơn nhiều so với RAM mà không gây gai dọn rác.
Topic changelog trong Kafka là bản sao lưu. Streams tự tạo nó, và nó dùng cleanup.policy=compact — phần 18 đã đo: nén theo khoá khiến topic hội tụ về đúng cỡ trạng thái thay vì tăng mãi.
Cặp này là toàn bộ cơ chế chịu lỗi: máy chết thì RocksDB mất, nhưng changelog còn, và Streams dựng lại từ đó.
Khởi động lại ngay chậm hơn khởi động lại chậm — 90 lần
Tôi xoá kho trạng thái cục bộ rồi khởi động lại bốn lần liên tiếp:
| Thời gian tới RUNNING | Bản ghi khôi phục | |
|---|---|---|
| lần 1 — kho trống | 363 ms | 5.340 |
| lần 2 — ngay sau lần 1 | 44.313 ms | 0 |
| lần 3 — ngay sau lần 2 | 29.140 ms | 0 |
| lần 4 — sau khi chờ 50 giây | 463 ms | 0 |
Đọc bảng này ngược với mọi trực giác.
Dựng lại toàn bộ trạng thái từ changelog mất 363 ms. Việc mà ai cũng sợ hoá ra rất rẻ ở quy mô này.
Khởi động lại ngay lập tức mất 44 giây — và không dựng lại gì cả. Toàn bộ thời gian đó là chờ.
Nguyên nhân: Kafka Streams cố ý không rời nhóm khi đóng (internal.leave.group.on.close=false). Ý định là tốt — nó tránh một lần cân bằng lại khi ứng dụng chỉ khởi động lại nhanh. Nhưng bản mới tham gia với một danh tính thành viên khác, nên trong một khoảng thời gian nhóm có hai thành viên cho cùng số partition, và bản mới phải đợi phiên cũ hết hạn (session.timeout.ms, mặc định 45.000) mới nhận đủ partition để tới RUNNING.
Cách chữa là group.instance.id — thành viên tĩnh, phần 9 đã đo. Khi đó bản mới khai đúng danh tính cũ và nhận lại ngay phần cũ, không phải đợi gì.
group.instance.id=streams-instance-1 # lấy từ tên pod nếu chạy StatefulSet
session.timeout.ms=60000 # dài hơn thời gian khởi động ứng dụng
Nếu bạn từng thấy ứng dụng Streams "treo 45 giây mỗi lần triển khai", đây là nguyên nhân, và một dòng cấu hình xoá nó.
Gộp theo cửa sổ
s.groupByKey()
.windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofSeconds(5)))
.count()
.toStream((w, v) -> w.key() + "@" + w.window().start())
Trên cùng dữ liệu:
5.000 bản ghi ra
khoá dạng K4952@1788029660000
kho trạng thái 440 KB (so với 380 KB khi đếm không cửa sổ)
Khoá cửa sổ mang thêm mốc bắt đầu, nên số khoá phân biệt trong kho trạng thái là số khoá × số cửa sổ đang mở. Ở đây dữ liệu được ghi trong một giây nên tất cả rơi vào một cửa sổ, và kho chỉ lớn hơn 16%.
Với dữ liệu thật trải dài, con số nhân này là thứ quyết định kích thước kho. Cửa sổ 1 phút giữ 24 giờ nghĩa là mỗi khoá có 1.440 mục — và đó là lý do retention của cửa sổ (mặc định 24 giờ) là tham số đáng nhìn trước khi kho trạng thái phình ra.
Thời gian sự kiện, không phải thời gian tường
Cửa sổ của Streams chạy theo dấu thời gian của bản ghi, không theo đồng hồ máy. Đây là điều khiến kết quả lặp lại được: xử lý lại dữ liệu cũ cho ra đúng cùng một kết quả, dù chạy vào lúc nào.
Nó cũng có nghĩa là dữ liệu đến muộn có thể rơi ra ngoài cửa sổ. ofSizeWithNoGrace là tên gọi thẳng thắn: không có thời gian ân hạn, tin đến sau khi cửa sổ đóng thì bị bỏ. ofSizeAndGrace(Duration.ofSeconds(5), Duration.ofSeconds(30)) cho thêm 30 giây ân hạn, đổi lại kho trạng thái phải giữ cửa sổ lâu hơn.
Phần sau đo chính xác chuyện gì xảy ra với dữ liệu đến muộn khi nối hai luồng.
Ba thứ Streams tự tạo mà bạn phải biết
Chạy một ứng dụng Streams là để nó tự tạo topic trong cụm của bạn:
- changelog — sao lưu kho trạng thái,
compact - repartition — khi có
groupByđổi khoá, dữ liệu phải được rải lại __consumer_offsets— như mọi consumer
Chúng mang tên bắt đầu bằng application.id, và xoá ứng dụng không xoá chúng. Công cụ kafka-streams-application-reset.sh là thứ dọn đúng cách; xoá tay dễ để sót và lần chạy sau sẽ dựng lại trạng thái từ dữ liệu cũ.
Cùng một cái bẫy "danh tính không liền mạch thì phải đợi chính bóng ma của mình", đổi ngăn xếp
Nghịch lý "khởi động lại ngay lại chậm hơn" không phải đặc sản Kafka Streams. Nó là hệ quả của một mẫu chung: một tối ưu giả định rằng thứ quay lại chính là thứ vừa rời đi sẽ gãy khi bản quay lại mang một danh tính mới.
Trong thế giới cơ sở dữ liệu, đây đúng là cái bẫy của khoá và phiên: một tiến trình giữ một advisory lock rồi chết, tiến trình mới khởi động không lấy được khoá cho tới khi phiên cũ bị dọn theo timeout — nó đang đợi chính bóng ma của mình. Một leader dùng lease trong etcd/ZooKeeper cũng vậy: quay lại với session ID mới thì phải đợi hết TTL của lease cũ, còn quay lại với cùng session thì giành lại ngay.
Trong thế giới Java, một resource có tên — một lock file, một named mutex, một distributed lock kiểu Redisson với lease — bị tiến trình chết giữ lại sẽ bắt tiến trình mới (PID khác) đợi tới khi lease hết hoặc ai đó dọn file. Danh tính liền mạch biến cái đợi đó thành con số không.
Mẫu số chung: một tối ưu "đừng dọn dẹp vội vì thứ vừa đi có thể quay lại ngay" chỉ đúng khi thứ quay lại được nhận ra là chính nó; nếu bản quay lại mang một danh tính phù du mới, nó sẽ đụng phải phần tài nguyên mà bản cũ vẫn đang giữ và phải đợi lease của chính bóng ma mình hết hạn — nên cách chữa luôn là cho nó một danh tính ổn định để được nhận ra và giành lại ngay, thay vì một danh tính mới mỗi lần khởi động. Nên câu hỏi đúng khi một thứ "treo một lúc sau mỗi lần triển khai" không phải "nó đang dựng lại cái gì" mà là "nó có đang đợi chính phiên bản trước của mình hết hạn không — và tôi đã cho nó một danh tính cố định để khỏi phải đợi chưa".
Muốn xem ứng dụng Streams của bạn đang giữ bao nhiêu trạng thái, chạy thử:
# kho cục bộ
du -sh /tmp/kafka-streams/<application-id>
# topic nội bộ Streams đã tạo
kafka-topics.sh --bootstrap-server kf:9092 --list | grep "^<application-id>-"
# kích thước changelog
kafka-log-dirs.sh --bootstrap-server kf:9092 --describe \
--topic-list <application-id>-...-changelog | grep -oE '"size":[0-9]+'
Nếu changelog lớn hơn nhiều so với kho cục bộ, việc nén chưa chạy tới — và min.cleanable.dirty.ratio là chỗ đầu tiên nên nhìn.
Phần sau đo phép nối hai luồng: cửa sổ 5 giây cho 0 kết quả, cửa sổ 60 giây cho 5.000 — trên cùng một tập dữ liệu.
