Oracle CDC 9 — Lab tổng hợp: Oracle → Debezium → Kafka → Sink

Mục tiêu của lab

Trong bài này chúng ta dựng một pipeline CDC (Change Data Capture) hoàn chỉnh, chạy được từ đầu đến cuối trên máy cá nhân bằng docker compose. Luồng dữ liệu:

Oracle (nguồn) → Debezium Oracle Connector (LogMiner) → Kafka → JDBC Sink Connector → PostgreSQL (đích).

Sau khi hoàn thành, mỗi thao tác INSERT / UPDATE / DELETE trên một bảng Oracle sẽ được bắt lại gần thời gian thực, chảy qua một topic Kafka, và được ghi (upsert/xóa) tương ứng vào một bảng trong PostgreSQL — không cần viết dòng code ứng dụng nào.

Bài này tổng hợp ba bài trước trong series:

• Oracle CDC 2 — Chuẩn bị Oracle cho CDC (ARCHIVELOG, supplemental logging, user CDC).
• Oracle CDC 6 — Debezium Oracle Connector (cấu hình source connector).
• Oracle CDC 7 — Đẩy dữ liệu xuống đích với JDBC Sink (sink + SMT unwrap).

Cảnh báo: Đây là môi trường LAB để học và thử nghiệm. Cấu hình dưới đây ưu tiên sự đơn giản (mật khẩu hard-code, không TLS, một broker duy nhất, PLAINTEXT). Tuyệt đối không dùng nguyên trạng cho production. Các khối lệnh Oracle / Kafka / shell / JSON bên dưới là lệnh lab — bạn tự chạy trong môi trường Docker của mình.

Kiến trúc & yêu cầu

Sơ đồ tổng thể

Yêu cầu tài nguyên

RAM khuyến nghị: tối thiểu 8 GB rảnh cho Docker (Oracle XE ăn ~2 GB, Kafka + Connect + Postgres + UI ~2–3 GB). Với máy 16 GB tổng sẽ chạy thoải mái.

Về phiên bản: Debezium 2.6 tương thích Oracle 19c/21c XE và Kafka Connect 3.x. Nếu bạn dùng Debezium 2.7+ thì các thuộc tính connector giữ nguyên tên như bài này. Trình điều khiển Oracle JDBC (ojdbc) đã có sẵn trong image debezium/connect, nhưng driver JDBC sink của Confluent và driver PostgreSQL cần được thêm vào (xem Bước 1).

BƯỚC 1 — docker-compose.yml

Tạo một thư mục lab, ví dụ ~/oracle-cdc-lab, và đặt file docker-compose.yml sau. Kafka chạy chế độ KRaft (không cần ZooKeeper) cho gọn.

