Oracle CDC 6 — Debezium + Kafka Connect cho Oracle (LogMiner)

Trong các bài trước của loạt Oracle CDC, chúng ta đã bàn về nguyên lý CDC, cách chuẩn bị Oracle (ARCHIVELOG, supplemental logging, user LogMiner) và các cơ chế đọc redo log. Bài này là một LAB thực hành đầy đủ: dựng một cụm Kafka + Kafka Connect + Debezium bằng Docker, cấu hình Debezium Oracle connector đọc redo qua LogMiner, chạy snapshot khởi tạo, rồi streaming các thay đổi INSERT/UPDATE/DELETE thành message CDC trên Kafka topic.

Toàn bộ tên thuộc tính connector trong bài dựa trên Debezium 2.x (loạt 2.5–2.7). Nếu bạn dùng phiên bản khác (1.x hoặc 3.x), một số tên thuộc tính có thể lệch — chỗ nào cần lưu ý sẽ được ghi chú rõ "kiểm tra tài liệu Debezium phiên bản của bạn".

Kiến trúc tổng quan

Debezium không phải là một tiến trình chạy riêng. Nó là một tập hợp connector chạy bên trong Kafka Connect — một framework runtime của Kafka chuyên để di chuyển dữ liệu vào/ra Kafka. Cụ thể với Oracle:

• Kafka Connect chạy Debezium Oracle connector như một source connector.
• Connector kết nối tới Oracle, dùng gói DBMS_LOGMNR (LogMiner) để đọc redo log (và archived redo log) → lấy ra các thay đổi ở mức hàng (row-level changes).
• Mỗi bảng được theo dõi sẽ ánh xạ tới một Kafka topic riêng. Tên topic mặc định có dạng <topic.prefix>.<SCHEMA>.<TABLE>.
• Mỗi message CDC là một envelope chứa before (trạng thái trước), after (trạng thái sau), op (loại thao tác: c/u/d/r), ts_ms (thời điểm) và metadata source.
• Connector còn ghi lịch sử schema (DDL) vào một topic nội bộ (schema.history.internal.kafka.topic) để có thể diễn giải đúng cấu trúc bảng theo thời gian.

Luồng đọc: Oracle → LogMiner → Debezium/Kafka Connect → Kafka topics → consumers.

Điều kiện tiên quyết trên Oracle

Đây là phần đã trình bày chi tiết ở bài 2 của loạt bài; ở đây nhắc lại những mục bắt buộc phải có trước khi khai báo connector, nếu thiếu connector sẽ báo lỗi ngay khi khởi động.

1. Database ở chế độ ARCHIVELOG. Kiểm tra:

-- lab step: chạy với quyền SYSDBA
SELECT log_mode FROM v$database;
-- kỳ vọng: ARCHIVELOG

2. Supplemental logging. Bật ở cấp database (tối thiểu) và cấp bảng (all columns) cho các bảng theo dõi:

-- lab step: bật minimal supplemental logging cấp database
ALTER DATABASE ADD SUPPLEMENTAL LOG DATA;

-- bật ALL COLUMNS cho từng bảng cần theo dõi
-- (để before-image đầy đủ khi UPDATE/DELETE)
ALTER TABLE INVENTORY.CUSTOMERS ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;
ALTER TABLE INVENTORY.ORDERS    ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;

Nếu không bật ALL COLUMNS, Debezium chỉ có được giá trị các cột PK trong before của UPDATE/DELETE, phần còn lại sẽ là null — thường không đủ dùng.

3. User LogMiner. Trong kiến trúc CDB/PDB (ví dụ Oracle XE với PDB XEPDB1), Debezium khuyến nghị dùng một common user C##DBZUSER với quyền LOGMINING. Việc cấp quyền chi tiết đã có ở bài 2; tóm tắt các quyền cốt lõi:

-- lab step: tạo common user (kết nối tới CDB$ROOT)
CREATE USER C##DBZUSER IDENTIFIED BY dbz DEFAULT TABLESPACE USERS
  QUOTA UNLIMITED ON USERS CONTAINER=ALL;

