Change data capture (CDC)

CDC là gì và mô hình cốt lõi

Change data capture (CDC) là kỹ thuật bắt mọi INSERT, UPDATE, DELETE đã commit trong một database và đẩy chúng ra ngoài thành luồng sự kiện có thứ tự. Các hệ thống khác (search index, cache, data warehouse, service khác) đọc luồng đó để cập nhật bản sao của mình. Ý này gần với việc dùng pub-sub để phân phối thay đổi dữ liệu: Grokking ch.18 liệt kê "replicating data" là một use case của pub-sub.¹

Vấn đề CDC giải quyết là dual write. Code kiểu này rất phổ biến:

await db.query("UPDATE products SET price = $1 WHERE id = $2", [price, id]);
await es.update({ index: "products", id, doc: { price } }); // có thể lỗi
await redis.del(`product:${id}`);                            // có thể không chạy tới

Nếu process chết giữa hai lệnh, hoặc Elasticsearch timeout, database và index lệch nhau mà không ai biết. Microservices Patterns nêu đúng rủi ro này: service cập nhật database rồi chết trước khi gửi message thì hệ thống rơi vào trạng thái không nhất quán, và transaction phân tán không phải lựa chọn tốt.² Thêm retry cũng không đủ: hai request cập nhật cùng một sản phẩm có thể ghi sang bên kia theo thứ tự ngược với thứ tự commit trong database (sách nêu bài toán thứ tự message ở §3.3.5).³

CDC đổi hướng: database là nguồn sự thật duy nhất. Ứng dụng chỉ ghi vào database. Một tiến trình riêng đọc nhật ký thay đổi của database theo đúng thứ tự commit và cập nhật mọi nơi khác. Nếu tiến trình này chết, nó đọc tiếp từ vị trí đã lưu, không mất thay đổi nào; event relay của Eventuate trong sách cũng lưu vị trí binlog (tên file và offset) vào một topic Kafka để khởi động lại đúng chỗ.⁴

Mỗi sự kiện CDC thường chứa:

Cấu trúc sự kiện dưới đây theo định dạng của Debezium.⁵

• op: c (create), u (update), d (delete), r (read, từ snapshot ban đầu).
• before và after: giá trị dòng trước và sau thay đổi.
• source: database, bảng, vị trí trong log (LSN của Postgres, binlog position của MySQL), transaction id.
• ts_ms: thời điểm commit.

Ba cách bắt thay đổi

Polling theo cột thời gian

Định kỳ chạy SELECT * FROM products WHERE updated_at > :last_seen. Đơn giản, không cần quyền đặc biệt. Microservices Patterns nhận xét polling database "works reasonably well at low scale" nhưng polling thường xuyên có thể tốn kém.⁶ Các nhược điểm còn lại: không thấy DELETE (trừ khi dùng soft delete), bỏ sót khi hai thay đổi cùng timestamp hoặc khi transaction dài commit muộn với updated_at cũ. Độ trễ bằng chu kỳ polling.⁷

Trigger

Tạo trigger trên mỗi bảng, ghi thay đổi vào bảng audit_log trong cùng transaction. Bắt được cả delete, chính xác theo transaction. Nhược điểm: mỗi lần ghi nghiệp vụ thành hai lần ghi, làm chậm đường ghi chính; trigger là logic ẩn khó bảo trì; vẫn phải polling bảng audit để đẩy ra ngoài.⁷

Đọc log của database (log-based)

Database ghi write-ahead log trước khi sửa dữ liệu để phục hồi khi crash, và các database dùng log này (hoặc log theo dòng) để replication. Grokking ch.9 phân biệt WAL shipping (gắn chặt với cấu trúc nội bộ của engine) và logical, row-based log replication (ghi thay đổi ở mức dòng, không phụ thuộc cách bố trí dữ liệu).⁸ MongoDB giữ oplog chứa mọi thao tác ghi của primary để secondary đọc và sao chép.⁹ CDC dựa trên log đọc chính loại log đó, giống như một replica:

Tên cấu hình ở các hàng PostgreSQL, MySQL và SQL Server là chi tiết sản phẩm.¹⁰ Hàng DynamoDB và ý "đọc log như replica" khớp với sách.²

Microservices Patterns gọi cách này là transaction log tailing và nhận xét rằng dù còn ít người biết, nó hoạt động rất tốt, nhưng cần công sức phát triển (tự viết code gọi API của database hoặc dùng framework như Debezium). Sách liệt kê các ví dụ như Debezium, LinkedIn Databus, DynamoDB Streams, Eventuate Tram.² Ưu điểm: không thêm truy vấn vào database, không sửa code ứng dụng, bắt được mọi thay đổi kể cả delete, đúng thứ tự commit (mỗi thay đổi đã commit là một mục trong transaction log).²

Bảng so sánh tổng hợp từ Microservices Patterns (polling vs log tailing) và nhận định của khoá học cho cột trigger.⁶⁷

Kiến trúc: Debezium và Kafka

Debezium là bộ connector mã nguồn mở chạy trên Kafka Connect. Mỗi connector đọc log của một database và ghi sự kiện vào Kafka, mặc định mỗi bảng một topic (ví dụ shop.public.products).⁵ Key của message là primary key của dòng, nên mọi thay đổi của một dòng vào cùng partition và giữ đúng thứ tự: đây là cơ chế sharded channel với shard key mà Microservices Patterns mô tả (mọi event của cùng một order vào cùng shard và được một consumer instance đọc).³

Luồng hoạt động:

1. Snapshot ban đầu. Khi mới bật, connector đọc toàn bộ dữ liệu hiện có (sự kiện op = r), rồi chuyển sang đọc log từ đúng vị trí lúc bắt đầu snapshot. Với bảng lớn, dùng incremental snapshot để chia nhỏ và chạy song song với streaming.⁵
2. Streaming. Đọc log liên tục, chuyển mỗi thay đổi thành sự kiện.
3. Lưu offset. Vị trí đã đọc (LSN) được lưu vào topic offset của Kafka Connect (cùng ý với event relay của Eventuate lưu vị trí binlog vào một topic Kafka).⁴ Với Postgres, connector cũng xác nhận LSN cho replication slot để database biết có thể dọn WAL cũ.

Cấu hình và sự kiện mẫu

{
  "name": "shop-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "pg-primary",
    "database.dbname": "shop",
    "plugin.name": "pgoutput",
    "slot.name": "debezium_shop",
    "topic.prefix": "shop",
    "table.include.list": "public.products,public.outbox"
  }
}

{
  "op": "u",
  "before": { "id": 42, "price": 120000 },
  "after":  { "id": 42, "price": 99000 },
  "source": { "table": "products", "lsn": 24023128, "txId": 5531 },
  "ts_ms": 1730534400123
}

Consumer đơn giản đồng bộ sang Elasticsearch:

await consumer.run({
  eachMessage: async ({ message }) => {
    const id = message.key!.toString();
    if (!message.value) return; // tombstone sau delete
    const ev = JSON.parse(message.value.toString());
    if (ev.op === "d") await es.delete({ index: "products", id }, { ignore: [404] });
    else await es.index({ index: "products", id, document: ev.after, version: ev.source.lsn, version_type: "external" });
  },
});

version_type: external với LSN làm version giúp Elasticsearch bỏ qua sự kiện cũ nếu chúng tới muộn hoặc bị gửi lại.¹¹ Đây cũng là cách làm event handler idempotent: ghi lại thông tin để phát hiện và bỏ sự kiện trùng.¹²

Outbox và CDC

CDC thô phát ra thay đổi ở mức bảng. Điều này có hai vấn đề khi dùng để giao tiếp giữa các service: schema bảng nội bộ bị lộ ra ngoài (đổi tên cột là phá consumer), và một thao tác nghiệp vụ có thể sửa nhiều bảng, consumer phải tự ghép lại. Building Microservices cũng so sánh: data pump chạy ở mức database gắn chặt với schema nội bộ, còn event data pump chỉ gắn với các event được thiết kế để công bố ra ngoài nên ít phụ thuộc hơn.¹³