services:
  oracle:
    image: gvenzl/oracle-xe:21-slim
    container_name: oracle
    ports:
      - "1521:1521"
    environment:
      ORACLE_PASSWORD: oracle          # mật khẩu cho SYS/SYSTEM (LAB)
      APP_USER: appuser                # user thường (không phải user CDC)
      APP_USER_PASSWORD: apppw
    healthcheck:
      test: ["CMD", "healthcheck.sh"]
      interval: 15s
      timeout: 10s
      retries: 20
    volumes:
      - oracle-data:/opt/oracle/oradata

  kafka:
    image: confluentinc/cp-kafka:7.6.1
    container_name: kafka
    ports:
      - "9092:9092"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: "broker,controller"
      KAFKA_CONTROLLER_QUORUM_VOTERS: "1@kafka:29093"
      KAFKA_LISTENERS: "PLAINTEXT://0.0.0.0:29092,CONTROLLER://0.0.0.0:29093,PLAINTEXT_HOST://0.0.0.0:9092"
      KAFKA_ADVERTISED_LISTENERS: "PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092"
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: "PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT"
      KAFKA_CONTROLLER_LISTENER_NAMES: "CONTROLLER"
      KAFKA_INTER_BROKER_LISTENER_NAME: "PLAINTEXT"
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      CLUSTER_ID: "MkU3OEVBNTcwNTJENDM2Qk"   # cluster id cố định cho lab

  connect:
    image: debezium/connect:2.6
    container_name: connect
    depends_on:
      - kafka
      - oracle
      - postgres
    ports:
      - "8083:8083"
    environment:
      BOOTSTRAP_SERVERS: "kafka:29092"
      GROUP_ID: "cdc-connect"
      CONFIG_STORAGE_TOPIC: "connect_configs"
      OFFSET_STORAGE_TOPIC: "connect_offsets"
      STATUS_STORAGE_TOPIC: "connect_statuses"
      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"
      # Thư mục plugin bổ sung (JDBC sink + driver Postgres)
      CONNECT_PLUGIN_PATH: "/kafka/connect"
    volumes:
      # Đặt sẵn confluentinc-kafka-connect-jdbc + postgresql-*.jar vào đây
      - ./connect-plugins:/kafka/connect/jdbc-sink

  postgres:
    image: postgres:16
    container_name: postgres
    ports:
      - "5433:5432"
    environment:
      POSTGRES_USER: sink
      POSTGRES_PASSWORD: sinkpw
      POSTGRES_DB: sinkdb
    volumes:
      - pg-data:/var/lib/postgresql/data

  # Tùy chọn: giao diện quan sát
  kafka-ui:
    image: provectuslabs/kafka-ui:latest
    container_name: kafka-ui
    depends_on:
      - kafka
      - connect
    ports:
      - "8080:8080"
    environment:
      KAFKA_CLUSTERS_0_NAME: "lab"
      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: "kafka:29092"
      KAFKA_CLUSTERS_0_KAFKACONNECT_0_NAME: "connect"
      KAFKA_CLUSTERS_0_KAFKACONNECT_0_ADDRESS: "http://connect:8083"

volumes:
  oracle-data:
  pg-data:

Chuẩn bị plugin JDBC Sink

Image debezium/connect không kèm sẵn JDBC Sink của Confluent. Tải và giải nén vào ./connect-plugins:

mkdir -p ~/oracle-cdc-lab/connect-plugins
cd ~/oracle-cdc-lab/connect-plugins

# 1) Confluent JDBC connector (chứa cả source lẫn sink)
#    Tải bản .zip từ Confluent Hub rồi giải nén, ví dụ:
#    confluentinc-kafka-connect-jdbc-10.7.x
#    Kết quả: có thư mục con "lib/" chứa các .jar

# 2) Driver PostgreSQL — copy vào cùng thư mục "lib/" của connector JDBC
#    ví dụ postgresql-42.7.3.jar

Lưu ý phiên bản: JDBC Sink của Debezium (io.debezium.connector.jdbc.JdbcSinkConnector, có sẵn trong image từ Debezium 2.x) là một lựa chọn thay thế cho connector của Confluent. Ở lab này ta dùng Confluent JDBC Sink (io.confluent.connect.jdbc.JdbcSinkConnector) vì tài liệu phổ biến hơn; nếu bạn muốn dùng bản Debezium thì tên connector.class và một vài thuộc tính (primary.key.mode, insert.mode) hơi khác — sẽ ghi chú ở Bước 4.

Khởi động

cd ~/oracle-cdc-lab
docker compose up -d
docker compose ps          # chờ oracle "healthy"
docker compose logs -f oracle | grep -m1 "DATABASE IS READY TO USE"

Oracle XE lần đầu mất vài phút để khởi tạo. Chờ đến khi thấy dòng DATABASE IS READY TO USE!.

BƯỚC 2 — Khởi tạo Oracle cho CDC

Phần này nhắc lại ngắn gọn; chi tiết đầy đủ xem Oracle CDC 2. Kết nối với vai trò SYSDBA:

docker exec -it oracle sqlplus sys/oracle@localhost:1521/XE as sysdba

2.1 Bật ARCHIVELOG (nếu chưa)

-- Kiểm tra trạng thái hiện tại
ARCHIVE LOG LIST;

