Oracle CDC 5 — GoldenGate for Big Data → Kafka

Ở bài 4, chúng ta đã dựng một Extract chạy trên hệ Oracle để đọc redo log và ghi ra trail file. Trail đó là nguồn dữ liệu thay đổi (CDC) chuẩn của GoldenGate. Bài này đi tiếp phần "đầu ra": làm sao đưa những thay đổi đó vào Apache Kafka để nhiều hệ tiêu thụ song song — kho phân tích, cảnh báo real-time, đồng bộ sang microservice, v.v.

Công cụ để làm việc này là Oracle GoldenGate for Big Data (GG BD). Đây là một sản phẩm riêng, khác với GoldenGate "cổ điển" (cho RDBMS). Trong bài, mọi lệnh và file cấu hình được ghi ở dạng lab step — bạn chép, chỉnh cho môi trường của mình rồi chạy, chứ không phải bấm-là-chạy.

Ghi chú về độ chính xác: tên tham số GG BD thay đổi ít nhưng có khác nhau giữa các phiên bản (12.3, 19c, 21c/23ai). Bài này viết theo quy ước phổ biến của dòng 19c/21c. Ở những chỗ tên tham số có thể khác theo phiên bản, mình sẽ ghi rõ.

GoldenGate for Big Data là gì

GoldenGate cổ điển có hai vai trò tiến trình: Extract (đọc thay đổi từ nguồn) và Replicat (áp thay đổi vào đích). Với đích là một cơ sở dữ liệu quan hệ, Replicat sinh câu SQL và INSERT/UPDATE/DELETE.

GG BD thay thế phần Replicat đó bằng một Replicat đặc biệt không sinh SQL, mà gọi vào một HANDLER (trình xử lý) được viết bằng Java. Handler nhận từng thao tác (operation) đọc từ trail rồi đẩy sang một đích phi-RDBMS: Kafka, HDFS, Amazon S3, Azure Data Lake, MongoDB, BigQuery, Elasticsearch...

Điểm mấu chốt cần nhớ:

• Nguồn dữ liệu vẫn là trail file giống hệt bài 4. GG BD không đọc redo log; nó đọc trail do Extract (hoặc một Pump/Distribution path) chuyển tới.
• GG BD chạy ở một máy riêng (thường là máy có JVM và kết nối được tới Kafka), không cần cài trên Oracle DB server.
• Handler chuyển từng operation thành message theo một formatter (JSON, Avro, delimited text...).

Kiến trúc tổng thể

Luồng: Oracle → Extract → trail → (tuỳ chọn Distribution path chuyển trail sang máy GG BD) → Replicat của GG BD gọi Kafka Handler → produce vào topic Kafka → nhiều consumer đọc độc lập.

Trong lab, để đơn giản, ta cho GG BD đọc trực tiếp trail nằm trên cùng máy. Ở production, thường Extract ghi trail cục bộ rồi một Distribution Service / Pump đẩy trail qua mạng sang máy GG BD.

Chuẩn bị môi trường (lab step)

Giả định:

• Đã có trail từ bài 4, ví dụ /ggs/dirdat/aa000000000.
• Kafka đang chạy, broker kafka1:9092.
• Đã cài Oracle GoldenGate for Big Data và có biến JAVA_HOME trỏ tới JDK phù hợp (GG BD chạy trên JVM).

Kiểm tra nhanh Kafka có sống không (lab step):

# Liệt kê topic hiện có
kafka-topics.sh --bootstrap-server kafka1:9092 --list

# (Tuỳ chọn) tạo sẵn topic để kiểm soát số partition / replication
kafka-topics.sh --bootstrap-server kafka1:9092 --create \
  --topic ogg.SALES.ORDERS --partitions 3 --replication-factor 2

Nếu không tạo trước, Kafka sẽ tự tạo topic khi có message đầu tiên (nếu broker bật auto.create.topics.enable=true). Ở môi trường ngân hàng, thường tắt auto-create và tạo topic có kiểm soát.

Hai loại handler Kafka

GG BD cung cấp hai handler để ghi vào Kafka. Chọn đúng loại là quyết định thiết kế quan trọng nhất của bài.

