Bài 3: Hướng dẫn cấu hình Debezium MySQL Connector và theo dõi dữ liệu thay đổi trên Kafka.

Ở Bài 2 chúng ta đã dựng xong "cơ sở hạ tầng" nhưng chúng mới chỉ đứng nhìn nhau chứ chưa làm việc. Hôm nay, chúng ta sẽ bắt tay vào việc quan trọng nhất: Cấu hình cho "chiếc camera" Debezium kết nối vào MySQL và tận mắt chứng kiến dữ liệu nhảy múa trên Kafka nhé.

1. Bản chất của việc "Cấu hình Debezium"

Debezium chạy như một plugin bên trong Kafka Connect. Để bảo Kafka Connect kích hoạt Debezium theo dõi một Database cụ thể, chúng ta không cần viết code mà chỉ cần gửi một đoạn cấu hình bằng định dạng JSON qua API (gọi là REST API) của Kafka Connect (chạy ở cổng 8083 mà ta đã mở ở Bài 2).

Nhưng trước khi theo dõi, ta cần có dữ liệu mẫu đã!

2. Thực hành: Cấu hình và theo dõi dữ liệu

Bước 1: Chuẩn bị dữ liệu trên MySQL (Tạo DB và bảng dữ liệu mẫu)

Mở Terminal và chạy lệnh sau để thọc thẳng vào container MySQL, tạo một database tên là inventory và bảng users:

docker-compose exec mysql bash -c 'mysql -u root -pdebezium -e "
CREATE DATABASE inventory;
USE inventory;
CREATE TABLE users (id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(255), email VARCHAR(255));
INSERT INTO users (name, email) VALUES (''Hoang'', ''hoang@example.com'');
"'

Giải thích: Ta vừa tạo một DB, một bảng users và chèn sẵn 1 dòng dữ liệu đầu tiên.

Bước 2: Đăng ký MySQL Connector (Kích hoạt Debezium)

Chúng ta sẽ dùng lệnh curl để ném đoạn cấu hình JSON vào Kafka Connect. Đứng tại Terminal của máy bạn, copy và chạy toàn bộ đoạn lệnh sau:

curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" localhost:8083/connectors/ -d '{
  "name": "inventory-connector",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "tasks.max": "1",
    "database.hostname": "mysql",
    "database.port": "3306",
    "database.user": "root",
    "database.password": "debezium",
    "database.server.id": "184054",
    "topic.prefix": "dbserver1",
    "database.include.list": "inventory",
    "schema.history.internal.kafka.bootstrap.servers": "kafka:9092",
    "schema.history.internal.kafka.topic": "schema-changes.inventory"
  }
}'

Ghi chú quan trọng:

• database.include.list: Chỉ báo cho Debezium theo dõi DB inventory.
• topic.prefix: Là tiền tố cho tên Topic trên Kafka (ở đây là dbserver1).

Bước 3: Kiểm tra trạng thái Connector

Để chắc chắn Connector đang chạy ngon lành, hãy chạy lệnh:

curl -H "Accept:application/json" localhost:8083/connectors/inventory-connector/status

Nếu bạn thấy "state":"RUNNING", chúc mừng! Camera đã bật và đang chĩa vào MySQL.

Bước 4: Mở Consumer xem Kafka (Lắng nghe dữ liệu theo thời gian thực)

Debezium sẽ tự động đẩy các thay đổi của bảng users vào một Topic có tên là: dbserver1.inventory.users.

Hãy mở một Tab Terminal MỚI (để treo đó xem log) và chạy lệnh sau để đọc Kafka:

docker-compose exec kafka /kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server kafka:9092 \
  --from-beginning \
  --property print.key=true \
  --topic dbserver1.inventory.users

Lúc này, bạn sẽ thấy nó in ra một nùi JSON cực dài. Đó chính là sự kiện cho dòng Hoang đã có sẵn trong DB!

Bước 5: Tạo thay đổi dữ liệu để test Real-time (Thời khắc của sự thật)

Quay lại Tab Terminal cũ, hãy thử Insert thêm 1 user mới vào MySQL xem tab Kafka bên kia có phản ứng ngay lập tức không nhé:

docker-compose exec mysql bash -c 'mysql -u root -pdebezium -e "
USE inventory;
INSERT INTO users (name, email) VALUES (''Debezium Fan'', ''fan@debezium.io'');
"'

Ngay khoảnh khắc bạn bấm Enter, bên tab Terminal của Kafka sẽ ngay lập tức "bắn" ra một đoạn JSON mới! Tốc độ bàn thờ!

3. Đọc hiểu Message của Debezium

Message bắn ra trên Kafka là một chuỗi JSON rất dài (bao gồm cả schema), nhưng trái tim của nó nằm ở cục payload. Hãy nhìn vào cấu trúc siêu thu gọn này để hiểu nó báo gì cho bạn:

"payload": {
  "before": null,
  "after": {
    "id": 2,
    "name": "Debezium Fan",
    "email": "fan@debezium.io"
  },
  "source": { ... },
  "op": "c",
  "ts_ms": 1690000000000
}

Giải mã 3 trường quan trọng nhất:

• op (Operation): Loại hành động. c là Create (Insert), u là Update, d là Delete, r là Read (đọc lần đầu tiên lúc snapshot DB).
• before: Dữ liệu CŨ trước khi thay đổi. (Vì đây là lệnh Insert nên trước đó không có gì, nó bằng null).
• after: Dữ liệu MỚI sau khi thay đổi.

Sức mạnh thực sự: Nếu bạn chạy lệnh UPDATE users SET name = 'Vip' WHERE id = 1, Debezium sẽ gửi một message với op: "u", trường before chứa chữ Hoang, trường after chứa chữ Vip. Nhờ thế, các hệ thống khác đọc Kafka sẽ biết chính xác cái gì vừa bị thay đổi để cập nhật theo!