GRANT CREATE SESSION            TO C##DBZUSER CONTAINER=ALL;
GRANT SET CONTAINER             TO C##DBZUSER CONTAINER=ALL;
GRANT SELECT ON V_$DATABASE     TO C##DBZUSER CONTAINER=ALL;
GRANT FLASHBACK ANY TABLE       TO C##DBZUSER CONTAINER=ALL;
GRANT SELECT ANY TABLE          TO C##DBZUSER CONTAINER=ALL;
GRANT SELECT_CATALOG_ROLE       TO C##DBZUSER CONTAINER=ALL;
GRANT EXECUTE_CATALOG_ROLE      TO C##DBZUSER CONTAINER=ALL;
GRANT SELECT ANY TRANSACTION    TO C##DBZUSER CONTAINER=ALL;
GRANT LOGMINING                 TO C##DBZUSER CONTAINER=ALL;
GRANT EXECUTE ON DBMS_LOGMNR    TO C##DBZUSER CONTAINER=ALL;
GRANT EXECUTE ON DBMS_LOGMNR_D  TO C##DBZUSER CONTAINER=ALL;
GRANT SELECT ON V_$LOG          TO C##DBZUSER CONTAINER=ALL;
GRANT SELECT ON V_$LOG_HISTORY  TO C##DBZUSER CONTAINER=ALL;
GRANT SELECT ON V_$LOGMNR_LOGS  TO C##DBZUSER CONTAINER=ALL;
GRANT SELECT ON V_$LOGMNR_CONTENTS TO C##DBZUSER CONTAINER=ALL;
GRANT SELECT ON V_$ARCHIVED_LOG TO C##DBZUSER CONTAINER=ALL;
GRANT SELECT ON V_$TRANSACTION  TO C##DBZUSER CONTAINER=ALL;

Ghi chú: danh sách quyền chính xác thay đổi theo phiên bản Oracle và Debezium. Luôn đối chiếu mục "Preparing the database" trong tài liệu Debezium Oracle connector của đúng phiên bản bạn dùng.

LAB — Dựng hạ tầng bằng docker-compose

Ta dùng image chính thức của Debezium: debezium/connect đã đóng gói sẵn Debezium Oracle connector. Cụm gồm Zookeeper, Kafka broker, Kafka Connect và một Kafka UI tuỳ chọn.

Về Oracle JDBC driver & Instant Client: image debezium/connect phiên bản gần đây đã kèm sẵn driver Oracle cần thiết cho chế độ LogMiner (không cần XStream/OCI, nên thường không cần Oracle Instant Client). Nếu bạn xây image tự chế hoặc dùng bản cũ, có thể phải thêm ojdbc*.jar vào thư mục plugin của connector, và chỉ khi dùng chế độ XStream mới cần Instant Client. Với LAB này (LogMiner) ta dựa vào image dựng sẵn.

# lab step: docker-compose.yml
services:
  zookeeper:
    image: quay.io/debezium/zookeeper:2.7
    ports:
      - "2181:2181"

  kafka:
    image: quay.io/debezium/kafka:2.7
    ports:
      - "9092:9092"
    depends_on:
      - zookeeper
    environment:
      ZOOKEEPER_CONNECT: zookeeper:2181
      # advertised listener cho client trong docker network + host
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092

  connect:
    image: quay.io/debezium/connect:2.7
    ports:
      - "8083:8083"
    depends_on:
      - kafka
    environment:
      BOOTSTRAP_SERVERS: kafka:9092
      GROUP_ID: cdc-connect-group
      # 3 topic nội bộ Kafka Connect lưu config/offset/status
      CONFIG_STORAGE_TOPIC: connect_configs
      OFFSET_STORAGE_TOPIC: connect_offsets
      STATUS_STORAGE_TOPIC: connect_statuses
      # replication factor = 1 cho môi trường lab 1 broker
      CONFIG_STORAGE_REPLICATION_FACTOR: 1
      OFFSET_STORAGE_REPLICATION_FACTOR: 1
      STATUS_STORAGE_REPLICATION_FACTOR: 1
      KEY_CONVERTER: org.apache.kafka.connect.json.JsonConverter
      VALUE_CONVERTER: org.apache.kafka.connect.json.JsonConverter

  # tuỳ chọn: giao diện xem topic/message
  kafka-ui:
    image: provectuslabs/kafka-ui:latest
    ports:
      - "8080:8080"
    depends_on:
      - kafka
    environment:
      KAFKA_CLUSTERS_0_NAME: local
      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:9092
      KAFKA_CLUSTERS_0_KAFKACONNECT_0_NAME: connect
      KAFKA_CLUSTERS_0_KAFKACONNECT_0_ADDRESS: http://connect:8083