• Kafka Handler (gg.handler.<name>.type=kafka): dùng trực tiếp KafkaProducer của Java client. Message được formatter của GG BD (JSON/Avro/delimited) tạo ra rồi produce nguyên xi. Không đi qua Kafka Connect, không dùng Schema Registry theo cơ chế Connect.
• Kafka Connect Handler (gg.handler.<name>.type=kafkaconnect): đóng gói operation thành Kafka Connect SourceRecord rồi dùng converter của Kafka Connect (JsonConverter hoặc AvroConverter) để serialize. Khi dùng AvroConverter, nó tích hợp với Confluent Schema Registry — schema được đăng ký tự động, message Avro chỉ chứa schema id thay vì toàn bộ schema.

Bảng so sánh

Lưu ý tên tham số: thuộc tính trỏ file cấu hình producer thường viết là gg.handler.<name>.KafkaProducerConfigFile (chú ý chữ hoa K). Một số phiên bản/handler chấp nhận biến thể chữ thường. Nếu Replicat báo lỗi "unknown property", hãy kiểm tra đúng chính tả theo tài liệu phiên bản bạn cài.

Trong lab dưới đây ta làm Kafka Handler (type=kafka) trước vì đơn giản nhất, rồi nêu cấu hình tương đương cho Kafka Connect Handler.

Cấu hình Replicat của GG BD

GG BD dùng một Replicat "thay thế" — về cấu hình có hai phần:

1. File tham số Replicat (.prm): giống Replicat cổ điển, khai báo tên, MAP bảng nguồn → đích logic, và trỏ tới file properties Java.
2. File properties Java (.props): khai báo handlerlist, loại handler, formatter, các mapping template.

File tham số Replicat (lab step)

Tạo dirprm/rkafka.prm:

REPLICAT rkafka
-- TARGETDB LIBFILE libggjava.so SET property=dirprm/rkafka.props
TARGETDB LIBFILE libggjava.so SET property=dirprm/rkafka.props
REPORTCOUNT EVERY 1 MINUTES, RATE
GROUPTRANSOPS 1000
MAP SALES.*, TARGET SALES.*;

Giải thích:

• TARGETDB LIBFILE libggjava.so SET property=...: nạp thư viện Java của GG BD và chỉ tới file properties. Trên Windows dùng ggjava.dll.
• MAP SALES.*, TARGET SALES.*;: ánh xạ mọi bảng schema SALES. "TARGET" ở đây chỉ là tên logic (schema.table) mà handler dùng để suy ra tên topic; không có bảng đích thật.
• GROUPTRANSOPS 1000: gom nhiều thao tác cho hiệu năng.

File properties Java (lab step) — Kafka Handler

Tạo dirprm/rkafka.props:

# Đăng ký danh sách handler đang bật
gg.handlerlist=kafkahandler

# Loại handler: kafka = KafkaProducer thô
gg.handler.kafkahandler.type=kafka

# File cấu hình cho KafkaProducer (bootstrap.servers, serializer...)
gg.handler.kafkahandler.KafkaProducerConfigFile=custom_kafka_producer.properties

# Chế độ ghi: op (mỗi thao tác 1 message) hoặc tx (gom cả transaction)
gg.handler.kafkahandler.mode=op

# Template ánh xạ topic: mỗi bảng → 1 topic tên ogg.<SCHEMA>.<TABLE>
gg.handler.kafkahandler.topicMappingTemplate=ogg.${fullyQualifiedTableName}

# Template khoá message: dùng khoá chính để cùng key vào cùng partition
gg.handler.kafkahandler.keyMappingTemplate=${primaryKeys}

# Formatter: json | json_row | avro_op | avro_row | delimitedtext ...
gg.handler.kafkahandler.format=json_op

# Tuỳ chọn format JSON
gg.handler.kafkahandler.format.insertOpKey=I
gg.handler.kafkahandler.format.updateOpKey=U
gg.handler.kafkahandler.format.deleteOpKey=D
gg.handler.kafkahandler.format.includeOpTimestamp=true

