Kafka 5 — Kafka Connect (tích hợp dữ liệu)

Kafka Connect là gì và vì sao cần nó

Trong ba bài trước, mỗi khi muốn đưa dữ liệu vào hoặc lấy dữ liệu ra khỏi Kafka, ta thường tự viết một ứng dụng dùng Producer API hoặc Consumer API. Cách này linh hoạt nhưng sinh ra rất nhiều mã lặp đi lặp lại và khó vận hành: bạn phải tự lo việc theo dõi offset đã đọc tới đâu, tự xử lý retry khi hệ thống đích tạm sập, tự cân bằng tải khi tăng số instance, tự chuyển đổi định dạng dữ liệu, tự giám sát và khởi động lại khi lỗi.

Kafka Connect là một framework tích hợp dữ liệu (data integration) đi kèm Apache Kafka, giải quyết chính xác lớp mã "nhàm chán nhưng quan trọng" đó. Thay vì viết code producer/consumer, bạn chỉ cần khai báo cấu hình (thường là một tệp JSON) cho một connector có sẵn — Connect sẽ lo phần còn lại: phân phối công việc, quản lý offset, scale, chịu lỗi và chuyển đổi dữ liệu.

Tư duy cốt lõi cần nắm:

• Connect không thay thế Producer/Consumer API cho logic nghiệp vụ phức tạp. Nó chuyên cho bài toán di chuyển dữ liệu giữa Kafka và hệ thống ngoài (database, object storage, search engine, message queue...).
• Connect được thiết kế để chạy như một dịch vụ dùng chung: một cụm worker có thể phục vụ hàng chục connector khác nhau, mỗi connector chỉ là một bản cấu hình.
• Hệ sinh thái connector rất lớn: Debezium (CDC cho Oracle, MySQL, PostgreSQL, SQL Server), JDBC, S3, Elasticsearch, JMS, MongoDB... nên phần lớn trường hợp bạn không phải viết connector mới.

Source connector và Sink connector

Đây là hai hướng dữ liệu cơ bản, cần phân biệt rõ:

• Source connector: đọc dữ liệu từ hệ thống nguồn bên ngoài (Oracle, MySQL, file, API...) và ghi vào Kafka topic. Ví dụ Debezium đọc redo log của Oracle rồi phát các bản ghi thay đổi (change event) vào Kafka.
• Sink connector: đọc dữ liệu từ Kafka topic và ghi ra hệ thống đích bên ngoài (data warehouse, S3, Elasticsearch, một database khác...). Ví dụ JDBC Sink ghi bản ghi từ Kafka xuống bảng trong PostgreSQL.

Một luồng end-to-end điển hình trong data platform: Oracle → (Debezium source) → Kafka → (JDBC/S3 sink) → warehouse. Đây chính là kiến trúc CDC mà series Oracle CDC đề cập — Connect là "keo dán" ở tầng vận chuyển.

Điểm quan trọng về offset:

• Source connector lưu source offset (ví dụ SCN của Oracle, vị trí file, id bản ghi cuối cùng đã đọc) để biết lần chạy sau đọc tiếp từ đâu.
• Sink connector không cần offset riêng — nó dùng consumer group offset của Kafka như một consumer bình thường, nên tận dụng toàn bộ cơ chế commit offset đã học.

Worker: standalone và distributed

Connect chạy bằng các tiến trình gọi là worker. Có hai chế độ:

Standalone

Một tiến trình duy nhất chạy tất cả connector và task, lưu offset vào file cục bộ (offset.storage.file.filename). Ưu điểm là đơn giản, phù hợp phát triển, thử nghiệm, hoặc kịch bản chạy trên một máy (ví dụ đọc log file cục bộ). Nhược điểm: không chịu lỗi, không scale ngang. Máy chết là dừng.

Distributed (khuyến nghị cho production)

Nhiều worker cùng chạy, tạo thành một Connect cluster. Đặc điểm:

• Tất cả worker chia sẻ cùng một group.id. Chúng dùng chính cơ chế group membership / rebalance của Kafka (giống consumer group) để phối hợp.
• Không lưu offset/config ra file. Thay vào đó dùng ba topic nội bộ trên chính Kafka:

• Khi một worker thêm/bớt hoặc một connector được tạo/xóa, cluster thực hiện rebalance để phân bổ lại các task cho các worker còn sống. Từ Kafka 2.3 trở đi, Connect dùng incremental cooperative rebalancing để tránh "stop-the-world" (không phải dừng toàn bộ task mỗi lần rebalance).
• Quản lý connector qua REST API; gọi vào worker nào cũng được, request sẽ được chuyển tiếp tới leader.

