50.000 UPDATE và 10.000 DELETE trong PostgreSQL sinh ra đúng 0 bản ghi trong Kafka — connector vẫn báo RUNNING, và dữ liệu sai nằm đó vĩnh viễn

Tưởng tượng một thủ thư kiểm tra sách mới bằng cách chỉ nhìn số thứ tự cao nhất trong sổ đăng ký. Cách này bắt được mọi cuốn mới nhập — nhưng một cuốn bị sửa lại nội dung (giữ nguyên số) thì vô hình, và một cuốn bị rút khỏi kệ thì càng vô hình. Thủ thư báo "đã cập nhật đầy đủ", trong khi bản sao của anh ta âm thầm trôi xa khỏi thực tế trên kệ. mode=incrementing của Kafka Connect chính là người thủ thư đó, và tôi sẽ đo đúng chỗ nó mù.

Kafka Connect đưa dữ liệu vào và ra Kafka bằng JSON cấu hình thay vì mã. Bài này dựng một đường ống PostgreSQL → Kafka, đo nó, và tìm ra một cách hỏng im lặng hoàn toàn.

Dựng bằng một lời gọi HTTP

curl -X POST http://cn:8083/connectors -H "Content-Type: application/json" -d '{
 "name":"pg-source",
 "config":{
  "connector.class":"io.confluent.connect.jdbc.JdbcSourceConnector",
  "connection.url":"jdbc:postgresql://pg:5432/shop",
  "connection.user":"postgres","connection.password":"pw",
  "table.whitelist":"don_hang",
  "mode":"incrementing","incrementing.column.name":"id",
  "topic.prefix":"pg-","poll.interval.ms":"1000"}}'

Không có mã nào. Connect đọc bảng, biến mỗi dòng thành một bản ghi Kafka, và nhớ vị trí đã đọc tới.

trạng thái: RUNNING | task: ['RUNNING']
{"id":1,"ma":"ORD-1","khach":"CUST-1","tien":137,"trang_thai":"PENDING",...}

Số đo

Hai điều đáng chú ý.

Connect nặng hơn broker. Một broker Kafka tốn 339 MB (phần 21); Connect tốn hơn một gigabyte. Nó là một ứng dụng Java đầy đủ với hệ thống nạp plugin, REST API và cơ chế phân tán riêng.

Thông lượng thì rất tốt — hơn một trăm nghìn dòng mỗi giây từ PostgreSQL. Nút thắt không nằm ở Connect.

Độ trễ bị chặn bởi poll.interval.ms. Đặt 1.000 nghĩa là trung bình chờ nửa giây trước khi Connect nhìn lại bảng. (Con số đo được còn cộng thêm khoảng 0,9 giây thời gian khởi động JVM của công cụ tôi dùng để kiểm — phần 2 đã đo. Độ trễ thật của Connect thấp hơn con số trong bảng.)

Cái bẫy: mode=incrementing mù với UPDATE và DELETE

Đây là phát hiện quan trọng nhất của bài.

Tôi chạy trong PostgreSQL:

update don_hang set trang_thai='DA_HUY', tien=0 where id <= 50000;   -- 50.000 dòng
delete from don_hang where id between 50001 and 60000;                -- 10.000 dòng

Rồi đo Kafka:

offset trước:   200.003
offset sau:     200.003        chênh 0

Không một bản ghi nào được sinh ra.

Trong PostgreSQL, 50.000 dòng đã là DA_HUY. Trong Kafka, đọc lại bản ghi id=1:

id 1   trang_thai = PENDING   tien = 137

Dữ liệu cũ, và nó sẽ ở đó vĩnh viễn.

Điều tệ nhất là không có lỗi, không cảnh báo, không dòng log nào. Connector vẫn RUNNING. Bảng theo dõi vẫn xanh. Mọi thứ trông hoàn hảo, và hạ nguồn đang tính toán trên dữ liệu sai.

Cơ chế thì rõ ràng khi nghĩ lại: mode=incrementing chạy truy vấn select * from don_hang where id > <đã_đọc_tới>. UPDATE không đổi id, nên dòng đó không bao giờ lọt vào lần đọc sau. DELETE thì càng không — dòng đã biến mất khỏi bảng.

Bốn chế độ, và cái gì chúng thấy

Không chế độ nào thấy DELETE.

Và timestamp có điều kiện riêng của nó: cột thời gian phải được mọi đường ghi cập nhật. Một câu UPDATE do ai đó chạy tay quên đụng tới cột đó là một thay đổi Connect không bao giờ thấy. Trigger giúp được, nhưng khi ấy bạn đã đang sửa lược đồ cơ sở dữ liệu nguồn — và nếu chấp nhận làm vậy thì CDC là lựa chọn tốt hơn.

Kết luận thực dụng: JDBC source chỉ phù hợp với bảng chỉ-thêm. Nhật ký sự kiện, bản ghi audit, dữ liệu đo. Với bảng nghiệp vụ có sửa và xoá, nó cho ra dữ liệu sai một cách im lặng — và phần sau đo giải pháp đúng.

Plugin phải cài riêng

Ảnh Connect chỉ có sẵn ba connector MirrorMaker. Mọi thứ khác phải cài thêm:

