如何正确重启Kafka S3 Sink Connector并避免重复写入旧数据？

我结合自己使用Confluent Kafka Connect和S3 Sink Connector的实战经验，来逐一解答你的问题：

1. Connect重启REST API会重置偏移量吗？ #

默认情况下，无论是重启整个连接器的POST /connectors/<name>/restart，还是重启单个任务的PUT /connectors/<name>/tasks/<id>/restart，都不会重置偏移量。

Kafka Connect会将任务的消费偏移量持久化到内部的Kafka主题connect-offsets中（默认名称，可通过offset.storage.topic配置修改）。重启操作只会让任务从上次成功提交的偏移量位置继续消费，不会主动重置或回滚偏移量。

你遇到的重启后重写5月3日旧数据的情况，大概率是因为：

• 任务崩溃时，已经处理了部分数据但未成功提交偏移量到connect-offsets；
• 少数情况下，connect-offsets主题中的偏移量记录损坏，导致任务从更早的位置开始消费。

2. 正确重启失败的连接器任务的方式是什么？ #

推荐的步骤如下：

• 第一步：定位故障根源：先查看连接器任务的日志，明确是AWS服务异常、权限问题、资源不足还是连接器本身的bug导致崩溃，避免重复踩坑。
• 第二步：尝试重启单个任务：如果只有单个任务失败，优先调用PUT /connectors/s3sink/task/0/restart，这种方式只会重启指定任务，不会影响其他连接器或任务。
• 第三步：重启整个连接器：如果单个任务重启无效，或者多个任务都失败，再调用POST /connectors/s3sink/restart，重启该连接器下的所有任务。
• 可选：暂停-重启-恢复：如果担心重启过程中出现数据混乱，可以先暂停连接器：PUT /connectors/s3sink/pause，重启完成后再恢复：PUT /connectors/s3sink/resume，不过重启API本身会自动处理任务的启停状态。

另外，如果确认是偏移量异常导致的重复写入，可以手动修改connect-offsets主题中的对应记录，但操作前一定要备份，避免数据丢失。

3. K8s环境下应删除Pod还是调用REST /task/0/restart？ #

这两种方式适用场景不同，推荐优先用REST API：

• 优先使用REST API重启任务：如果只是单个任务失败，调用PUT /connectors/s3sink/task/0/restart更精准，只会重启目标任务，不会影响同一Connect Worker上的其他任务，操作更轻量。
• 删除Pod的适用场景：当Connect Worker进程本身崩溃（比如Pod OOM、容器进程挂掉），或者REST API无法正常访问时，可以删除对应的Pod。K8s会自动重新拉起Pod，Connect集群会重新分配该Worker上的所有任务。由于偏移量存在connect-offsets主题中，删除Pod不会丢失偏移量，但会影响该Worker上的所有任务，属于更“重”的操作。

4. 何时应使用/connectors/s3sink/restart？ #

这个API适用于以下场景：

• 该连接器下的所有任务都失败，需要批量重启所有任务时；
• 修改了连接器的全局配置（比如调整S3存储路径、修改批量写入参数）后，需要让所有任务加载新配置时（不过修改配置的PUT /connectors/s3sink/config通常会自动触发重启）；
• 单个任务多次重启无效，怀疑是连接器层面的全局状态异常（比如配置加载缓存、连接器内部资源泄漏）时。

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