Kafka Connect配置方式验证及分布式高可用集群搭建咨询

1. 使用properties文件的配置方式是否正确？ #

完全正确。Kafka Connect原生支持两种配置方式：properties文件和环境变量，两者都是官方认可的有效方式，各有适用场景：

• 你当前通过env.sh生成worker.properties再COPY到镜像的流程是合理的，适合需要固化大部分配置、同时通过CI/CD注入动态参数的场景。
• 环境变量方式（比如CONNECT_BOOTSTRAP_SERVERS）更适合动态注入敏感信息（如密码）或快速切换环境配置，但本质上是Kafka Connect会将环境变量转换为对应的properties参数。

优化建议：

• 敏感信息如SASL密码可以直接通过环境变量传递，避免写入worker.properties文件（比如用CONNECT_SASL_JAAS_CONFIG环境变量覆盖），减少敏感数据泄露风险。
• 确保env.sh中生成的配置参数没有冲突（比如你同时在worker.properties和Docker ENV里设置了kafka.jmx.port和kafka.opts，建议统一到一处）。

2. 从standalone模式切换到distributed模式搭建高可用集群 #

要搭建至少3个worker的高可用Kafka Connect集群，需按以下步骤调整：

步骤1：修改启动命令 #

将Dockerfile中的启动命令从connect-standalone改为connect-distributed：

CMD connect-distributed /etc/kafka-connect/worker.properties

步骤2：调整worker.properties配置 #

编辑env.sh，修改/新增以下配置：

• 移除standalone专属配置：删除offset.storage.file.filename=/tmp/offset.txt，因为distributed模式下偏移量会存储在Kafka的offset.storage.topic中，不再依赖本地文件。
• 确保集群一致性配置：所有worker实例的以下参数必须完全一致：

  group.id=sv-connect  # 同一集群的worker必须使用相同的group.id
  bootstrap.servers=$KAFKA_BOOTSTRAP_SERVERS
  config.storage.topic=sv-connect-configs
  offset.storage.topic=sv-connect-offsets
  status.storage.topic=sv-connect-status
  config.storage.replication.factor=3  # 需与Kafka集群副本数匹配，至少3
  offset.storage.replication.factor=3
  status.storage.replication.factor=3
  plugin.path=/usr/share/java,/etc/kafka-connect/jars,/usr/share/confluent-hub-components  # 所有worker必须有相同的插件路径和插件
  

• 优化存储主题配置：
  • config.storage.topic必须是单分区、多副本的主题（Kafka Connect会自动创建，但建议提前创建确保配置正确）。
  • offset.storage.topic的分区数建议设置为worker数量的2-3倍（比如3个worker设为6分区），提升并行处理能力。

步骤3：提前创建Kafka存储主题（可选但推荐） #

虽然Kafka Connect会自动创建三个存储主题，但提前创建可以确保配置符合要求：

# 创建config主题（单分区，3副本）
kafka-topics --create --topic sv-connect-configs --bootstrap-server $KAFKA_BOOTSTRAP_SERVERS --partitions 1 --replication-factor 3 --config cleanup.policy=compact

# 创建offset主题（6分区，3副本）
kafka-topics --create --topic sv-connect-offsets --bootstrap-server $KAFKA_BOOTSTRAP_SERVERS --partitions 6 --replication-factor 3 --config cleanup.policy=compact

# 创建status主题（3分区，3副本）
kafka-topics --create --topic sv-connect-status --bootstrap-server $KAFKA_BOOTSTRAP_SERVERS --partitions 3 --replication-factor 3 --config cleanup.policy=compact

步骤4：在ECS-Ec2部署多个worker实例 #

• 在ECS中创建至少3个Kafka Connect任务实例，确保所有实例使用同一镜像（插件一致）、同一配置参数。
• ECS服务配置中，设置任务数为3，启用自动扩展（可选）。
• 端口映射方面，由于是集群模式，每个worker的8083端口可以使用ECS的动态端口映射，或者通过负载均衡统一暴露REST接口（客户端可以访问任意worker的REST接口提交连接器配置，集群会自动同步）。

步骤5：验证集群状态 #

启动所有worker后，访问任意worker的/connectors接口，提交一个连接器配置，检查所有worker是否都能同步任务：

curl -X POST http://<worker-ip>:8083/connectors -H "Content-Type: application/json" -d '{
  "name": "test-connector",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    # 其他连接器配置...
  }
}'

# 查看任务分配情况
curl http://<worker-ip>:8083/connectors/test-connector/tasks

内容的提问来源于stack exchange，提问作者Pablo Marques