-- Bật ARCHIVELOG (cần restart instance)
SHUTDOWN IMMEDIATE;
STARTUP MOUNT;
ALTER DATABASE ARCHIVELOG;
ALTER DATABASE OPEN;

Với gvenzl/oracle-xe, một số bản đã bật ARCHIVELOG sẵn. Nếu ARCHIVE LOG LIST báo Archive Mode thì bỏ qua bước trên.

2.2 Supplemental logging ở cấp database

ALTER DATABASE ADD SUPPLEMENTAL LOG DATA;
ALTER DATABASE ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;

2.3 Tạo user CDC C##DBZUSER (common user) + quyền

Debezium LogMiner cần một common user (C##) có quyền trên cả CDB lẫn PDB:

-- Làm việc ở CDB$ROOT
ALTER SESSION SET CONTAINER = 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 CREATE TABLE                   TO c##dbzuser CONTAINER=ALL;
GRANT LOCK ANY TABLE                 TO c##dbzuser CONTAINER=ALL;
GRANT CREATE SEQUENCE                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_$LOGFILE           TO c##dbzuser CONTAINER=ALL;
GRANT SELECT ON V_$ARCHIVED_LOG      TO c##dbzuser CONTAINER=ALL;
GRANT SELECT ON V_$ARCHIVE_DEST_STATUS TO c##dbzuser CONTAINER=ALL;

2.4 Tạo bảng demo trong PDB XEPDB1

ALTER SESSION SET CONTAINER = XEPDB1;

CREATE USER inventory IDENTIFIED BY invpw
  DEFAULT TABLESPACE users QUOTA UNLIMITED ON users;
GRANT CREATE SESSION, CREATE TABLE TO inventory;

CREATE TABLE inventory.customers (
  id        NUMBER(9)     PRIMARY KEY,
  name      VARCHAR2(100) NOT NULL,
  email     VARCHAR2(150),
  balance   NUMBER(12,2)  DEFAULT 0,
  updated_at TIMESTAMP    DEFAULT SYSTIMESTAMP
);

-- Supplemental logging ở cấp bảng (bắt buộc để có đủ cột cũ/mới)
ALTER TABLE inventory.customers ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;

INSERT INTO inventory.customers (id, name, email, balance)
  VALUES (1, 'An Nguyen', '[email protected]', 100.00);
INSERT INTO inventory.customers (id, name, email, balance)
  VALUES (2, 'Binh Tran', '[email protected]', 250.50);
COMMIT;

BƯỚC 3 — Đăng ký Debezium Oracle Source Connector

Kafka Connect REST API ở http://localhost:8083. Lưu cấu hình source vào oracle-source.json:

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

    "database.hostname": "oracle",
    "database.port": "1521",
    "database.user": "c##dbzuser",
    "database.password": "dbz",
    "database.dbname": "XE",
    "database.pdb.name": "XEPDB1",

    "topic.prefix": "ora",
    "table.include.list": "INVENTORY.CUSTOMERS",

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

    "log.mining.strategy": "online_catalog",
    "snapshot.mode": "initial",

    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "key.converter.schemas.enable": "true",
    "value.converter.schemas.enable": "true",

    "decimal.handling.mode": "double",
    "time.precision.mode": "connect"
  }
}

Đăng ký:

curl -s -X POST http://localhost:8083/connectors \
  -H "Content-Type: application/json" \
  -d @oracle-source.json | jq .

# Kiểm tra trạng thái
curl -s http://localhost:8083/connectors/oracle-inventory-source/status | jq .

Trạng thái mong đợi: connector.state = RUNNING và tasks[0].state = RUNNING.

Giải thích các thuộc tính chính (giống bài 6):

