Kafka Connect 与 CDC 集成实战：Debezium、Schema Registry 与数据同步

# Standalone 模式启动
bin/connect-standalone.sh \
  config/connect-standalone.properties \
  config/connect-file-source.properties \
  config/connect-file-sink.properties

# Distributed 模式启动
bin/connect-distributed.sh config/connect-distributed.properties

# 与 Kafka 集群通信
bootstrap.servers=kafka1:9092,kafka2:9092,kafka3:9092

# 连接器配置存储 Topic，建议单分区多副本
group.id=connect-cluster
config.storage.topic=connect-configs
config.storage.replication.factor=3

# 偏移量存储 Topic
offset.storage.topic=connect-offsets
offset.storage.replication.factor=3
offset.storage.partitions=25

# 状态存储 Topic
status.storage.topic=connect-status
status.storage.replication.factor=3
status.storage.partitions=5

# Worker 心跳与超时
heartbeat.interval.ms=3000
session.timeout.ms=10000

# 转换与错误处理
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
errors.tolerance=all
errors.log.enable=true

# 列出所有连接器
curl http://localhost:8083/connectors

# 创建连接器
curl -X POST http://localhost:8083/connectors \
  -H "Content-Type: application/json" \
  -d '{
    "name": "file-source-connector",
    "config": {
      "connector.class": "org.apache.kafka.connect.file.FileStreamSourceConnector",
      "tasks.max": "1",
      "topic": "connect-test",
      "file": "/tmp/test.txt"
    }
  }'

# 查看连接器状态
curl http://localhost:8083/connectors/file-source-connector/status

# 暂停连接器（停止拉取数据但保留状态）
curl -X PUT http://localhost:8083/connectors/file-source-connector/pause

# 恢复连接器
curl -X PUT http://localhost:8083/connectors/file-source-connector/resume

# 删除连接器
curl -X DELETE http://localhost:8083/connectors/file-source-connector

# JDBC Source Connector 配置示例
name=jdbc-source-postgres
connector.class=io.confluent.connect.jdbc.JdbcSourceConnector
tasks.max=4

# 数据库连接
connection.url=jdbc:postgresql://db:5432/ecommerce?sslmode=require
connection.user=connect_user
connection.password=${file:/secrets/connect-password.txt:password}

# 采集模式：timestamp+incrementing 实现增量同步
mode=timestamp+incrementing
timestamp.column.name=updated_at
incrementing.column.name=id

# 定时轮询间隔
poll.interval.ms=5000

# 表白名单与 Topic 前缀
table.whitelist=orders,users,products
topic.prefix=db.ecommerce.

# 并行化：按表分配任务
tables.tasks.max.per.table=2

name=file-source
connector.class=org.apache.kafka.connect.file.FileStreamSourceConnector
tasks.max=1
topic=log-lines
file=/var/log/app/server.log

name=es-sink-orders
connector.class=io.confluent.connect.elasticsearch.ElasticsearchSinkConnector
tasks.max=4

# Kafka 来源
topics=db.ecommerce.orders

# ES 集群地址
connection.url=http://es1:9200,http://es2:9200
connection.username=elastic
connection.password=${file:/secrets/es-password.txt:password}

# 索引配置
index.prefix=ecommerce-
type.name=_doc

# 写入行为：upsert 模式
write.method=UPSERT
key.ignore=false

# 数据结构转换
schema.ignore=true
drop.invalid.message=true

# 批处理优化
batch.size=2000
linger.ms=1000
max.in.flight.requests=5

# 重试与超时
max.retries=10
retry.backoff.ms=3000

name=s3-sink-events
connector.class=io.confluent.connect.s3.S3SinkConnector
tasks.max=4

# 来源 Topic
topics=events.clickstream,events.payments

# S3 配置
s3.bucket.name=my-data-lake
s3.region=cn-north-1
s3.part.size=5242880