Distributed vừa scale (thêm worker để tăng thông lượng) vừa fault-tolerant (worker chết thì task được chuyển sang worker khác, tiếp tục từ offset đã commit).

Task và max.tasks

Bên trong một connector, công việc thực tế được chia thành các task. Task mới là đơn vị được phân phối lên worker và chạy song song.

• tasks.max: số task tối đa mà connector được phép tạo. Đây là giới hạn trên, không phải con số cố định.
• Connector tự quyết định tạo bao nhiêu task dựa trên khả năng chia nhỏ công việc. Ví dụ JDBC Source đọc 3 bảng, đặt tasks.max=10 thì thường chỉ tạo tối đa 3 task (mỗi task một bảng). Debezium Oracle thường chỉ dùng 1 task vì phải đọc redo log tuần tự.
• Với Sink connector, số task hiệu quả bị chặn bởi số partition của topic nguồn (một partition chỉ được một task xử lý tại một thời điểm), giống nguyên tắc consumer group.

Vì vậy đặt tasks.max cao không tự động nhanh hơn — nó chỉ hữu ích khi công việc thực sự chia nhỏ được.

Converter: cầu nối giữa byte và dữ liệu có cấu trúc

Kafka lưu message dưới dạng byte array cho cả key và value. Connect làm việc với một mô hình dữ liệu trung gian có schema (Connect data + schema). Converter là thành phần dịch qua lại giữa byte trên Kafka và mô hình đó:

• Source: Connect data → (converter serialize) → bytes → Kafka.
• Sink: bytes từ Kafka → (converter deserialize) → Connect data → connector ghi ra đích.

Cấu hình bằng key.converter và value.converter (có thể đặt mức worker hoặc override ở mức từng connector).

Vai trò của Schema Registry

Với Avro/Protobuf/JSON Schema, converter không nhúng schema vào message. Thay vào đó nó đăng ký schema lên Schema Registry và chỉ gắn một schema id (vài byte) vào đầu payload. Điều này:

• Giảm mạnh kích thước message so với JsonConverter có schemas.enable=true.
• Cho phép kiểm soát tương thích schema (compatibility) khi schema tiến hóa — quan trọng cho pipeline dài hạn Oracle → warehouse, nơi cấu trúc bảng nguồn thay đổi theo thời gian.

Cấu hình đi kèm khi dùng Avro: value.converter.schema.registry.url.

Lưu ý mấu chốt: key converter và value converter độc lập. Rất phổ biến kiểu key dùng StringConverter còn value dùng AvroConverter.

SMT — Single Message Transforms

SMT cho phép biến đổi từng message một cách nhẹ, ngay trong pipeline Connect, mà không cần viết stream processing riêng. SMT chạy sau converter (khi source) hoặc trước converter (khi sink) và xử lý mỗi bản ghi độc lập (không join, không aggregate — cho việc đó dùng Kafka Streams).

Các SMT hay dùng:

• Debezium ExtractNewRecordState (thường gọi là unwrap): Debezium phát bản ghi có cấu trúc before/after/op/source. SMT này "mở gói" để chỉ giữ trạng thái after phẳng, giúp JDBC/Elasticsearch sink ghi thẳng như một hàng dữ liệu bình thường.
• MaskField: che (mask) trường nhạy cảm như số CMND, số thẻ.
• RegexRouter / TimestampRouter: đổi tên topic đích theo quy tắc, ví dụ gộp/định tuyến topic theo pattern.
• InsertField, ValueToKey, Cast, ReplaceField: thêm metadata, dựng key từ trường, ép kiểu, đổi tên/bỏ trường.

Ví dụ khai báo unwrap của Debezium:

{
  "transforms": "unwrap,route",
  "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
  "transforms.unwrap.drop.tombstones": "false",
  "transforms.unwrap.delete.handling.mode": "rewrite",
  "transforms.route.type": "org.apache.kafka.connect.transforms.RegexRouter",
  "transforms.route.regex": "oracle\\.SALES\\.(.*)",
  "transforms.route.replacement": "warehouse_$1"
}

transforms liệt kê tên logic (thứ tự áp dụng), mỗi tên có transforms.<name>.type và các tham số riêng.

Sơ đồ luồng dữ liệu qua Connect

Quản lý bằng REST API