• connector.class: io.debezium.connector.oracle.OracleConnector.
• database.dbname = tên CDB (XE), database.pdb.name = PDB chứa bảng (XEPDB1). Kết nối CDC luôn qua common user c##dbzuser.
• topic.prefix: tiền tố tên topic + tên server logic. Topic dữ liệu sẽ là ora.INVENTORY.CUSTOMERS.
• table.include.list: dạng SCHEMA.TABLE (viết hoa theo mặc định Oracle).
• schema.history.internal.*: topic nội bộ lưu lịch sử DDL — bắt buộc cho Oracle.
• log.mining.strategy = online_catalog: đơn giản, phù hợp lab (không ghi catalog vào redo).
• decimal.handling.mode = double + time.precision.mode = connect: giúp kiểu số/thời gian được biểu diễn "phẳng" để JDBC sink map dễ dàng.

Version note: Từ Debezium 2.x, các khóa lịch sử schema dùng tiền tố schema.history.internal. (trước 1.9 là database.history.). Nếu bạn dùng bản cũ hơn, đổi tiền tố tương ứng.

BƯỚC 4 — Đăng ký JDBC Sink Connector (ghi vào Postgres)

Topic Debezium phát ra bản ghi có envelope (before, after, op, source...). JDBC Sink cần bản ghi phẳng, nên ta dùng SMT ExtractNewRecordState (unwrap) để lấy phần after và xử lý xóa. Lưu pg-sink.json:

{
  "name": "pg-customers-sink",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "tasks.max": "1",
    "topics": "ora.INVENTORY.CUSTOMERS",

    "connection.url": "jdbc:postgresql://postgres:5432/sinkdb",
    "connection.username": "sink",
    "connection.password": "sinkpw",

    "insert.mode": "upsert",
    "pk.mode": "record_key",
    "pk.fields": "ID",
    "delete.enabled": "true",

    "auto.create": "true",
    "auto.evolve": "true",

    "table.name.format": "customers",

    "transforms": "unwrap",
    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
    "transforms.unwrap.drop.tombstones": "false",
    "transforms.unwrap.delete.handling.mode": "none",

    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "key.converter.schemas.enable": "true",
    "value.converter.schemas.enable": "true"
  }
}

Đăng ký:

curl -s -X POST http://localhost:8083/connectors \
  -H "Content-Type: application/json" \
  -d @pg-sink.json | jq .

curl -s http://localhost:8083/connectors/pg-customers-sink/status | jq .

Giải thích các thuộc tính chính (giống bài 7):

• insert.mode = upsert: dòng đã tồn tại thì update, chưa có thì insert — đúng ngữ nghĩa CDC.
• pk.mode = record_key + pk.fields = ID: khóa chính lấy từ key của message (chính là PK của bảng Oracle). Đây là điều kiện để upsert/delete hoạt động.
• delete.enabled = true: khi Debezium phát tombstone (op=d), sink sẽ xóa dòng tương ứng ở Postgres.
• auto.create / auto.evolve: tự tạo bảng customers và tự thêm cột khi schema thay đổi (tiện cho lab; production nên tắt và quản lý DDL thủ công).
• transforms.unwrap = ExtractNewRecordState: gỡ envelope, lấy trạng thái mới. Để delete.handling.mode = none và drop.tombstones = false thì tombstone được giữ lại cho delete.enabled xử lý.

Nếu dùng Debezium JDBC Sink (io.debezium.connector.jdbc.JdbcSinkConnector), một số khóa đổi tên: primary.key.mode=record_key, primary.key.fields=ID, insert.mode=upsert, và không cần SMT unwrap (nó tự hiểu envelope Debezium). Chọn một trong hai, đừng trộn.

BƯỚC 5 — Kiểm thử INSERT / UPDATE / DELETE

5.1 Mở một consumer để xem topic

Terminal 1 — theo dõi topic dữ liệu:

docker exec -it kafka kafka-console-consumer \
  --bootstrap-server kafka:29092 \
  --topic ora.INVENTORY.CUSTOMERS \
  --from-beginning

Ban đầu bạn sẽ thấy 2 message snapshot (id=1, id=2) với "op":"r" (read).

5.2 Sinh thay đổi trên Oracle

Terminal 2 — kết nối PDB rồi thực hiện thao tác:

docker exec -it oracle sqlplus inventory/invpw@localhost:1521/XEPDB1