# 分区策略：按日期分区
partitioner.class=io.confluent.connect.storage.partitioner.TimeBasedPartitioner
path.format='year'=YYYY/'month'=MM/'day'=dd/'hour'=HH
partition.duration.ms=3600000
locale=zh-CN
timezone=Asia/Shanghai

# 文件格式
format.class=io.confluent.connect.s3.format.parquet.ParquetFormat
storage.class=io.confluent.connect.s3.storage.S3Storage

# 滚动策略：100MB 或 10分钟触发写入
flush.size=100000
rotate.schedule.interval.ms=600000

# Schema 集成
schema.compatibility=FULL

name=jdbc-sink-analytics
connector.class=io.confluent.connect.jdbc.JdbcSinkConnector
tasks.max=4

# 来源
topics=db.ecommerce.orders,db.ecommerce.users

# 目标数据库
connection.url=jdbc:postgresql://analytics:5432/warehouse
connection.user=warehouse_writer
connection.password=${file:/secrets/db-password.txt:password}

# 表名映射：自动按 Topic 名创建/映射表
table.name.format=${topic}

# 写入模式
insert.mode=upsert
pk.mode=record_key
pk.fields=id

# 自动创建/更新 DDL
auto.create=true
auto.evolve=true

# 批处理
batch.size=1000

# postgresql.conf 配置
wal_level = logical
max_replication_slots = 5
max_wal_senders = 5

CREATE USER debezium WITH REPLICATION LOGIN PASSWORD 'debezium_password';
GRANT CONNECT ON DATABASE ecommerce TO debezium;
GRANT USAGE ON SCHEMA public TO debezium;
GRANT SELECT ON ALL TABLES IN SCHEMA public TO debezium;

-- 为逻辑复制创建发布（PostgreSQL 10+）
CREATE PUBLICATION dbz_publication FOR TABLE orders, users, products;

# pg_hba.conf
host    replication     debezium        10.0.0.0/8              md5
host    ecommerce       debezium        10.0.0.0/8              md5

# my.cnf 配置
[mysqld]
server-id         = 1
log_bin           = mysql-bin
binlog_format     = ROW
binlog_row_image  = FULL
expire_logs_days  = 7

CREATE USER 'debezium'@'%' IDENTIFIED BY 'debezium_password';
GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'debezium'@'%';
FLUSH PRIVILEGES;

# my.cnf 启用 GTID
gtid_mode                 = ON
enforce_gtid_consistency  = ON

# MongoDB 需要以副本集模式运行
rs.initiate({
  _id: "rs0",
  members: [
    { _id: 0, host: "mongo1:27017" },
    { _id: 1, host: "mongo2:27017" },
    { _id: 2, host: "mongo3:27017" }
  ]
})

db.createUser({
  user: "debezium",
  pwd: "debezium_password",
  roles: [
    { role: "read", db: "ecommerce" },
    { role: "read", db: "config" },
    { role: "clusterMonitor", db: "admin" }
  ]
})

{
  "schema": { },
  "payload": {
    "before": {
      "id": 123,
      "status": "pending",
      "total": 199.99
    },
    "after": {
      "id": 123,
      "status": "paid",
      "total": 199.99
    },
    "source": {
      "version": "2.5.0.Final",
      "connector": "postgresql",
      "name": "ecommerce",
      "db": "ecommerce",
      "schema": "public",
      "table": "orders",
      "lsn": 245678912345,
      "txId": 567,
      "ts_ms": 1723800000000
    },
    "op": "u",
    "ts_ms": 1723800000123
  }
}

{
  "type": "record",
  "name": "OrderEvent",
  "namespace": "com.ecommerce.events",
  "fields": [
    { "name": "orderId", "type": "string" },
    { "name": "userId", "type": "long" },
    { "name": "totalAmount", "type": "double" },
    { "name": "status", "type": { "type": "enum", "name": "OrderStatus", "symbols": ["PENDING", "PAID", "SHIPPED", "CANCELLED"] } },
    { "name": "createdAt", "type": "long", "logicalType": "timestamp-millis" },
    { "name": "items", "type": { "type": "array", "items": {
      "type": "record",
      "name": "OrderItem",
      "fields": [
        { "name": "productId", "type": "long" },
        { "name": "quantity", "type": "int" },
        { "name": "price", "type": "double" }
      ]
    }}}
  ]
}

