harry.le

· kafka · debezium · cdc · event-driven · postgresql

Change data capture với Debezium và Kafka: bài học khi chạy thật

Khi nào nên dùng CDC, outbox pattern, và các bẫy tôi gặp với replication slot, snapshot, thứ tự event và consumer nhận trùng trên PostgreSQL và SQL Server.

Read in English →

Tôi dùng Debezium ở hai chỗ trên một nền tảng SaaS logistics. Chỗ thứ nhất đưa thay đổi từ một SQL Server cũ sang các service PostgreSQL mới. Chỗ thứ hai chuyển bảng outbox từ PostgreSQL ra Kafka. Cả hai chạy tốt, và cả hai dạy tôi những điều tài liệu chỉ nhắc qua. Bài này là thứ tôi muốn tự nói với mình trước khi bắt đầu.

Khi nào đáng dùng CDC

Dùng CDC khi hai hệ thống phải khớp nhau mà bạn không sửa được nhiều phía ghi, hoặc khi không được phép mất event. SQL Server cũ là trường hợp đầu: không ai muốn đụng vào code của nó, nhưng service mới cần dữ liệu trong vài giây. Một cron poll thay đổi vẫn chạy làm lưới an toàn, nhưng nó chỉ mới được bằng chu kỳ của nó.

Trường hợp thứ hai là dual write. Service lưu một dòng rồi gọi Kafka: hai commit trên hai hệ thống, không có transaction chung. Chết giữa chừng thì mất event. Publish trước mà DB rollback thì gửi event cho việc chưa từng xảy ra. Chúng tôi dính đúng lỗi đầu: có user tồn tại ở service này mà không có ở service khác, vì bước publish sau commit fail mà không báo.

Nếu chưa có cụm Kafka Connect, polling publisher trên bảng outbox đơn giản hơn. Tôi chọn CDC cho outbox vì cụm Connect đã có sẵn từ pipeline SQL Server.

Kiến trúc

Với outbox, service ghi dòng nghiệp vụ và dòng outbox trong cùng một local transaction. Debezium đọc WAL của PostgreSQL, bộ định tuyến outbox của nó đẩy mỗi dòng ra một topic. Consumer đọc từ Kafka.

service -> [dòng nghiệp vụ + dòng outbox] một transaction
        -> WAL -> Debezium (outbox router) -> topic Kafka -> consumers

Config connector rút gọn, dùng tên giả:

{
  "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
  "plugin.name": "pgoutput",
  "slot.name": "outbox_slot",
  "publication.name": "dbz_outbox",
  "table.include.list": "public.outbox",
  "snapshot.mode": "never",
  "heartbeat.interval.ms": "300000",
  "transforms": "outbox",
  "transforms.outbox.type": "io.debezium.transforms.outbox.EventRouter",
  "transforms.outbox.table.field.event.id": "event_id",
  "transforms.outbox.route.topic.replacement": "domain.events"
}

Cột event id thành header của message, aggregate id thành key, payload thành value. Ba lựa chọn này quan trọng ở các phần sau.

Các bẫy

Slot bị bỏ quên làm đầy đĩa. Replication slot của PostgreSQL giữ WAL cho tới khi consumer đọc. Debezium chết, hoặc connector bị xoá mà slot còn, thì WAL chất đống đến đầy đĩa, trên một DB dùng chung cho hơn chục service. Tôi đặt max_slot_wal_keep_size = 10GB để slot mồ côi bị invalidate thay vì kéo sập DB, và kiểm pg_replication_slots tìm slot không active. Heartbeat ở trên giúp slot tiến lên khi bảng im.

Snapshot. snapshot.mode=never nghĩa là các dòng có trước slot sẽ không bao giờ được relay. Với outbox thì ổn, nhưng tôi từng ghi dòng test vào bảng trước đó và chúng bị bỏ qua mà không báo. Chọn mode có chủ đích, và đừng insert dòng test vào outbox đang route tới topic thật, vì consumer sẽ xử lý chúng.