Khởi động cụm và kiểm tra Kafka Connect đã sống:

# lab step: dựng cụm ở chế độ nền
docker compose up -d

# kiểm tra Kafka Connect REST API (mất ~20-30s để sẵn sàng)
curl -s http://localhost:8083/ | jq .

# liệt kê các plugin connector có sẵn — phải thấy OracleConnector
curl -s http://localhost:8083/connector-plugins | jq '.[].class'
# kỳ vọng có: "io.debezium.connector.oracle.OracleConnector"

Nếu connector-plugins không có OracleConnector, image bạn dùng chưa gói Debezium Oracle — hãy dùng đúng image quay.io/debezium/connect (Docker Hub debezium/connect là mirror). Với chế độ KRaft (không Zookeeper), bạn thay service Kafka bằng biến KAFKA_NODE_ID, KAFKA_PROCESS_ROLES=broker,controller, KAFKA_CONTROLLER_QUORUM_VOTERS... — kiểm tra tài liệu image Debezium phiên bản để lấy đúng biến môi trường.

LAB — Đăng ký Oracle connector qua REST API

Kafka Connect quản lý connector qua REST API tại cổng 8083. Ta gửi POST /connectors với body JSON gồm name và config. Dưới đây là cấu hình LogMiner chuẩn cho Debezium 2.x, kèm giải thích từng thuộc tính.

{
  "name": "oracle-inventory-connector",
  "config": {
    "connector.class": "io.debezium.connector.oracle.OracleConnector",
    "tasks.max": "1",

    "database.hostname": "oracle-host",
    "database.port": "1521",
    "database.user": "C##DBZUSER",
    "database.password": "dbz",

    "database.dbname": "XE",
    "database.pdb.name": "XEPDB1",

    "topic.prefix": "server1",

    "schema.history.internal.kafka.bootstrap.servers": "kafka:9092",
    "schema.history.internal.kafka.topic": "schema-history.inventory",

    "table.include.list": "INVENTORY.CUSTOMERS,INVENTORY.ORDERS",

    "snapshot.mode": "initial",
    "log.mining.strategy": "online_catalog",
    "decimal.handling.mode": "double",
    "tombstones.on.delete": "true"
  }
}

Ý nghĩa các thuộc tính chính xác (Debezium 2.x):