syntax = "proto3";
package ecommerce;

message OrderEvent {
  string order_id = 1;
  int64 user_id = 2;
  double total_amount = 3;
  OrderStatus status = 4;
  int64 created_at = 5;
  repeated OrderItem items = 6;
}

enum OrderStatus {
  PENDING = 0;
  PAID = 1;
  SHIPPED = 2;
  CANCELLED = 3;
}

message OrderItem {
  int64 product_id = 1;
  int32 quantity = 2;
  double price = 3;
}

# 查看当前兼容性策略
curl -X GET http://schema-registry:8081/config/db.ecommerce.orders-value

# 设置 FULL 兼容性
curl -X PUT http://schema-registry:8081/config/db.ecommerce.orders-value \
  -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  -d '{"compatibility": "FULL"}'

# 注册新 Schema（如果已存在则返回已有 ID）
curl -X POST http://schema-registry:8081/subjects/db.ecommerce.orders-value/versions \
  -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  -d '{
    "schema": "{\"type\":\"record\",\"name\":\"OrderEvent\",\"fields\":[{\"name\":\"orderId\",\"type\":\"string\"}]}"
  }'

# 返回：{"id": 42}

# 按 ID 查询 Schema
curl -X GET http://schema-registry:8081/schemas/ids/42

# 列出某 Subject 的所有版本
curl -X GET http://schema-registry:8081/subjects/db.ecommerce.orders-value/versions

# 查询最新版本
curl -X GET http://schema-registry:8081/subjects/db.ecommerce.orders-value/versions/latest

# 检查 Schema 兼容性（不注册）
curl -X POST http://schema-registry:8081/compatibility/subjects/db.ecommerce.orders-value/versions/latest \
  -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  -d '{
    "schema": "{...new schema...}"
  }'

version: '3.8'
services:
  postgres:
    image: postgres:16
    environment:
      POSTGRES_USER: ecommerce
      POSTGRES_PASSWORD: ecommerce_password
      POSTGRES_DB: ecommerce
    volumes:
      - ./init.sql:/docker-entrypoint-initdb.d/init.sql
    command: >
      postgres -c wal_level=logical
               -c max_replication_slots=5
               -c max_wal_senders=5

  zookeeper:
    image: confluentinc/cp-zookeeper:7.6.0
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181

  kafka:
    image: confluentinc/cp-kafka:7.6.0
    depends_on: [zookeeper]
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

  schema-registry:
    image: confluentinc/cp-schema-registry:7.6.0
    depends_on: [kafka]
    environment:
      SCHEMA_REGISTRY_HOST_NAME: schema-registry
      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: kafka:9092
      SCHEMA_REGISTRY_LISTENERS: http://0.0.0.0:8081

  connect:
    image: confluentinc/cp-kafka-connect:7.6.0
    depends_on: [kafka, schema-registry, postgres]
    environment:
      CONNECT_BOOTSTRAP_SERVERS: kafka:9092
      CONNECT_GROUP_ID: connect-cluster
      CONNECT_CONFIG_STORAGE_TOPIC: connect-configs
      CONNECT_OFFSET_STORAGE_TOPIC: connect-offsets
      CONNECT_STATUS_STORAGE_TOPIC: connect-status
      CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR: 1
      CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR: 1
      CONNECT_STATUS_STORAGE_REPLICATION_FACTOR: 1
      CONNECT_KEY_CONVERTER: io.confluent.connect.avro.AvroConverter
      CONNECT_KEY_CONVERTER_SCHEMA_REGISTRY_URL: http://schema-registry:8081
      CONNECT_VALUE_CONVERTER: io.confluent.connect.avro.AvroConverter
      CONNECT_VALUE_CONVERTER_SCHEMA_REGISTRY_URL: http://schema-registry:8081
      CONNECT_PLUGIN_PATH: /usr/share/java,/usr/share/confluent-hub-components
    volumes:
      - ./plugins:/usr/share/confluent-hub-components