Trong chế độ distributed, mọi thao tác quản lý đi qua REST API của worker (mặc định cổng 8083). Đây là cách chuẩn để tạo, xem, sửa, tạm dừng, khởi động lại connector.

# Xem danh sách connector đang có
curl -s http://localhost:8083/connectors | jq

# Tạo connector mới bằng file cấu hình JSON
curl -s -X POST http://localhost:8083/connectors \
  -H "Content-Type: application/json" \
  -d @debezium-oracle.json

# Xem trạng thái tổng thể của connector và từng task
curl -s http://localhost:8083/connectors/inventory-oracle/status | jq

# Cập nhật cấu hình (idempotent) - chỉ gửi phần "config"
curl -s -X PUT http://localhost:8083/connectors/inventory-oracle/config \
  -H "Content-Type: application/json" \
  -d @debezium-oracle-config-only.json

# Tạm dừng / tiếp tục
curl -s -X PUT http://localhost:8083/connectors/inventory-oracle/pause
curl -s -X PUT http://localhost:8083/connectors/inventory-oracle/resume

# Khởi động lại connector và các task đang FAILED
curl -s -X POST "http://localhost:8083/connectors/inventory-oracle/restart?includeTasks=true&onlyFailed=true"

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

Một số endpoint hữu ích khác:

• GET /connector-plugins — liệt kê các loại connector đã cài trên worker.
• PUT /connector-plugins/<class>/config/validate — validate cấu hình trước khi tạo.
• GET /connectors/<name>/tasks/<id>/status — trạng thái từng task cụ thể.

Khi gọi .../status, trường state cho biết connector/task đang RUNNING, PAUSED, hay FAILED kèm trace là stack trace giúp chẩn đoán lỗi.

Xử lý lỗi và Dead Letter Queue

Lỗi trong Connect chủ yếu xảy ra ở khâu converter (deserialize hỏng) và SMT/ghi ra đích phía sink. Mặc định, một bản ghi lỗi có thể làm task chết (fail-fast). Connect cung cấp cơ chế cấu hình để kiểm soát:

Dead Letter Queue (DLQ): khi errors.tolerance=all và có errors.deadletterqueue.topic.name, những message mà sink connector không xử lý được (ví dụ JSON hỏng, sai kiểu) sẽ được ghi sang topic DLQ thay vì làm chết task. Sau đó bạn có thể phân tích, sửa, và replay riêng — pipeline chính không bị nghẽn.

Lưu ý: DLQ chỉ dành cho sink connector. Source connector không có DLQ tương đương; lỗi phía source thường phải xử lý bằng retry và giám sát.

Cấu hình mẫu cho phần error handling của một sink:

{
  "errors.tolerance": "all",
  "errors.log.enable": "true",
  "errors.log.include.messages": "true",
  "errors.deadletterqueue.topic.name": "dlq.warehouse.sink",
  "errors.deadletterqueue.topic.replication.factor": "3",
  "errors.deadletterqueue.context.headers.enable": "true"
}

Các connector phổ biến và ví dụ cấu hình

Debezium Oracle (source, CDC)

Đọc thay đổi từ Oracle qua LogMiner / redo log và phát vào Kafka.

{
  "name": "inventory-oracle",
  "config": {
    "connector.class": "io.debezium.connector.oracle.OracleConnector",
    "tasks.max": "1",
    "database.hostname": "oracle-db",
    "database.port": "1521",
    "database.user": "c##dbzuser",
    "database.password": "dbz",
    "database.dbname": "ORCLCDB",
    "database.pdb.name": "ORCLPDB1",
    "topic.prefix": "oracle",
    "table.include.list": "SALES.ORDERS,SALES.CUSTOMERS",
    "schema.history.internal.kafka.bootstrap.servers": "kafka:9092",
    "schema.history.internal.kafka.topic": "schema-history.oracle",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "http://schema-registry:8081"
  }
}

JDBC Sink (ghi vào database đích / warehouse)

{
  "name": "warehouse-jdbc-sink",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "tasks.max": "3",
    "topics": "oracle.SALES.ORDERS",
    "connection.url": "jdbc:postgresql://warehouse:5432/dw",
    "connection.user": "dw_writer",
    "connection.password": "secret",
    "insert.mode": "upsert",
    "pk.mode": "record_key",
    "pk.fields": "ORDER_ID",
    "auto.create": "true",
    "auto.evolve": "true",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "http://schema-registry:8081",
    "transforms": "unwrap",
    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
    "errors.tolerance": "all",
    "errors.deadletterqueue.topic.name": "dlq.warehouse.sink",
    "errors.deadletterqueue.context.headers.enable": "true"
  }
}