• connector.class = io.debezium.connector.oracle.OracleConnector — lớp connector Oracle. Bắt buộc, viết đúng tuyệt đối.
• tasks.max — Oracle connector luôn dùng đúng 1 task (không song song hoá theo bảng). Đặt 1.
• database.hostname / database.port / database.user / database.password — thông tin kết nối JDBC tới Oracle. Với LAB, hostname là tên host/IP Oracle mà container connect truy cập được.
• database.dbname — tên CDB (ví dụ XE), tức tên service của container database gốc.
• database.pdb.name = XEPDB1 — tên PDB chứa bảng nghiệp vụ. Chỉ đặt thuộc tính này khi Oracle là kiến trúc CDB/PDB. Với non-CDB (Oracle 11g/ single-tenant) thì bỏ thuộc tính này. Kiểm tra tài liệu phiên bản nếu dùng non-CDB.
• topic.prefix — tiền tố (logical server name) cho mọi topic connector sinh ra, và cũng là namespace trong metadata source. Trước đây (Debezium 1.x) thuộc tính này tên là database.server.name — nếu bạn dùng 1.x hãy đổi lại; ở 2.x là topic.prefix.
• schema.history.internal.kafka.bootstrap.servers và schema.history.internal.kafka.topic — nơi lưu lịch sử schema/DDL. Debezium 2.x dùng tiền tố schema.history.internal.*; Debezium 1.x dùng database.history.*. Đây là điểm rất hay nhầm khi copy config cũ.
• table.include.list — danh sách bảng theo dõi, dạng SCHEMA.TABLE, ngăn cách bằng dấu phẩy (là regex). Có thể thay bằng schema.include.list để lấy cả schema.
• snapshot.mode = initial — lần đầu chụp toàn bộ dữ liệu hiện có (op=r) rồi chuyển sang streaming. Các giá trị khác gồm no_data / schema_only (chỉ schema, không snapshot dữ liệu — tên có thể lệch theo phiên bản), always, when_needed... Kiểm tra tài liệu phiên bản cho tên chính xác.
• log.mining.strategy = online_catalog — dùng data dictionary đang online (nhanh, ít tải), phù hợp khi schema ít thay đổi trong lúc mining. Giá trị còn lại là redo_log_catalog (ghi dictionary vào redo, chịu được DDL trong khi mining nhưng nặng hơn).
• decimal.handling.mode = double — cách biểu diễn NUMBER/DECIMAL. precise (mặc định, dùng java.math.BigDecimal encode base64), double (số thực, dễ đọc, có thể mất chính xác), string. Chọn theo yêu cầu độ chính xác.
• tombstones.on.delete = true — sau message DELETE (op=d), phát thêm một tombstone (message có key nhưng value = null) để hỗ trợ log compaction. Đặt false nếu không muốn tombstone.

Gửi request tạo connector:

# lab step: đăng ký connector
curl -s -X POST http://localhost:8083/connectors \
  -H "Content-Type: application/json" \
  -d @oracle-connector.json | jq .

# kiểm tra trạng thái — cả connector và task phải RUNNING
curl -s http://localhost:8083/connectors/oracle-inventory-connector/status | jq .

Kết quả trạng thái kỳ vọng:

{
  "name": "oracle-inventory-connector",
  "connector": { "state": "RUNNING", "worker_id": "connect:8083" },
  "tasks": [ { "id": 0, "state": "RUNNING", "worker_id": "connect:8083" } ],
  "type": "source"
}

Nếu task ở trạng thái FAILED, lấy stack trace chi tiết trong tasks[0].trace của cùng endpoint status để biết lỗi cụ thể. Một số REST hữu ích khác:

# lab step: liệt kê connector đang có
curl -s http://localhost:8083/connectors | jq .

# xem config hiện tại
curl -s http://localhost:8083/connectors/oracle-inventory-connector/config | jq .

# restart task nếu bị treo
curl -s -X POST http://localhost:8083/connectors/oracle-inventory-connector/tasks/0/restart

# xoá connector
curl -s -X DELETE http://localhost:8083/connectors/oracle-inventory-connector

LAB — Kiểm thử: sinh thay đổi và đọc message CDC

Sau khi connector RUNNING, snapshot khởi tạo sẽ ghi mỗi hàng hiện có thành một message với op=r (read) lên topic tương ứng. Với bảng INVENTORY.CUSTOMERS và topic.prefix=server1, topic sẽ là server1.INVENTORY.CUSTOMERS.

Mở consumer để theo dõi topic:

# lab step: liệt kê topic — sẽ thấy server1.INVENTORY.CUSTOMERS, ...
docker compose exec kafka \
  /kafka/bin/kafka-topics.sh --bootstrap-server kafka:9092 --list

# consume từ đầu topic, in cả key
docker compose exec kafka \
  /kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server kafka:9092 \
  --topic server1.INVENTORY.CUSTOMERS \
  --from-beginning \
  --property print.key=true

Hoặc mở Kafka UI tại http://localhost:8080 để duyệt topic và message trực quan.

Bây giờ sinh thay đổi trên Oracle (kết nối vào PDB XEPDB1):

-- lab step: chạy trong PDB XEPDB1
INSERT INTO INVENTORY.CUSTOMERS (ID, NAME, EMAIL)
VALUES (1001, 'Nguyen Van A', '[email protected]');
COMMIT;

UPDATE INVENTORY.CUSTOMERS SET EMAIL = '[email protected]' WHERE ID = 1001;
COMMIT;