# Classpath: thư mục chứa Kafka client jar + dependencies
gg.classpath=/opt/kafka/libs/*

# JVM
javawriter.bootoptions=-Xmx1024m -Xms256m -Djava.class.path=ggjava/ggjava.jar

Vài điểm cần chú ý:

• topicMappingTemplate: đây là cơ chế "1 bảng → 1 topic". ${fullyQualifiedTableName} cho ra dạng SCHEMA.TABLE, nên topic thành ogg.SALES.ORDERS. Bạn có thể dùng ${tableName} (chỉ tên bảng), ${schemaName}, hoặc hằng để gộp nhiều bảng vào một topic. Nếu muốn cố định một topic duy nhất, ghi thẳng: topicMappingTemplate=all_changes.
• keyMappingTemplate: quyết định message key. Dùng ${primaryKeys} để mọi thay đổi của cùng một bản ghi rơi vào cùng partition → giữ đúng thứ tự theo khoá. Nếu để trống, message không có key và phân phối round-robin.
• format: các giá trị hay dùng:
  • json_op / json_row: JSON. *_op giữ envelope operation (op type, before/after, table, timestamp); *_row gần với hình ảnh bản ghi.
  • avro_op / avro_row: Avro theo formatter GG (Object Container / generic).
  • delimitedtext: CSV/TSV.
• mode=op vs mode=tx: op cho độ trễ thấp (mỗi thao tác produce ngay); tx gom cả giao dịch, ít message hơn nhưng trễ hơn.

File producer Kafka (lab step)

Tạo dirprm/custom_kafka_producer.properties (đường dẫn khớp với KafkaProducerConfigFile ở trên):

bootstrap.servers=kafka1:9092,kafka2:9092
acks=all
compression.type=lz4
batch.size=16384
linger.ms=100
buffer.memory=33554432

# Key/value serializer: dùng ByteArray vì formatter GG đã tạo bytes sẵn
key.serializer=org.apache.kafka.common.serialization.ByteArraySerializer
value.serializer=org.apache.kafka.common.serialization.ByteArraySerializer

# (Production) bảo mật
# security.protocol=SASL_SSL
# sasl.mechanism=SCRAM-SHA-512

Với Kafka Handler (type=kafka), formatter của GG BD đã sinh ra mảng byte, nên serializer thường là ByteArraySerializer. Đừng đặt Avro serializer của Confluent ở đây — đó là việc của Kafka Connect Handler.

Cấu hình cho Kafka Connect Handler (biến thể)

Nếu muốn tích hợp Schema Registry, đổi properties Replicat sang loại kafkaconnect:

gg.handlerlist=kchandler
gg.handler.kchandler.type=kafkaconnect

# File cấu hình producer (lưu ý: tên tham số dạng kafkaProducerConfigFile)
gg.handler.kchandler.kafkaProducerConfigFile=kafkaconnect_producer.properties

# Topic & key template giống Kafka Handler
gg.handler.kchandler.topicMappingTemplate=ogg.${fullyQualifiedTableName}
gg.handler.kchandler.keyMappingTemplate=${primaryKeys}

# Bao gồm/loại bỏ before/after image trong Connect record
gg.handler.kchandler.messageFormatting=op

gg.classpath=/opt/kafka/libs/*

File kafkaconnect_producer.properties khi dùng Avro + Schema Registry:

bootstrap.servers=kafka1:9092
key.converter=io.confluent.connect.avro.AvroConverter
value.converter=io.confluent.connect.avro.AvroConverter
key.converter.schema.registry.url=http://schema-registry:8081
value.converter.schema.registry.url=http://schema-registry:8081

Muốn dùng JSON thay Avro thì đặt key.converter/value.converter là org.apache.kafka.connect.json.JsonConverter và bật/tắt schemas.enable tuỳ nhu cầu.

Ghi rõ chỗ chưa chắc: tên tham số converter và cách khai báo Schema Registry thuộc về Kafka Connect Handler và có thể khác chút giữa các phiên bản GG BD. Với bản 19c/21c, kafkaProducerConfigFile và các key converter kiểu Connect như trên là quy ước phổ biến; hãy đối chiếu tài liệu phiên bản trước khi lên production.

Hiểu message: op type và before/after image

Mỗi thao tác trong trail mang loại (op type): Insert (I), Update (U), Delete (D). Với formatter *_op (JSON hoặc Avro), message giữ nguyên envelope gồm:

• Tên bảng, op type, timestamp commit, vị trí (position/CSN).
• before image: giá trị cột trước khi thay đổi (chỉ có với U/D nếu Extract bắt supplemental logging đủ cột).
• after image: giá trị cột sau khi thay đổi (I/U).

Ví dụ một message JSON (format=json_op) cho một UPDATE:

{
  "table": "SALES.ORDERS",
  "op_type": "U",
  "op_ts": "2026-06-30 09:15:22.000123",
  "current_ts": "2026-06-30T09:15:22.145000",
  "pos": "00000000020000001234",
  "primary_keys": ["ORDER_ID"],
  "before": {
    "ORDER_ID": 1001,
    "STATUS": "NEW",
    "AMOUNT": 500000
  },
  "after": {
    "ORDER_ID": 1001,
    "STATUS": "PAID",
    "AMOUNT": 500000
  }
}

Muốn có before đầy đủ cho UPDATE/DELETE, ở phía Extract (bài 4) phải bật supplemental logging đủ cột, ví dụ ADD TRANDATA SALES.ORDERS COLS (...) hoặc ALL, nếu không before chỉ chứa khoá.

Một INSERT sẽ có op_type: "I", không có before; một DELETE có op_type: "D", after thường rỗng và before chứa hình ảnh bản ghi bị xoá.

Khởi động và kiểm thử (lab step)

Vào GGSCI (hoặc adminclient với Microservices Architecture) trên máy GG BD:

# Khởi động GGSCI
cd $OGG_HOME
./ggsci

Trong GGSCI (lab step):

-- Đăng ký & thêm Replicat, cho đọc từ trail /ggs/dirdat/aa
ADD REPLICAT rkafka, EXTTRAIL ./dirdat/aa

-- Bật Replicat
START REPLICAT rkafka

-- Theo dõi
INFO REPLICAT rkafka
STATS REPLICAT rkafka, TOTAL
VIEW REPORT rkafka

Bây giờ sinh DML trên Oracle để tạo thay đổi (lab step, chạy trên máy Oracle):

sqlplus sales/pass@ORCLPDB1 <<'SQL'
INSERT INTO ORDERS(ORDER_ID, STATUS, AMOUNT) VALUES (1001, 'NEW', 500000);
UPDATE ORDERS SET STATUS='PAID' WHERE ORDER_ID=1001;
COMMIT;
SQL

Đọc message trên Kafka bằng console consumer (lab step):

kafka-console-consumer.sh \
  --bootstrap-server kafka1:9092 \
  --topic ogg.SALES.ORDERS \
  --from-beginning \
  --property print.key=true \
  --property key.separator=" => "

Bạn sẽ thấy lần lượt một message op_type: "I" rồi một op_type: "U" như mẫu JSON ở trên. Nếu topic được đặt tên theo template khác, đổi --topic cho khớp.

Nếu dùng Avro + Schema Registry (Kafka Connect Handler), dùng consumer Avro của Confluent thay vì console consumer thường:

kafka-avro-console-consumer \
  --bootstrap-server kafka1:9092 \
  --topic ogg.SALES.ORDERS \
  --property schema.registry.url=http://schema-registry:8081 \
  --from-beginning

Xử lý lỗi

Vài loại lỗi hay gặp và hướng xử lý trong lab:

• Replicat ABENDED ngay khi start: thường do classpath thiếu Kafka client jar hoặc sai JAVA_HOME. Xem VIEW REPORT rkafka và log JVM (ggserr.log, dirrpt/RKAFKA.rpt). Kiểm tra gg.classpath trỏ đúng libs/* của Kafka.
• Không produce được, broker unreachable: kiểm tra bootstrap.servers, firewall, và security.protocol nếu Kafka bật SASL/SSL. Producer sẽ retry theo retries/delivery.timeout.ms.
• Topic không tồn tại: nếu broker tắt auto-create, tạo topic trước (lệnh kafka-topics.sh --create ở trên) đúng tên do topicMappingTemplate sinh ra.
• before bị thiếu cột: quay lại Extract bật supplemental logging đủ cột (ADD TRANDATA ... ALLCOLS / COLS(...)).
• Chặn lỗi ở mức thao tác: GG BD có tuỳ chọn ghi thao tác lỗi ra file/topic riêng thay vì ABEND. Cơ chế/thuộc tính chi tiết khác nhau theo phiên bản; hãy tra "handler error handling" trong tài liệu bản bạn dùng. Ở mức Replicat cổ điển còn có REPERROR / HANDLECOLLISIONS, nhưng với đích Kafka thì không có "collision" theo nghĩa DB.
• Message quá lớn: nếu bản ghi lớn hơn message.max.bytes của broker, producer báo RecordTooLargeException — tăng max.request.size phía producer và message.max.bytes/max.message.bytes phía broker/topic.

Use case: Core Banking → Kafka

Trong ngân hàng, một bảng như ACCOUNTS, TRANSACTIONS hay LOAN trên core banking Oracle cần phục vụ nhiều hệ tiêu thụ cùng lúc: kho dữ liệu phân tích, hệ chống gian lận real-time, dịch vụ thông báo, và các microservice đọc trạng thái mới nhất.

Mô hình GG BD → Kafka hợp lý vì:

• Tách nguồn khỏi tiêu thụ: core banking chỉ cần Extract đọc redo (bài 4); GG BD produce vào Kafka; các đội khác tự viết consumer, không đụng vào DB nguồn.
• Một bảng → một topic: mỗi domain có topic riêng (ogg.CORE.TRANSACTIONS), consumer group độc lập, mở rộng theo partition.
• Giữ thứ tự theo tài khoản: dùng keyMappingTemplate=${primaryKeys} (hoặc chọn khoá là số tài khoản) để mọi thay đổi của cùng tài khoản đi cùng partition → thứ tự đảm bảo.
• Schema evolution: khi cần đổi cấu trúc bảng, dùng Kafka Connect Handler + Avro + Schema Registry để quản lý tương thích schema, tránh vỡ consumer.

Lưu ý tuân thủ: dữ liệu core banking nhạy cảm → bật security.protocol=SASL_SSL, mã hoá trên đường truyền, và cân nhắc loại bỏ/che cột PII ngay ở tầng formatter hoặc consumer.

Tóm tắt

• GG BD dùng một Replicat + Handler Java để đẩy thay đổi từ trail (không đọc redo trực tiếp) sang đích phi-RDBMS như Kafka.
• Hai handler Kafka: Kafka Handler (type=kafka, KafkaProducer thô, formatter GG) và Kafka Connect Handler (type=kafkaconnect, qua converter Connect + Schema Registry cho Avro).
• Cấu hình chính nằm ở .props: gg.handlerlist, gg.handler.<name>.type, KafkaProducerConfigFile, topicMappingTemplate, keyMappingTemplate, format.
• 1 bảng → 1 topic qua topicMappingTemplate=ogg.${fullyQualifiedTableName}; giữ thứ tự bằng keyMappingTemplate=${primaryKeys}.
• Message mang op type (I/U/D) cùng before/after image; muốn before đủ cột phải bật supplemental logging ở Extract.
• Kiểm thử bằng DML trên Oracle rồi đọc bằng kafka-console-consumer (hoặc kafka-avro-console-consumer nếu dùng Avro/Schema Registry).

Tự kiểm tra

1. GG BD đọc dữ liệu thay đổi từ đâu — redo log hay trail? Tại sao điều này khiến nó không cần cài trên DB server?
2. Khác biệt cốt lõi giữa Kafka Handler và Kafka Connect Handler là gì, và khi nào bạn bắt buộc chọn loại có Schema Registry?
3. Tham số nào quyết định "một bảng → một topic", và giá trị ${fullyQualifiedTableName} sinh ra tên topic như thế nào?
4. Vì sao đặt keyMappingTemplate=${primaryKeys} lại giúp giữ đúng thứ tự thay đổi cho từng bản ghi?
5. Để message UPDATE có before đầy đủ cột, bạn phải cấu hình gì ở phía Extract (bài 4)?
6. Với đích là Kafka, tại sao khái niệm HANDLECOLLISIONS của Replicat cổ điển không còn ý nghĩa?

Đọc tiếp

Oracle CDC 6 — Debezium cho Oracle