如何在Confluent 6.2.0环境下运行Kafka S3 Sink Connector

以下操作基于你已完成3节点Confluent 6.2.0部署、S3连接器插件全节点部署、quickstart-s3.properties配置修改的前提执行：

1. 前置校验：确认Kafka Connect集群运行正常 #

S3连接器运行在Kafka Connect集群之上，需先确保所有节点的Connect服务处于正常运行状态：

• 若为systemd部署的生产环境，所有节点执行：systemctl status confluent-kafka-connect，确认服务状态为active (running)
• 若为本地测试环境，所有节点执行：confluent local services connect status，确认服务状态正常

2. 校验配置文件合法性 #

避免配置错误导致启动失败，先在任意节点执行配置校验：

confluent local services connect connector config validate --config /path/to/your/quickstart-s3.properties

返回结果无报错则说明配置合法，可继续后续操作。

3. 启动S3连接器 #

根据你的部署场景选择对应启动方式：

方式1：REST API启动（生产环境分布式集群推荐） #

3节点分布式集群建议通过统一的REST接口管理连接器，操作如下：

1. 将quickstart-s3.properties中的配置转换为JSON请求体，保存为s3-connector.json，示例结构：

{
  "name": "s3-sink-connector",
  "config": {
    "connector.class": "io.confluent.connect.s3.S3SinkConnector",
    "tasks.max": "3",
    // 此处直接复制你properties文件中的所有配置项，key和value一一对应即可
  }
}

2. 向任意Connect节点发送启动请求：

curl -X POST -H "Content-Type: application/json" --data @/path/to/s3-connector.json http://<Connect节点IP>:8083/connectors

方式2：Confluent CLI启动（本地测试场景使用） #

仅适用于本地测试环境，在任意节点执行如下命令即可加载启动连接器：

confluent local services connect connector load s3-sink-connector --config /path/to/your/quickstart-s3.properties

4. 验证启动结果 #

执行如下命令查看连接器运行状态：

curl http://<Connect节点IP>:8083/connectors/s3-sink-connector/status

返回结果中connector.state为RUNNING，且所有子任务的state也为RUNNING，说明连接器启动成功。

常见异常排查 #

• 插件找不到报错：检查所有Connect节点的plugin.path配置是否包含S3连接器的部署目录，且目录权限为Connect服务运行用户可读
• S3权限报错：确认配置中填写的AWS访问密钥对有效，且对应账号拥有目标S3桶的读写、列举权限
• 任务启动失败：查看Connect服务日志 /var/log/confluent/kafka-connect/connect.log 定位具体错误原因

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