confluent-hub install --no-prompt confluentinc/kafka-connect-jdbc:10.7.6
# rồi khởi động lại Connect

Bước khởi động lại là chỗ dễ quên nhất khi dựng lần đầu: cài xong, gọi API tạo connector, nhận lỗi "class not found", và tưởng cài hỏng.

Kiểm bằng:

curl -s http://cn:8083/connector-plugins | jq -r '.[].class'

Sink connector: chiều ngược lại

Cùng cơ chế, ngược hướng — Kafka → cơ sở dữ liệu:

{"connector.class":"io.confluent.connect.jdbc.JdbcSinkConnector",
 "topics":"pg-don_hang",
 "connection.url":"jdbc:postgresql://dw:5432/kho",
 "insert.mode":"upsert","pk.mode":"record_key","pk.fields":"id",
 "auto.create":"true"}

insert.mode=upsert là thứ khiến việc đọc trùng (phần 8) không gây hại: đọc lại cùng bản ghi thì ghi đè, không tạo dòng thừa. Đây là ví dụ cụ thể của nguyên tắc "làm phía tiêu thụ chịu được lặp".

Ba thứ nên biết khi vận hành

Connect chạy phân tán được và nên chạy như vậy. Nhiều thực thể cùng group.id chia nhau các task, và một thực thể chết thì task được giao lại. Cấu hình, offset và trạng thái nằm trong ba topic Kafka nội bộ, nên không có trạng thái cục bộ nào cần sao lưu.

tasks.max là mức song song. JDBC source chia theo bảng: một bảng thì tasks.max lớn hơn 1 không giúp gì.

Trạng thái RUNNING của connector không có nghĩa task đang chạy. Phải xem cả hai:

curl -s http://cn:8083/connectors/pg-source/status | jq '{c:.connector.state, t:[.tasks[].state]}'

Connector RUNNING với task FAILED là tình huống có thật, và bảng theo dõi chỉ nhìn trường đầu sẽ báo khoẻ.

Cùng một giới hạn "cơ chế phát hiện quyết định cái thấy được", đổi ngăn xếp

Cái bẫy lõi ở đây — theo dõi một khoá tăng đơn điệu thì chỉ thấy phần thêm vào, mù với sửa và xoá — không phải lỗi của Kafka Connect. Nó là hệ quả cấu trúc của cách bạn chọn để phát hiện thay đổi.

Trong thế giới cơ sở dữ liệu, đây đúng là cái bẫy của ETL gia tăng kiểu WHERE updated_at > lan_chay_truoc: một dòng bị UPDATE mà quên chạm cột updated_at, hay một dòng bị xoá cứng, sẽ không bao giờ lọt vào mẻ delta — delta chạy xanh, dữ liệu kho trôi dần. Đây chính là lý do người ta phải dùng xoá mềm kèm cờ deleted_at, hoặc chuyển hẳn sang đọc nhật ký thay đổi (WAL/CDC).

Trong thế giới lập trình nói chung, cùng cái mù: một file-watcher thăm dò theo mtime bỏ sót cả một lần sửa nội dung giữ nguyên mtime lẫn một lần xoá (trừ khi bạn diff cả danh sách); một công cụ đồng bộ "chỉ thêm" bỏ sót mọi thứ git status bắt được bằng cách so toàn bộ trạng thái. Cùng một nguồn, ba cách phát hiện, ba tập con khác nhau của "cái gì đã đổi".

Mẫu số chung: có ba cách phát hiện thay đổi của một nguồn — theo dõi phần thêm (con trỏ tăng đơn điệu), so toàn bộ trạng thái (diff, thấy hết nhưng đắt), hoặc đọc nhật ký thay đổi (WAL/CDC, thấy hết kể cả xoá mà vẫn rẻ) — và mỗi cách về mặt cấu trúc chỉ thấy một tập con: theo dõi phần thêm không bao giờ thấy được sửa hay xoá, dù bạn cấu hình thế nào đi nữa; và khi chọn sai, cái mù diễn ra trong im lặng vì "không có gì mới sau con trỏ" trông y hệt "không có gì thay đổi". Nên câu hỏi đúng khi dựng bất kỳ đường ống đồng bộ nào không phải "nó có chạy không" mà là "cơ chế phát hiện của tôi về mặt cấu trúc thấy được những loại thay đổi nào — và những loại nó không thể thấy có xảy ra trong dữ liệu của tôi không".

Muốn tự kiểm connector JDBC của bạn có đang bỏ sót thay đổi không, chạy thử:

# số dòng trong nguồn
psql -h pg -U postgres -d shop -tAc "select count(*) from don_hang"

# số bản ghi trong Kafka
kafka-get-offsets.sh --bootstrap-server kf:9092 --topic pg-don_hang

Nếu Kafka lớn hơn nguồn, đã có DELETE mà Kafka không biết. Nếu chúng bằng nhau nhưng nội dung khác — như bảng trang_thai ở trên — đã có UPDATE bị bỏ qua. Cả hai đều là dữ liệu sai đang được phục vụ.

Phần sau đo Debezium: cùng bài toán, nhưng đọc từ WAL nên thấy được cả UPDATE và DELETE.