DELETE FROM INVENTORY.CUSTOMERS WHERE ID = 1001;
COMMIT;

Debezium chỉ phát message sau khi COMMIT (nó theo dõi transaction). Với ba câu lệnh trên, consumer sẽ nhận lần lượt ba message CDC với op = c, u, d.

Cấu trúc message CDC (envelope)

Vì trong docker-compose ta đặt VALUE_CONVERTER=JsonConverter (mặc định kèm schema), mỗi message value gồm hai phần: schema (mô tả kiểu dữ liệu của payload) và payload (dữ liệu thực). Dưới đây là phần payload của một message UPDATE (đã lược schema cho gọn):

{
  "payload": {
    "before": {
      "ID": 1001,
      "NAME": "Nguyen Van A",
      "EMAIL": "[email protected]"
    },
    "after": {
      "ID": 1001,
      "NAME": "Nguyen Van A",
      "EMAIL": "[email protected]"
    },
    "source": {
      "version": "2.7.0.Final",
      "connector": "oracle",
      "name": "server1",
      "ts_ms": 1712345678000,
      "snapshot": "false",
      "db": "XE",
      "schema": "INVENTORY",
      "table": "CUSTOMERS",
      "scn": "1546887",
      "commit_scn": "1546890"
    },
    "op": "u",
    "ts_ms": 1712345678123
  }
}

Diễn giải các trường quan trọng:

• op — loại thao tác: c = create (INSERT), u = update, d = delete, r = read (bản ghi lấy trong lúc snapshot). Đây là trường bạn dùng để phân loại sự kiện ở phía consumer.
• before — trạng thái hàng trước thay đổi. Với INSERT (op=c) và snapshot (op=r) thì before = null. Với UPDATE/DELETE, before chỉ đầy đủ nếu đã bật supplemental logging ALL COLUMNS.
• after — trạng thái hàng sau thay đổi. Với DELETE (op=d) thì after = null.
• source — metadata nguồn: db/schema/table, scn/commit_scn (vị trí trong redo), snapshot (true/false), ts_ms (thời điểm sự kiện ở nguồn).
• ts_ms (ngoài source) — thời điểm Debezium xử lý sự kiện.

Với DELETE và tombstones.on.delete=true, ngay sau message op=d (có before đầy đủ, after=null) sẽ có thêm một message tombstone: cùng key nhưng value = null. Consumer/compaction dùng tombstone để dọn key khỏi topic compacted.

Message key mặc định chứa các cột primary key của bảng, cũng ở dạng {schema, payload}. Nếu bảng không có PK, cần cấu hình message.key.columns để chỉ định khoá — nếu không, key sẽ là null và ảnh hưởng tới phân vùng/compaction.

Xử lý lỗi thường gặp và mẹo

