Oracle CDC 7 — Kafka Connect, Converters & Schema Registry

Kafka Connect là gì?

Ở các bài trước, Debezium đã đọc redo log của Oracle và đẩy sự kiện thay đổi (change event) vào Kafka. Nhưng Debezium không tự nó là một tiến trình độc lập — nó là một source connector chạy bên trong một framework tên là Kafka Connect. Hiểu Kafka Connect là hiểu bộ khung đỡ toàn bộ pipeline CDC của bạn.

Kafka Connect là framework tích hợp dữ liệu đi kèm Apache Kafka, dùng để chuyển dữ liệu vào và ra khỏi Kafka một cách khai báo (declarative), không cần viết code producer/consumer thủ công. Bạn cấu hình connector bằng JSON, còn framework lo phần khó: phân phối tải, theo dõi offset, xử lý lỗi, tự khởi động lại (restart), và mở rộng theo chiều ngang.

Có hai loại connector:

• Source connector: đọc dữ liệu từ hệ thống ngoài (Oracle, MySQL, file, API...) và ghi vào topic Kafka. Debezium Oracle connector là một source connector.
• Sink connector: đọc dữ liệu từ topic Kafka và ghi ra hệ thống ngoài (PostgreSQL, Elasticsearch, S3, warehouse...). JDBC Sink của Confluent là ví dụ điển hình.

Một pipeline CDC hoàn chỉnh thường ghép cả hai: source connector (Debezium) → Kafka → sink connector (JDBC) → DB đích.

Worker: standalone vs distributed

Connector không tự chạy — nó được host bởi một tiến trình gọi là worker. Có hai chế độ:

• Standalone: một tiến trình duy nhất, cấu hình nằm trong file .properties, offset lưu vào file cục bộ. Đơn giản, hợp cho dev/test hoặc chạy một máy. Không có khả năng chịu lỗi (nếu worker chết, pipeline dừng).
• Distributed: nhiều worker cùng một group.id tạo thành cluster. Connector và các task của nó được phân phối (rebalance) tự động giữa các worker. Nếu một worker chết, task của nó được chuyển sang worker khác. Đây là chế độ dùng cho production.

Điểm mấu chốt của distributed mode: trạng thái không lưu trên đĩa cục bộ mà lưu trong ba topic Kafka nội bộ:

Ba topic này nên là compacted để giữ giá trị mới nhất. Vì trạng thái nằm trong Kafka, bất kỳ worker nào trong cluster cũng phục hồi được — đó là nguồn gốc khả năng chịu lỗi.

Connector → Task

Một connector là một cấu hình logic. Framework chia công việc thực tế thành các task (đơn vị thực thi song song). Ví dụ một JDBC source đọc 10 bảng có thể chia thành nhiều task; tasks.max giới hạn số task. Riêng Debezium Oracle connector thường chỉ tạo một task vì việc đọc redo log là tuần tự — đừng kỳ vọng tăng tasks.max sẽ tăng thông lượng cho nó.

REST API

Trong distributed mode, bạn quản lý mọi thứ qua REST API (mặc định cổng 8083). Đây là cách chuẩn để đăng ký, cập nhật, xem trạng thái và xóa connector.

# Liệt kê connector đang chạy
curl -s http://localhost:8083/connectors | jq

# Xem trạng thái + task
curl -s http://localhost:8083/connectors/oracle-src/status | jq

# Danh sách plugin connector đã cài trên worker
curl -s http://localhost:8083/connector-plugins | jq

# Tạm dừng / khởi động lại
curl -s -X PUT http://localhost:8083/connectors/oracle-src/pause
curl -s -X POST http://localhost:8083/connectors/oracle-src/restart

# Xóa connector
curl -s -X DELETE http://localhost:8083/connectors/oracle-src

Converter — cầu nối giữa Connect và Kafka

Đây là khái niệm hay bị hiểu nhầm nhất. Bên trong Kafka Connect, mỗi bản ghi được biểu diễn bằng một cấu trúc trung lập gọi là Connect data API (org.apache.kafka.connect.data.Struct + Schema). Nhưng Kafka chỉ lưu byte thô. Converter là thành phần chuyển đổi giữa Connect Struct và byte:

• Với source connector: converter serialize Struct → byte trước khi ghi vào Kafka.
• Với sink connector: converter deserialize byte → Struct trước khi đưa cho connector.

Bạn cấu hình riêng cho key và value:

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

Các converter thường dùng

schemas.enable với JsonConverter