S3 Sink (đổ dữ liệu thô xuống object storage)

{
  "name": "s3-raw-sink",
  "config": {
    "connector.class": "io.confluent.connect.s3.S3SinkConnector",
    "tasks.max": "4",
    "topics": "oracle.SALES.ORDERS",
    "s3.bucket.name": "dp-datalake-raw",
    "s3.region": "ap-southeast-1",
    "flush.size": "10000",
    "rotate.interval.ms": "600000",
    "storage.class": "io.confluent.connect.s3.storage.S3Storage",
    "format.class": "io.confluent.connect.s3.format.parquet.ParquetFormat",
    "partitioner.class": "io.confluent.connect.storage.partitioner.TimeBasedPartitioner",
    "path.format": "'year'=YYYY/'month'=MM/'day'=dd",
    "partition.duration.ms": "86400000",
    "locale": "en",
    "timezone": "Asia/Ho_Chi_Minh"
  }
}

Elasticsearch Sink (phục vụ tìm kiếm)

{
  "name": "orders-es-sink",
  "config": {
    "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
    "tasks.max": "2",
    "topics": "oracle.SALES.ORDERS",
    "connection.url": "http://elasticsearch:9200",
    "key.ignore": "false",
    "schema.ignore": "true",
    "write.method": "upsert",
    "transforms": "unwrap",
    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState"
  }
}

Đặt trong bối cảnh pipeline Oracle → Kafka → Warehouse

Ghép lại, một pipeline CDC hoàn chỉnh dùng Connect gồm:

1. Debezium Oracle source đọc redo log, phát change event vào topic oracle.SALES.ORDERS, dùng AvroConverter + Schema Registry để chuẩn hóa và kiểm soát tiến hóa schema.
2. Các SMT ExtractNewRecordState làm phẳng bản ghi để dễ tiêu thụ.
3. JDBC Sink upsert vào warehouse (giữ trạng thái mới nhất) song song với S3 Sink đổ bản thô dạng Parquet xuống data lake (để reprocess/kiểm toán).
4. DLQ hứng các bản ghi lỗi ở phía sink; đội vận hành theo dõi status qua REST API và cảnh báo khi có task FAILED.

Toàn bộ chỉ là cấu hình JSON trên một Connect cluster distributed — không dòng code producer/consumer nào, và vẫn có scale, chịu lỗi, giám sát chuẩn hóa. Đó là giá trị lớn nhất của Kafka Connect trong một data platform.

Tóm tắt

Kafka Connect là framework tích hợp dữ liệu giúp di chuyển dữ liệu giữa Kafka và hệ thống ngoài bằng cấu hình thay vì code. Source đưa dữ liệu vào Kafka, sink đưa ra ngoài. Chế độ distributed dùng ba topic nội bộ (config/offset/status) để scale và chịu lỗi qua cơ chế rebalance, quản lý bằng REST API cổng 8083. Converter (JSON/Avro/Protobuf/JSON Schema) dịch giữa byte và dữ liệu có schema, trong đó Avro/Protobuf cần Schema Registry. SMT biến đổi từng message (unwrap, mask, route) mà không cần stream processing. Error handling với errors.tolerance=all và DLQ (chỉ cho sink) giúp pipeline không nghẽn vì vài bản ghi hỏng. Các connector phổ biến gồm Debezium, JDBC, S3, Elasticsearch — đủ dựng luồng Oracle → Kafka → warehouse hoàn chỉnh.

Tự kiểm tra

1. Khác biệt cốt lõi giữa source connector và sink connector là gì, và mỗi loại lưu offset ở đâu?
2. Tại sao chế độ distributed cần ba topic nội bộ, và topic config.storage.topic phải có bao nhiêu partition?
3. Đặt tasks.max=10 cho một Debezium Oracle source có làm nó chạy 10 task song song không? Vì sao?
4. Khi nào converter cần Schema Registry, và điều đó giúp ích gì so với JsonConverter có schemas.enable=true?
5. DLQ áp dụng cho source hay sink connector, và cần bật những cấu hình nào để một bản ghi lỗi đi vào DLQ thay vì làm chết task?
6. SMT ExtractNewRecordState của Debezium giải quyết vấn đề gì khi ghi xuống JDBC/Elasticsearch sink?

Đọc tiếp

Kafka 6 — Kafka Streams & ksqlDB