-- INSERT
INSERT INTO customers (id, name, email, balance)
  VALUES (3, 'Chi Le', '[email protected]', 42.00);
COMMIT;

-- UPDATE
UPDATE customers SET balance = 999.99 WHERE id = 1;
COMMIT;

-- DELETE
DELETE FROM customers WHERE id = 2;
COMMIT;

Ở Terminal 1, bạn sẽ thấy lần lượt các message với op = c (create), u (update), d (delete), kèm một message tombstone (value = null) sau lệnh delete.

5.3 Kiểm tra bảng đích ở Postgres

Terminal 3:

docker exec -it postgres psql -U sink -d sinkdb -c \
  "SELECT id, name, email, balance FROM customers ORDER BY id;"

5.4 Bảng checklist kỳ vọng

Nếu bảng cuối cùng khớp (id=1 và id=3, không còn id=2) thì pipeline CDC end-to-end đã chạy đúng.

BƯỚC 6 — Độ trễ & lỗi thường gặp

Quan sát độ trễ

Độ trễ trong lab thường 1–5 giây cho mỗi thay đổi (phụ thuộc chu kỳ LogMiner quét redo). Bạn có thể ước lượng bằng cách so source.ts_ms (thời điểm Oracle) với ts_ms (thời điểm Debezium phát) trong message. Trên máy chậm, snapshot ban đầu và lần LogMiner khởi động đầu tiên có thể mất lâu hơn — hãy kiên nhẫn.

Lỗi thường gặp

Lệnh chẩn đoán hữu ích:

# Liệt kê connector class mà Connect nạp được
curl -s http://localhost:8083/connector-plugins | jq '.[].class'

# Xem log Connect
docker compose logs -f connect | tail -100

# Xem chi tiết lỗi task
curl -s http://localhost:8083/connectors/oracle-inventory-source/status | jq '.tasks[].trace'

BƯỚC 7 — Dọn dẹp

Khi kết thúc lab, hạ toàn bộ stack kèm volume để giải phóng ổ đĩa (Oracle XE chiếm khá nhiều):

cd ~/oracle-cdc-lab

# Xóa các connector (tùy chọn, vì down -v sẽ xóa cả topic cấu hình)
curl -s -X DELETE http://localhost:8083/connectors/pg-customers-sink
curl -s -X DELETE http://localhost:8083/connectors/oracle-inventory-source

# Hạ stack + xóa volume dữ liệu
docker compose down -v

# Kiểm tra không còn container/volume của lab
docker ps -a | grep -E "oracle|kafka|connect|postgres" || echo "Da don sach"
docker volume ls | grep -E "oracle-data|pg-data" || echo "Volume da xoa"

docker compose down -v xóa vĩnh viễn dữ liệu trong oracle-data và pg-data. Chỉ chạy khi bạn chắc chắn không cần giữ lại.

Tóm tắt

• Ta đã dựng một pipeline CDC hoàn chỉnh bằng docker compose: Oracle XE → Debezium (LogMiner) → Kafka → JDBC Sink → PostgreSQL, chạy near-real-time mà không viết code ứng dụng.
• Oracle cần: ARCHIVELOG, supplemental logging (DB + bảng), và một common user C##DBZUSER với đầy đủ quyền LogMiner.
• Source connector dùng io.debezium.connector.oracle.OracleConnector, khai báo database.dbname (CDB) + database.pdb.name (PDB), topic.prefix, table.include.list, và topic lịch sử schema schema.history.internal.*.
• Sink connector dùng SMT ExtractNewRecordState để làm phẳng, insert.mode=upsert, pk.mode=record_key, delete.enabled=true, cùng auto.create/auto.evolve cho lab.
• Kiểm thử bằng INSERT/UPDATE/DELETE trên Oracle rồi đối chiếu topic (kafka-console-consumer) và bảng đích Postgres theo bảng checklist.
• Nhiều chi tiết phụ thuộc phiên bản (Debezium 2.x đổi tiền tố schema.history.internal.; driver JDBC sink và driver Postgres phải tự nạp) — luôn kiểm tra version trước khi copy nguyên xi.

