Source & Debezium CDC

Source connector là đầu vào: nó lấy dữ liệu từ một nguồn bên ngoài rồi sản xuất message vào Kafka topic. Debezium là họ source connector nổi tiếng nhất, chuyên làm CDC (Change Data Capture).

Thay vì poll bảng bằng SQL (tốn tài nguyên, dễ sót), Debezium đọc transaction log mà database tự ghi khi có thay đổi:

Nhờ đọc log, Debezium bắt được cả những thay đổi rất nhanh, không bỏ sót, và gần như không tạo tải truy vấn lên DB nguồn.

Một connector chỉ là JSON. Dưới đây là cấu hình Debezium cho MySQL:

{
  "name": "mysql-source",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "tasks.max": "1",
    "database.hostname": "mysql",
    "database.port": "3306",
    "database.user": "debezium",
    "database.password": "dbz",
    "database.server.id": "184054",
    "topic.prefix": "mysqlsrv",
    "database.include.list": "shop",
    "table.include.list": "shop.orders,shop.customers",
    "schema.history.internal.kafka.bootstrap.servers": "kafka:9092",
    "schema.history.internal.kafka.topic": "schema-history.shop",
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter"
  }
}

Lưu ý database.server.id phải duy nhất trong cụm MySQL replication (Debezium giả lập như một replica). topic.prefix là tiền tố cho mọi topic do connector này sinh ra. Đăng ký bằng REST API:

# đăng ký connector từ file JSON
curl -X POST -H "Content-Type: application/json" \
  --data @mysql-source.json \
  http://localhost:8083/connectors

Mỗi thay đổi sinh ra một message, phần payload mô tả đầy đủ "trước và sau". Các field quan trọng:

Ví dụ message khi update một bản ghi (op = u):

{
  "payload": {
    "before": { "id": 1, "name": "An", "status": "new" },
    "after":  { "id": 1, "name": "An", "status": "paid" },
    "op": "u",
    "ts_ms": 1719500000000,
    "source": {
      "db": "shop",
      "table": "orders",
      "file": "mysql-bin.000003",
      "pos": 1547
    }
  }
}

Snapshot — khi khởi động lần đầu, Debezium đọc toàn bộ dữ liệu hiện có của các bảng để có "trạng thái nền". Các message này có op = r (read).

Streaming — sau snapshot, connector chuyển sang đọc transaction log liên tục, phát ra các sự kiện c/u/d theo thời gian thực. Vị trí log đã đọc được lưu lại, nên restart không phải snapshot lại từ đầu.

Debezium tạo một topic cho mỗi bảng, theo mẫu:

# mẫu
<topic.prefix>.<database>.<table>

# với cấu hình ở trên
mysqlsrv.shop.orders
mysqlsrv.shop.customers

Nhờ tách topic theo bảng, mỗi sink connector có thể subscribe đúng những topic nó cần. Tên topic này cũng là thứ ta sẽ map sang tên bảng đích ở bài Sink (qua SMT RegexRouter).