JsonConverter không dùng Schema Registry. Nếu value.converter.schemas.enable=true, nó nhúng cả schema vào từng message dưới dạng object {"schema": {...}, "payload": {...}}. Điều này làm message phình to đáng kể (schema lặp lại ở mọi bản ghi) nhưng consumer tự đọc được kiểu dữ liệu. Nếu đặt false, chỉ còn payload thuần — nhỏ gọn hơn nhưng consumer phải tự biết schema.

Lưu ý quan trọng với Debezium: nếu bạn đặt schemas.enable=false cho value, envelope Debezium (chứa before/after/op/source) vẫn còn nhưng mất thông tin kiểu. Thông thường với JSON để schemas.enable=true; còn với Avro thì tham số này không áp dụng.

Schema Registry — quản lý schema tập trung

Khi dữ liệu chảy liên tục qua Kafka trong nhiều năm, schema sẽ thay đổi: thêm cột, đổi kiểu, xóa cột. Nếu producer đổi schema mà consumer không biết, hệ thống vỡ. Schema Registry (của Confluent, cổng mặc định 8081) giải quyết bằng cách:

1. Lưu trữ tất cả các phiên bản schema, mỗi schema có một ID số nguyên duy nhất.
2. Khi source ghi bằng AvroConverter, message trên Kafka không chứa schema — chỉ chứa một byte magic + schema ID + dữ liệu Avro. Nhờ đó message rất gọn.
3. Consumer/sink đọc schema ID, gọi registry lấy schema tương ứng để deserialize.
4. Kiểm tra tương thích (compatibility) khi đăng ký schema mới, chặn thay đổi phá vỡ.

Vì sao cần Avro + Registry thay vì JSON nhúng schema?

• Kích thước: Avro chỉ gửi schema ID (vài byte) thay vì nhúng cả schema (hàng KB) vào mỗi message.
• Ràng buộc: Registry ép các schema tuân theo chính sách tương thích, không để lọt thay đổi phá vỡ.
• Chuẩn hóa: mọi consumer tham chiếu cùng một định nghĩa schema.

Chế độ tương thích (compatibility)

Đây là phần cốt lõi để schema evolution an toàn. Mỗi subject (thường là <topic>-value) có một chính sách:

Quy tắc thực dụng cho CDC: dùng BACKWARD (mặc định). Khi thêm cột vào bảng Oracle, Debezium sẽ thêm field vào Avro schema — miễn field mới có giá trị default thì consumer cũ vẫn deserialize được và schema mới được registry chấp nhận. Ngược lại, xóa cột cũng an toàn với BACKWARD vì consumer mới bỏ qua field không còn.

# Xem/đặt compatibility cho một subject
curl -s http://localhost:8081/config/oracle.CDCUSER.CUSTOMERS-value | jq
curl -s -X PUT -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  --data '{"compatibility": "BACKWARD"}' \
  http://localhost:8081/config/oracle.CDCUSER.CUSTOMERS-value

# Liệt kê subject và các phiên bản
curl -s http://localhost:8081/subjects | jq
curl -s http://localhost:8081/subjects/oracle.CDCUSER.CUSTOMERS-value/versions | jq

SMT — Single Message Transforms

SMT là các phép biến đổi từng bản ghi một (không có state, không join, không aggregate) áp dụng ngay trong pipeline Connect — sau khi source đọc/trước khi sink ghi. Chúng cực kỳ hữu ích để "gọt" dữ liệu mà không cần một tầng xử lý riêng.

Vấn đề: envelope Debezium

Message Debezium mặc định là một envelope dạng:

{
  "before": { "ID": 1, "NAME": "cũ" },
  "after":  { "ID": 1, "NAME": "mới" },
  "op": "u",
  "ts_ms": 1719750000000,
  "source": { "scn": "123456", "table": "CUSTOMERS" }
}

Cấu trúc này giàu thông tin nhưng một JDBC sink không hiểu được — nó muốn một bản ghi phẳng chỉ gồm các cột (ID, NAME). Đây là chỗ SMT ExtractNewRecordState vào cuộc.

ExtractNewRecordState — "unwrap" envelope

SMT nổi tiếng nhất trong CDC là của Debezium:

{
  "transforms": "unwrap",
  "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
  "transforms.unwrap.drop.tombstones": "false",
  "transforms.unwrap.delete.handling.mode": "rewrite",
  "transforms.unwrap.add.fields": "op,source.ts_ms,source.scn"
}

Nó lấy nội dung after đưa lên cấp cao nhất → biến envelope thành bản ghi phẳng đúng như bảng nguồn. Vài tham số quan trọng:

• delete.handling.mode: xử lý sự kiện DELETE. rewrite giữ lại bản ghi và thêm cờ __deleted=true (hữu ích cho soft-delete ở đích); drop bỏ hẳn; none chuyển thành tombstone.
• drop.tombstones: có bỏ các message tombstone (value = null) hay không.
• add.fields: chèn thêm metadata từ envelope (op, ts, scn) vào bản ghi phẳng, tiền tố mặc định là __.

Các SMT hữu dụng khác

Nhiều SMT có thể xếp chuỗi qua transforms (thứ tự có ý nghĩa):

{
  "transforms": "unwrap,route,mask",

  "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",

  "transforms.route.type": "org.apache.kafka.connect.transforms.RegexRouter",
  "transforms.route.regex": "oracle\\.CDCUSER\\.(.*)",
  "transforms.route.replacement": "cdc_$1",

  "transforms.mask.type": "org.apache.kafka.connect.transforms.MaskField$Value",
  "transforms.mask.fields": "SSN,PHONE",
  "transforms.mask.replacement": "***"
}

• RegexRouter (class org.apache.kafka.connect.transforms.RegexRouter): đổi tên topic đích. Ví dụ trên biến topic oracle.CDCUSER.CUSTOMERS thành cdc_CUSTOMERS.
• MaskField$Value (class org.apache.kafka.connect.transforms.MaskField$Value): che (mask) các field nhạy cảm. Lưu ý hậu tố $Value (áp dụng cho value) hoặc $Key (áp dụng cho key) — sai hậu tố là lỗi phổ biến.
• Các SMT chuẩn khác: InsertField, ReplaceField, TimestampConverter, Cast, ValueToKey, ExtractField.

Sink connector: JDBC Sink

Bây giờ ghép nửa còn lại của pipeline: đưa dữ liệu từ Kafka vào DB đích. Confluent JDBC Sink Connector (class io.confluent.connect.jdbc.JdbcSinkConnector) đọc topic và ghi vào bất kỳ DB nào có driver JDBC (PostgreSQL, MySQL, SQL Server...).

Hai chế độ chính:

• insert: chỉ INSERT (append-only). Không hợp cho CDC vì UPDATE sẽ nhân đôi dữ liệu.
• upsert: INSERT nếu chưa có, UPDATE nếu đã có — dựa trên primary key. Đây là chế độ đúng cho CDC.

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

    "connection.url": "jdbc:postgresql://pg:5432/dw",
    "connection.user": "dw_user",
    "connection.password": "secret",

    "insert.mode": "upsert",
    "pk.mode": "record_key",
    "pk.fields": "ID",

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

    "key.converter": "io.confluent.connect.avro.AvroConverter",
    "key.converter.schema.registry.url": "http://schema-registry:8081",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "http://schema-registry:8081"
  }
}

Các điểm quan trọng:

• pk.mode + pk.fields: xác định khóa để upsert. record_key lấy khóa từ key của message (Debezium đặt PK bảng làm key), record_value lấy từ các field trong value.
• auto.create: tự tạo bảng đích nếu chưa tồn tại, suy ra kiểu cột từ Connect schema.
• auto.evolve: khi source thêm cột (schema evolution), sink tự ALTER TABLE ADD COLUMN. Đây là lý do bạn cần schema chuẩn (Avro + Registry) — không có schema thì sink không biết phải tạo cột kiểu gì.
• delete.enabled: khi kết hợp với pk.mode=record_key, một message có value = null (tombstone) sẽ được dịch thành DELETE ở đích. Muốn dùng nó, đừng để SMT drop tombstone.

Lưu ý deserialize envelope ở sink

Đây là cái bẫy lớn nhất khi ghép Debezium với JDBC sink: nếu bạn không unwrap, sink sẽ cố ghi cả envelope (before, after, op...) thành các cột — hoặc báo lỗi vì cấu trúc lồng nhau. Có hai cách xử lý:

1. Unwrap ở source (khuyến nghị): cấu hình ExtractNewRecordState ngay trong Debezium connector, để topic đã chứa bản ghi phẳng.
2. Unwrap ở sink: đặt cùng SMT unwrap trong cấu hình JDBC sink. Cùng một transform, chỉ khác nơi áp dụng.

Ngoài ra, converter ở sink phải khớp với converter mà source đã dùng để ghi. Nếu source ghi Avro thì sink phải đọc bằng AvroConverter cùng trỏ tới đúng Schema Registry, nếu không sẽ lỗi deserialize (SerializationException).