SQL Server có kiểu lỗi riêng. CDC phụ thuộc SQL Server Agent: job capture dừng thì Debezium vẫn kết nối nhưng không gửi gì. Job cleanup xoá dòng trong change table sau thời gian retention, mặc định 3 ngày; connector lag quá mức đó thì phải snapshot lại. Bảng đã bật CDC cũng không TRUNCATE được.

Topic nội bộ của Connect. Các topic offsets, configs, status tự tạo theo default của broker và bị sai. Triệu chứng là connector ở trạng thái RUNNING nhưng 0 task, không có lỗi, trong khi WAL cứ tăng. Hãy tự tạo các topic này với đúng số partition và cleanup.policy=compact trước lần khởi động đầu.

Thứ tự. Kafka chỉ đảm bảo thứ tự trong một partition. Dùng aggregate id làm key thì mọi event của một thực thể giữ đúng thứ tự. Event của các thực thể khác nhau không có thứ tự với nhau, và consumer đừng giả định có. Tôi thà thiết kế handler chịu được điều đó còn hơn đuổi theo thứ tự toàn cục.

Đổi schema. Tôi ít kinh nghiệm xương máu ở mục này hơn mức nó đáng có, nên hãy coi đây là ý định thiết kế. Payload outbox là một hợp đồng riêng, nên đổi tên cột ở bảng nghiệp vụ không lan tới consumer. CDC thô trên bảng, như pipeline SQL Server, thì lộ schema bảng, và ở đó tôi dựa vào thay đổi cộng thêm và deploy consumer trước producer. Quy tắc thứ tự deploy trong ghi chú của chúng tôi cùng ý đó: producer thêm field không được lên trước consumer cần field đó, nếu không event vào dead letter topic hàng loạt.

Consumer idempotent

Relay là at-least-once. Debezium có thể gửi một dòng rồi chết trước khi ghi offset, nên trùng lặp sẽ xảy ra. Idempotent producer của Kafka không giúp ở đây vì nó chỉ bao biên giới Kafka, không bao DB của bạn.

Vì vậy mỗi consumer ghi event id đã xử lý vào một bảng, trong cùng transaction với phần ghi nghiệp vụ:

INSERT INTO processed_event (consumer_id, event_id, processed_at)
VALUES (:consumer, :event, now())
ON CONFLICT DO NOTHING;
-- 0 dòng được insert: trùng, bỏ qua phần ghi nghiệp vụ

Khoá là consumer group cộng event id lấy từ header, nên hai consumer xử lý cùng một event độc lập với nhau. Dòng cũ hơn một tuần được cron dọn. Lấy id từ header thay vì payload cũng giúp tôi khỏi parse gì cả chỉ để kiểm tra trùng.

Hai lỗi nhỏ đáng biết. Ghi payload dạng chuỗi vào cột JSONB làm lưu ra một chuỗi bị escape hai lần, consumer decode sai. Và vài hàm factory tạo event quên set event id, gây null pointer ở phía consumer.

Nếu làm lại

  1. Bắt đầu bằng outbox, không phải CDC mức bảng, khi bạn sở hữu service ghi.
  2. Đặt max_slot_wal_keep_size và alert trạng thái slot, lag trước khi connector đầu tiên lên. Slot không được giám sát là lỗi gây hại nhất, mà phòng thì rất rẻ.
  3. Giữ batch sync định kỳ cạnh CDC cho nguồn legacy. Nó vá khoảng trống sau khi connector restart.
  4. Diễn tập kill và replay: dừng Debezium giữa chừng, bật lại, replay offset cũ, rồi kiểm không mất không trùng.

CDC chuyển phần khó từ “publish cho đáng tin” sang “vận hành replication slot và viết consumer chịu được lặp”. Đó là bài toán tốt hơn, nhưng vẫn là bài toán.