Tự kiểm tra

1. Vì sao Debezium Oracle bắt buộc dùng common user (C##) thay vì user thường trong PDB?
2. topic.prefix ảnh hưởng đến tên topic dữ liệu như thế nào, và topic của bảng INVENTORY.CUSTOMERS sẽ tên gì?
3. SMT ExtractNewRecordState giải quyết vấn đề gì giữa envelope Debezium và JDBC Sink?
4. Cần những thuộc tính nào ở sink để một lệnh DELETE trên Oracle thực sự xóa được dòng ở Postgres?
5. Nếu source RUNNING nhưng consumer không thấy message sau khi bạn UPDATE, hai nguyên nhân đầu tiên bạn kiểm tra là gì?
6. Lệnh docker compose down -v khác gì docker compose down, và vì sao cần cẩn thận với -v?

Đọc tiếp

• Quay lại tổng quan series: Oracle CDC 1 — Tổng quan CDC.
• Ôn lại từng mảnh của lab: Chuẩn bị Oracle (bài 2), Debezium Oracle Connector (bài 6), JDBC Sink (bài 7).
• Khám phá thêm các chủ đề khác trong mục Data Engineering để mở rộng kiến trúc streaming và data pipeline của bạn.

Use case thực tế

Bối cảnh (minh hoạ tại NCB). Đội Core Banking giữ bảng số dư và hồ sơ khách hàng trên Oracle, nhưng đội Phân tích cần một bản sao gần thời gian thực trong PostgreSQL để chạy báo cáo và dashboard mà không truy vấn trực tiếp lên hệ thống lõi (tránh tăng tải giờ cao điểm). Trước đây họ dùng batch ETL chạy mỗi đêm — dữ liệu trễ tới ~24 giờ, và một job SELECT lớn thỉnh thoảng đụng vào cửa sổ đóng sổ cuối ngày.

Áp dụng lab này. Nhóm dựng đúng pipeline Oracle → Debezium (LogMiner) → Kafka → JDBC Sink → PostgreSQL như bài mô tả, mở rộng table.include.list cho vài bảng nghiệp vụ trọng yếu. Vì Debezium đọc redo log thay vì quét bảng, tải sinh thêm trên Oracle rất nhỏ so với batch SELECT.

Kết quả (ước lượng minh hoạ). Độ trễ đồng bộ từ ~24 giờ giảm còn 1–5 giây mỗi thay đổi (đúng dải độ trễ quan sát ở Bước 6). Báo cáo số dư cuối ngày không còn phải chờ job đêm; đội phân tích đọc từ bản sao Postgres nên hệ thống lõi được giảm tải truy vấn ad-hoc. Các lệnh DELETE/UPDATE trên Oracle cũng được phản chiếu chính xác nhờ delete.enabled=true và insert.mode=upsert, tránh tình trạng dữ liệu "mồ côi" ở đích như cách sao chép chỉ-append trước đây.

Lưu ý: đây là lab học tập; khi lên môi trường thật cần bổ sung TLS, xác thực, nhiều broker, tắt auto.create/auto.evolve và quản lý DDL đích thủ công.

Nguồn tham khảo

• Debezium Documentation — Oracle Connector — LogMiner, supplemental logging, schema.history.internal.*, common user C##.
• Debezium Documentation — JDBC Sink Connector — primary.key.mode, insert.mode, xử lý envelope không cần SMT unwrap.
• Debezium Documentation — Transformations: New Record State Extraction (ExtractNewRecordState / unwrap SMT).
• Apache Kafka Documentation — Kafka Connect — kiến trúc connector/task, converter, REST API.
• Confluent Documentation — JDBC Sink Connector (io.confluent.connect.jdbc.JdbcSinkConnector): pk.mode, delete.enabled, auto.create/auto.evolve.
• Oracle Database Utilities Guide — Using LogMiner to Analyze Redo Log Files.
• Oracle Database Administrator's Guide — Managing Archived Redo Log Files (ARCHIVELOG mode) và Supplemental Logging.