Sơ đồ pipeline CDC hoàn chỉnh

Luồng đọc từ trái sang phải: Debezium đọc redo log → SMT unwrap gọt thành bản ghi phẳng → AvroConverter serialize và đăng ký schema với Registry → ghi vào topic Kafka → JDBC sink đọc topic, deserialize bằng Avro (lấy schema từ Registry), rồi upsert vào PostgreSQL. Toàn bộ source lẫn sink cùng chạy trong một Connect cluster, chia sẻ ba topic trạng thái nội bộ.

Lab: đăng ký sink qua REST

Giả sử Debezima source đã chạy và topic cdc_CUSTOMERS đã có dữ liệu phẳng (đã unwrap). Đăng ký sink:

# 1. Ghi cấu hình sink ra file
cat > pg-sink.json <<'EOF'
{
  "name": "pg-sink",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "tasks.max": "1",
    "topics": "cdc_CUSTOMERS",
    "connection.url": "jdbc:postgresql://pg:5432/dw",
    "connection.user": "dw_user",
    "connection.password": "secret",
    "insert.mode": "upsert",
    "pk.mode": "record_key",
    "pk.fields": "ID",
    "auto.create": "true",
    "auto.evolve": "true",
    "delete.enabled": "true",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "http://schema-registry:8081",
    "key.converter": "io.confluent.connect.avro.AvroConverter",
    "key.converter.schema.registry.url": "http://schema-registry:8081"
  }
}
EOF

# 2. Đăng ký (lab step)
curl -s -X POST -H "Content-Type: application/json" \
  --data @pg-sink.json http://localhost:8083/connectors | jq

# 3. Kiểm tra trạng thái — mong đợi state RUNNING
curl -s http://localhost:8083/connectors/pg-sink/status | jq

# 4. Nếu FAILED, xem log lỗi của task
curl -s http://localhost:8083/connectors/pg-sink/status \
  | jq '.tasks[].trace'

Sau khi sink RUNNING, thử một UPDATE trên bảng Oracle nguồn và kiểm tra dữ liệu đã xuất hiện/được cập nhật trong PostgreSQL. Nếu bạn DELETE một hàng ở Oracle và delete.enabled=true, hàng tương ứng ở đích cũng biến mất.

Tóm tắt

Kafka Connect là bộ khung tích hợp khai báo: source đưa dữ liệu vào Kafka, sink đưa dữ liệu ra; ở production dùng distributed worker với ba topic trạng thái (configs, offsets, status) và quản lý qua REST API cổng 8083. Converter dịch giữa Connect Struct và byte — JsonConverter (không cần registry, có thể nhúng schema), AvroConverter/JsonSchemaConverter (cần Schema Registry cổng 8081). Schema Registry lưu schema theo ID, ép compatibility (BACKWARD/FORWARD/FULL) để schema evolution an toàn. SMT biến đổi từng bản ghi: ExtractNewRecordState unwrap envelope Debezium thành bản ghi phẳng, RegexRouter đổi tên topic, MaskField$Value che dữ liệu nhạy cảm. JDBC Sink (io.confluent.connect.jdbc.JdbcSinkConnector) ghi vào DB đích ở chế độ upsert theo PK, với auto.create/auto.evolve/delete.enabled. Ghép Debezium + unwrap SMT + JDBC sink cho một pipeline CDC hoàn chỉnh — nhớ unwrap trước khi tới sink và dùng cùng converter hai đầu.

Tự kiểm tra

1. Ba topic nội bộ nào lưu trạng thái của một Kafka Connect distributed cluster, và tại sao chúng nên là compacted?
2. Phân biệt hành vi của JsonConverter khi schemas.enable=true và false; khác biệt gì so với AvroConverter về những gì được ghi lên Kafka?
3. Nếu bạn thêm một cột có default vào bảng Oracle, chế độ compatibility BACKWARD cho phép hay chặn schema mới? Consumer nào cần nâng cấp trước?
4. Tên class đầy đủ của SMT dùng để "unwrap" envelope Debezium là gì, và điều gì xảy ra ở JDBC sink nếu bạn quên áp dụng nó?
5. Trong JDBC sink, pk.mode=record_key kết hợp delete.enabled=true xử lý một tombstone (value null) như thế nào?
6. Vì sao auto.evolve của JDBC sink chỉ hoạt động đáng tin cậy khi dùng Avro + Schema Registry chứ không phải JSON không schema?

Đọc tiếp

Oracle CDC 8 — Vận hành & So sánh