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

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:

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

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"

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

-- 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;

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

-- 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;

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;

{
  "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"
  }
}

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 .

{
  "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"
  }
}

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 .

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

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;

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

# 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'

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"