Transactional outbox (Microservices Patterns §3.3.7) giải quyết bằng cách: trong cùng transaction với thay đổi nghiệp vụ, ứng dụng chèn một dòng vào bảng outbox chứa sự kiện nghiệp vụ đã được thiết kế cẩn thận (OrderCreated với payload công khai). Một message relay đọc bảng outbox và phát ra broker. Sách nêu hai cách làm relay: polling publisher (truy vấn bảng outbox định kỳ, đơn giản nhưng tốn tải khi poll nhiều) và transaction log tailing; việc ghi vào outbox nguyên tử vì là một local ACID transaction.²⁶ Kết hợp outbox với CDC chính là cách thứ hai: Debezium chỉ theo dõi bảng outbox, và Outbox Event Router của Debezium định tuyến mỗi dòng tới topic theo cột aggregatetype.⁵

BEGIN;
UPDATE orders SET status = 'PAID' WHERE id = 991;
INSERT INTO outbox (id, aggregatetype, aggregateid, type, payload)
VALUES (gen_random_uuid(), 'order', '991', 'OrderPaid', '{"orderId":991,"amount":250000}');
COMMIT;

Con số và sizing

Các con số dưới đây là ước tính (giả định của khoá học), cần đo lại với hệ thống thật.¹⁴

• Kích thước sự kiện. Envelope có cả before và after nên lớn hơn dòng gốc; ví dụ dưới đây giả định 2 KB mỗi thay đổi. Định dạng nhị phân có schema thường gọn hơn JSON.
• Thông lượng connector. Log là tuần tự nên một connector đọc một nguồn log; thông lượng phụ thuộc kích thước dòng và cấu hình, cần đo. Muốn hơn thì tách theo database hoặc theo nhóm bảng.
• Kafka. Giả sử một partition chịu được cỡ 10 MB/s (giả định, cần đo). 5.000 thay đổi/s × 2 KB = 10 MB/s, nên topic bảng nóng cần vài partition. Key theo primary key giữ thứ tự cho từng dòng.³
• Độ trễ đầu cuối. Từ commit tới consumer thường là từ mili giây tới vài giây khi hệ thống khỏe (ước lượng). Snapshot ban đầu của bảng rất lớn có thể mất nhiều giờ.

Ví dụ: shop có 3.000 đơn/phút lúc flash sale, mỗi đơn sửa khoảng 5 dòng (order, order_items, inventory, payment, outbox). 3.000 × 5 / 60 = 250 thay đổi/s, cao điểm ×10 là 2.500/s. Một connector và topic 6 partition đủ dùng. WAL sinh ra khoảng 2.500 × 2 KB ≈ 5 MB/s; nếu connector dừng 6 giờ, slot giữ lại khoảng 100 GB WAL.

Khi nào dùng và khi nào không

Không nên dùng CDC khi team chưa vận hành Kafka và nhu cầu nhỏ. Chi phí vận hành connector, slot, Schema Registry lớn hơn lợi ích. Microservices Patterns cũng khuyên polling hợp ở quy mô nhỏ.⁶

Bảng trên là nhận định tổng hợp; ý về outbox, CQRS và polling có nguồn ở Microservices Patterns, các hàng khác là kinh nghiệm của khoá học.¹⁵⁷

Vận hành và các bẫy hay gặp

Replication slot làm đầy đĩa

Postgres giữ WAL cho tới khi slot xác nhận đã đọc. Connector chết cuối tuần là đĩa primary đầy dần, và khi đầy thì database ngừng ghi. Luôn có alert theo pg_replication_slots (WAL bị giữ lại) và đặt max_slot_wal_keep_size để giới hạn. Xóa slot không dùng nữa.¹⁰

Thứ tự