# 在 Connect 容器中执行
confluent-hub install --no-prompt debezium/debezium-connector-postgresql:2.5.0
# 安装 S3 Sink 插件
confluent-hub install --no-prompt confluentinc/kafka-connect-s3:10.5.0

curl -X POST http://localhost:8083/connectors \
  -H "Content-Type: application/json" \
  -d '{
    "name": "debezium-postgres-source",
    "config": {
      "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
      "tasks.max": "2",

      "database.hostname": "postgres",
      "database.port": "5432",
      "database.user": "debezium",
      "database.password": "debezium_password",
      "database.dbname": "ecommerce",
      "database.server.name": "ecommerce",

      "plugin.name": "pgoutput",
      "slot.name": "debezium_slot",
      "publication.name": "dbz_publication",

      "table.include.list": "public.orders,public.order_items,public.users",
      "column.exclude.list": "public.users.password_hash",

      "snapshot.mode": "initial",
      "snapshot.fetch.size": "10000",

      "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,ts_ms,source.ts_ms",

      "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",
      "value.converter.schemas.enable": "true",

      "heartbeat.interval.ms": "10000",
      "heartbeat.action.query": "INSERT INTO public.debezium_heartbeat (id, ts) VALUES (1, NOW()) ON CONFLICT (id) DO UPDATE SET ts = NOW()",

      "tombstones.on.delete": "true",
      "decimal.handling.mode": "string",
      "time.precision.mode": "connect"
    }
  }'

INSERT INTO orders (user_id, total_amount, status, created_at)
VALUES (1001, 299.50, 'PENDING', NOW());

UPDATE orders SET status = 'PAID' WHERE id = 1;

DELETE FROM orders WHERE id = 1;

# 查看 Topic 列表
kafka-topics --bootstrap-server kafka:9092 --list
# 应看到：ecommerce.public.orders

# 消费 Avro 消息（使用 Schema Registry）
kafka-avro-console-consumer \
  --bootstrap-server kafka:9092 \
  --topic ecommerce.public.orders \
  --property schema.registry.url=http://schema-registry:8081 \
  --from-beginning

# 预期输出（INSERT 事件，经过 unwrap 转换后）
# {"orderId": 1, "userId": 1001, "totalAmount": "299.50", "status": "PENDING", "createdAt": 1723800000000, "__op": "c", "__ts_ms": 1723800000123}

# UPDATE 事件
# {"orderId": 1, "userId": 1001, "totalAmount": "299.50", "status": "PAID", "createdAt": 1723800000000, "__op": "u", "__ts_ms": 1723800001000}

# DELETE 事件（rewrite 模式）
# {"orderId": 1, "userId": 1001, "totalAmount": "299.50", "status": "PAID", "createdAt": 1723800000000, "__deleted": "true", "__op": "d", "__ts_ms": 1723800002000}

curl -X POST http://localhost:8083/connectors \
  -H "Content-Type: application/json" \
  -d '{
    "name": "s3-sink-orders",
    "config": {
      "connector.class": "io.confluent.connect.s3.S3SinkConnector",
      "tasks.max": "4",
      "topics": "ecommerce.public.orders,ecommerce.public.order_items",

      "s3.bucket.name": "my-data-lake-raw",
      "s3.region": "ap-northeast-1",
      "s3.part.size": "5242880",

      "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/'hour'=HH"'"'",
      "partition.duration.ms": "3600000",
      "locale": "zh-CN",
      "timezone": "Asia/Shanghai",

      "flush.size": "10000",
      "rotate.schedule.interval.ms": "600000",

      "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",

      "schema.compatibility": "FULL",
      "enhanced.avro.schema.support": "true"
    }
  }'

ALTER TABLE orders ADD COLUMN coupon_code VARCHAR(50);

UPDATE orders SET coupon_code = 'SUMMER2025' WHERE id = 2;