ORA-01031: insufficient privileges (hoặc task FAILED do quyền). Thường do user LogMiner thiếu một trong các quyền LOGMINING, SELECT ANY TRANSACTION, EXECUTE ON DBMS_LOGMNR, hoặc quyền SELECT trên các view V_$.... Trong CDB, quyền phải được cấp với CONTAINER=ALL và user phải là common user (C##). Đọc tasks[0].trace để biết chính xác quyền/đối tượng nào bị thiếu, rồi cấp bổ sung.

Thiếu supplemental logging. Triệu chứng: UPDATE/DELETE có before gần như rỗng (chỉ có PK), hoặc log connector cảnh báo thiếu supplemental log. Khắc phục: bật ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS cho từng bảng theo dõi (như phần tiên quyết). Lưu ý các thay đổi phát sinh trước khi bật supplemental logging sẽ không có before-image đầy đủ.

LogMiner chậm / độ trễ cao (lag). Nguyên nhân hay gặp: khối lượng redo lớn, khoảng SCN mining quá rộng, dùng redo_log_catalog khi không cần. Mẹo:

• Ưu tiên log.mining.strategy=online_catalog khi schema ổn định.
• Tinh chỉnh kích thước bước mining qua các thuộc tính log.mining.batch.size.min / log.mining.batch.size.max và log.mining.sleep.time.* (tên có thể lệch theo phiên bản — kiểm tra tài liệu Debezium phiên bản).
• Giữ archived redo log được lưu trữ đủ lâu; nếu Oracle xoá archive log trước khi Debezium đọc kịp (ví dụ RMAN xoá), connector sẽ lỗi không tìm thấy log cho SCN cần thiết.
• Chỉ theo dõi đúng bảng cần thiết bằng table.include.list để giảm khối lượng mining.

Snapshot lớn / lâu. Với bảng hàng chục triệu hàng, snapshot initial có thể chạy rất lâu và giữ khoá đọc nhất quán. Mẹo:

• Cân nhắc snapshot.mode chỉ lấy schema (không dữ liệu) rồi backfill bằng phương án khác, nếu bạn chỉ cần thay đổi từ thời điểm hiện tại (tên giá trị mode kiểm tra theo phiên bản).
• Dùng incremental snapshot (signal table) của Debezium để chụp lại dần từng chunk mà không dừng streaming — tính năng mạnh khi cần re-snapshot một bảng lớn giữa chừng.
• Đảm bảo topic nội bộ Kafka Connect (connect_offsets) không bị mất, vì offset ở đó quyết định connector tiếp tục từ SCN nào sau khi restart.

Topic không xuất hiện / consumer không thấy message. Kiểm tra: connector RUNNING chưa; đã có COMMIT trên Oracle chưa; đúng tên topic (<topic.prefix>.<SCHEMA>.<TABLE>, chú ý chữ HOA của schema/table trong Oracle); và advertised.listeners của Kafka trỏ tới địa chỉ mà consumer truy cập được (lỗi kinh điển khi consume từ host bên ngoài docker network).

Chuyển đổi/định dạng message. Nếu message JSON quá dài do kèm schema trong mỗi record, bạn có thể tắt schema bằng value.converter.schemas.enable=false, hoặc chuyển sang Avro + Schema Registry để gọn và có kiểm soát schema — chủ đề của bài kế tiếp.

Tóm tắt

• Debezium chạy như một source connector trong Kafka Connect, đọc redo của Oracle qua LogMiner, mỗi bảng → một Kafka topic.
• Hạ tầng LAB dựng nhanh bằng docker-compose với image quay.io/debezium/connect (đã kèm Oracle connector cho chế độ LogMiner, thường không cần Instant Client).
• Trước khi chạy phải có: ARCHIVELOG, supplemental logging ALL COLUMNS, và user LogMiner (C##DBZUSER) với quyền LOGMINING cùng các quyền catalog.
• Connector đăng ký qua REST API với các thuộc tính chính xác của Debezium 2.x: connector.class, database.*, database.pdb.name, topic.prefix, schema.history.internal.kafka.*, table.include.list, snapshot.mode, log.mining.strategy, decimal.handling.mode, tombstones.on.delete.
• Message CDC là envelope {schema, payload} với before/after/op (c/u/d/r)/ts_ms/source; DELETE có thể kèm tombstone.
• Lỗi hay gặp xoay quanh quyền, supplemental logging, độ trễ LogMiner và snapshot lớn — hầu hết chẩn đoán được qua tasks[].trace trong endpoint status.

Tự kiểm tra

1. Vì sao Debezium Oracle connector luôn chỉ dùng tasks.max=1, và điều đó ảnh hưởng gì tới khả năng mở rộng?
2. Sự khác nhau giữa topic.prefix (Debezium 2.x) và database.server.name (1.x) là gì, và nhóm thuộc tính lưu lịch sử schema đổi tên như thế nào giữa hai phiên bản?
3. Nếu UPDATE cho ra before chỉ có cột PK còn các cột khác null, nguyên nhân gốc là gì và khắc phục ra sao?
4. Giải thích ý nghĩa từng giá trị của op (c/u/d/r) và trong trường hợp nào before hoặc after bằng null.
5. Khi nào nên chọn log.mining.strategy=online_catalog thay vì redo_log_catalog?
6. Vai trò của tombstone khi tombstones.on.delete=true là gì, và nó liên quan thế nào tới log compaction?

Đọc tiếp

Oracle CDC 7 — Kafka Connect & Schema Registry