Kafka chỉ đảm bảo thứ tự trong một partition (mỗi shard của sharded channel hoạt động như một channel riêng).³ Thay đổi của cùng một dòng giữ thứ tự vì cùng key. Nhưng thay đổi trên hai bảng khác nhau (order và order_items) nằm ở hai topic, consumer có thể thấy items trước order. Cách xử lý: thiết kế consumer chịu được dữ liệu tới lệch nhau, hoặc dùng outbox để gói một sự kiện chứa đủ thông tin.

Gửi trùng

Phần lớn message broker chỉ hứa giao ít nhất một lần (at-least-once) vì exactly-once thường quá đắt, nên khi client, mạng hoặc broker lỗi, message có thể được giao nhiều lần.¹⁶ Sau khi restart, vài sự kiện cuối có thể được gửi lại. Consumer phải idempotent: handler không idempotent (ví dụ cộng tiền vào số dư) phải ghi lại ID sự kiện đã xử lý cùng lúc, nguyên tử, với việc cập nhật dữ liệu; còn upsert theo id và dùng LSN hoặc version để bỏ sự kiện cũ là cách cụ thể hóa.¹²

Schema evolution

Thêm cột nullable thường an toàn. Đổi tên cột, đổi kiểu, xóa cột sẽ phá consumer. Quy trình đổi schema theo kiểu thêm cột mới, ghi cả hai, chuyển consumer, rồi mới xóa cột cũ chính là mẫu expand and contract mà Building Microservices mô tả cho việc đổi interface (cho hai phiên bản cùng tồn tại một thời gian).¹⁷ Schema Registry với chế độ tương thích BACKWARD là công cụ hỗ trợ.⁵ Với Postgres, sự kiện update mặc định không chứa giá trị cũ đầy đủ và cột TOAST không đổi có thể bị bỏ trống; đặt REPLICA IDENTITY FULL cho bảng cần before đầy đủ, chấp nhận WAL lớn hơn.¹⁰

Quy mô

• Nhỏ: dịch vụ managed, hoặc ghi thẳng vào một broker nhẹ mà không cần tự vận hành Kafka. Một connector cho mỗi database.¹⁸
• Lớn: connector riêng cho từng database, Kafka Connect cluster nhiều worker, Schema Registry bắt buộc, dashboard độ trễ theo connector, quy trình chạy lại snapshot cho từng bảng khi consumer mới cần dữ liệu cũ. Đặt retention Kafka đủ dài hoặc dùng compacted topic để consumer mới có thể đọc lại; tuy vậy Microservices Patterns lưu ý broker không nhằm lưu event vô hạn, nên dựng view mới thường phải đọc thêm event đã archive.¹⁹

Điểm nói trong phỏng vấn

1. Gọi tên vấn đề dual write khi thấy thiết kế ghi vào database rồi ghi tiếp sang search, cache hay broker. Đề xuất CDC hoặc outbox.
2. So sánh ba cách bắt thay đổi và chọn log-based vì bắt được delete, đúng thứ tự, ít tải.
3. Nêu Debezium + Kafka, key theo primary key để giữ thứ tự cho từng dòng.
4. Phân biệt CDC trực tiếp và outbox theo người nhận: bản sao dữ liệu hay service khác.
5. Chủ động nói về bẫy: at-least-once nên consumer idempotent,¹⁶ thứ tự chỉ trong partition,³ slot giữ WAL, schema evolution.
6. Độ trễ là eventual consistency. Index search chậm hơn database một khoảng trễ; Microservices Patterns gọi đây là replication lag của CQRS: client vừa ghi rồi query ngay có thể thấy bản cũ. Nói rõ màn hình nào cần đọc thẳng database hoặc cập nhật model cục bộ sau khi ghi.¹⁵

Bài liên quan: Kafka, Caching cho chiến lược invalidation, Quy trình nhiều bước cho saga dùng outbox.

Nguồn trích dẫn

