4. Kafka Connect 集成：连接器配置与生命周期管理

Kafka Connect 是 Kafka 生态里一个非常实用的组件。说白了，它就是帮你把数据从外部系统搬进 Kafka（Source），或者从 Kafka 搬到外部系统（Sink）。对于 CDC 场景来说，我们通常用 Source Connector 去抓取数据库的变更日志，再用 Sink Connector 写到目标端。这一章，我带你过一遍它的架构、配置和日常管理。

4.1 Kafka Connect 架构概览

Kafka Connect 有两种运行模式：单机模式（Standalone）和分布式模式（Distributed）。我个人习惯在生产环境用分布式模式，因为它支持水平扩展和故障转移。

它的核心组件包括：

• Connector：定义数据从哪里来、到哪里去。它是一个逻辑概念，负责管理任务的分配。
• Task：实际干活的工作单元。一个 Connector 可以拆成多个 Task，并行处理数据。
• Worker：运行 Connector 和 Task 的进程。分布式模式下，多个 Worker 组成一个集群。
• Converters：负责数据格式的转换，比如从 Avro 转成 JSON，或者从 JSON 转成 Protobuf。
• Offset Storage：记录数据读取的进度。Source Connector 用它来保证“断点续传”。

下面这张图展示了 Kafka Connect 在 CDC 场景下的整体架构：

4.2 Source Connector 配置

Source Connector 负责从外部系统拉取数据。在 CDC 场景下，最常用的是 Debezium 系列连接器。配置 Source Connector 时，有几个关键参数必须搞清楚。

下面是一个典型的 Debezium MySQL Source Connector 配置示例：

{
  "name": "mysql-connector-01",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "database.hostname": "192.168.1.100",
    "database.port": "3306",
    "database.user": "cdc_user",
    "database.password": "cdc_pass",
    "database.server.id": "223344",
    "database.server.name": "prod-db-01",
    "database.include.list": "inventory",
    "table.include.list": "inventory.orders,inventory.customers",
    "database.history.kafka.bootstrap.servers": "kafka1:9092,kafka2:9092",
    "database.history.kafka.topic": "schema-changes.inventory",
    "include.schema.changes": "true",
    "snapshot.mode": "initial",
    "offset.storage.topic": "connect-offsets",
    "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": "false"
  }
}

这里有几个参数，我特别想强调一下：

• database.server.name：这个值会作为 Kafka Topic 的前缀。比如你设成 prod-db-01，那么 Topic 名就是 prod-db-01.inventory.orders。命名要规范，不然以后找数据很头疼。
• snapshot.mode：决定连接器启动时是否做全量快照。我建议第一次用 initial，后续重启用 schema_only 或 when_needed，避免重复拉取全量数据。
• offset.storage.topic：记录读取进度。如果这个 Topic 丢了，连接器会从头开始读，后果很严重。我在项目中遇到过因为误删这个 Topic 导致数据重复消费的情况，嗯，从那以后我都是把它设置成 min.insync.replicas=2 来保护。

4.3 Sink Connector 配置

Sink Connector 负责把 Kafka 里的数据写到外部系统。配置思路和 Source 类似，但侧重点不同。Sink 更关心“怎么写”和“写到哪里”。

下面是一个 JDBC Sink Connector 的配置示例：

{
  "name": "jdbc-sink-connector-01",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "connection.url": "jdbc:postgresql://pg-host:5432/target_db",
    "connection.user": "sink_user",
    "connection.password": "sink_pass",
    "topics": "prod-db-01.inventory.orders",
    "table.name.format": "orders",
    "insert.mode": "upsert",
    "pk.fields": "id",
    "pk.mode": "record_key",
    "auto.create": "true",
    "auto.evolve": "true",
    "batch.size": "3000",
    "max.retries": "3",
    "retry.backoff.ms": "5000",
    "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": "false"
  }
}

配置 Sink Connector 时，有几个坑需要注意：

• insert.mode：我建议用 upsert 模式。这样如果目标表有主键冲突，会执行更新而不是报错。用 insert 模式的话，重复数据会导致任务失败。
• auto.create 和 auto.evolve：这两个参数很实用，但也很危险。自动建表省事，但如果源端 Schema 频繁变更，目标表结构可能会乱掉。我曾经在生产环境吃过这个亏——自动加了一列，结果类型不匹配，整个 Sink 任务挂了。后来我改成手动管理表结构，只在测试环境用自动模式。
• batch.size：控制每次写入的批大小。设太大容易 OOM，设太小吞吐量上不去。我个人习惯从 1000 开始调，观察 CPU 和内存使用率，逐步增加到 5000 左右。

4.4 连接器生命周期管理

连接器不是配好就完事了。日常运维中，我们需要管理它的整个生命周期：创建、暂停、恢复、更新、删除。Kafka Connect 提供了 REST API 来干这些事。

下面这张表总结了常用的 REST API 端点：

操作	HTTP 方法	端点	说明
创建连接器	POST	/connectors	提交 JSON 配置，创建新连接器
查看所有连接器	GET	/connectors	返回连接器名称列表
查看连接器状态	GET	/connectors/{name}/status	返回 RUNNING / PAUSED / FAILED 状态
暂停连接器	PUT	/connectors/{name}/pause	暂停数据读取或写入
恢复连接器	PUT	/connectors/{name}/resume	恢复数据流
更新连接器配置	PUT	/connectors/{name}/config	动态更新配置，无需重启
删除连接器	DELETE	/connectors/{name}	删除连接器及其任务

实际运维中，我经常用这几个命令：

# 查看连接器状态
curl -s http://localhost:8083/connectors/mysql-connector-01/status | jq .

# 暂停连接器（比如要维护目标数据库）
curl -X PUT http://localhost:8083/connectors/mysql-connector-01/pause

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

# 更新配置（比如修改 batch.size）
curl -X PUT -H "Content-Type: application/json" \
  http://localhost:8083/connectors/jdbc-sink-connector-01/config \
  -d '{"batch.size": 5000}'

这里有个细节：更新配置时，你只需要传要改的参数，不用把整个 JSON 都发过去。Kafka Connect 会合并新旧配置。但要注意，有些参数修改后需要重启连接器才能生效，比如 connector.class。而像 batch.size 这种参数，改了之后 Task 会自动重启，不需要手动干预。