1. Grokking Modern System Design · ch.18 Pub-sub · mục "Use cases of pub-sub" (Replicating data). ↩

2. Microservices Patterns (file .zip, tải về giải nén) · ch.3 · §3.3.7 Transactional messaging (tr. 97–100): "Using a database table as a message queue", "Publishing events by applying the Transaction log tailing pattern". ↩ ↩² ↩³ ↩⁴ ↩⁵

3. Microservices Patterns (file .zip, tải về giải nén) · ch.3 · §3.3.5 Competing receivers and message ordering (tr. 94–95, sharded channel và shard key). ↩ ↩² ↩³ ↩⁴ ↩⁵

4. Microservices Patterns (file .zip, tải về giải nén) · ch.6 · §6.2.1 The Eventuate Local event store · "The Eventuate Local event relay propagates events from the database to the message broker" (binlog, lưu vị trí vào topic Kafka). ↩ ↩²

5. Chưa có nguồn trong thư viện: sách chỉ nhắc tên Debezium; định dạng sự kiện, connector, snapshot, Outbox Event Router và Schema Registry không có trong sách, chỉ nêu khái niệm cơ bản. ↩ ↩² ↩³ ↩⁴ ↩⁵

6. Microservices Patterns (file .zip, tải về giải nén) · ch.3 · §3.3.7 Transactional messaging · "Publishing events by using the Polling publisher pattern" (tr. 98–99). ↩ ↩² ↩³ ↩⁴

7. Chưa có nguồn trong thư viện: so sánh trigger với polling và các điểm yếu chi tiết của polling theo timestamp là nhận định của khoá học. ↩ ↩² ↩³ ↩⁴

8. Grokking Modern System Design · ch.9 Databases · mục "Replication" · "Primary-secondary replication methods" (WAL shipping, logical row-based log replication). ↩

9. MongoDB: The Definitive Guide 2e (file .zip, tải về giải nén) · ch.10 Components of a Replica Set · mục "Syncing" (oplog). ↩

10. Chưa có nguồn trong thư viện: thư viện không có sách về logical decoding, replication slot hay cấu hình WAL/binlog của PostgreSQL và MySQL; chỉ nêu khái niệm cơ bản. ↩ ↩² ↩³

11. Chưa có nguồn trong thư viện: thư viện không có sách về versioning ngoài (external version) của Elasticsearch; chỉ nêu ý tưởng bỏ sự kiện cũ. ↩

12. Microservices Patterns (file .zip, tải về giải nén) · ch.7 · §7.3.2 Data access module design · "Idempotent event handlers" (tr. 240–241). ↩ ↩²

13. Building Microservices (file .zip, tải về giải nén) · ch.5 Splitting the Monolith · mục "Data Pumps" và "Event Data Pump" (tr. 96–99). ↩

14. Chưa có nguồn trong thư viện: các con số kích thước sự kiện, thông lượng và độ trễ là giả định của khoá học để tính nhanh, không phải số liệu từ sách. ↩

15. Microservices Patterns (file .zip, tải về giải nén) · ch.7 · §7.2 Using the CQRS pattern · "Overview of CQRS" và "The drawbacks of CQRS" (Dealing with the replication lag, tr. 235–236). ↩ ↩²

16. Microservices Patterns (file .zip, tải về giải nén) · ch.3 · §3.3.6 Handling duplicate messages (tr. 95–96, at-least-once, idempotent handler, theo dõi và bỏ message trùng). ↩ ↩²

17. Building Microservices (file .zip, tải về giải nén) · ch.4 Integration · mục "Defer It for as Long as Possible" và "Coexist Different Endpoints" (expand and contract, tr. 62–64). ↩

18. Chưa có nguồn trong thư viện: gợi ý vận hành theo quy mô là nhận định của khoá học. ↩

19. Microservices Patterns (file .zip, tải về giải nén) · ch.7 · §7.3.3 Adding and updating CQRS views (tr. 241–242, broker không lưu event vô hạn, dựng view từ event đã archive). ↩