如何检测Kafka-Connect的刷盘操作完成状态？

一、如何确认刷盘操作已完成 #

• 检查连接器状态与核心指标
  调用Kafka Connect的REST接口获取关键信息：
  • 执行 curl -X GET http://<connect-host>:<port>/connectors/<your-sink-name>/status，确认所有任务处于RUNNING状态，无失败、暂停异常。
  • 执行 curl -X GET http://<connect-host>:<port>/connectors/<your-sink-name>/metrics，重点关注s3-sink-files-completed（已完成的刷盘文件数）、s3-sink-records-committed（已提交到S3的记录总数），结合业务预期的记录量、文件数做对比，匹配则说明刷盘完成。
• 验证S3文件状态
  S3 Sink连接器会先写入带临时后缀（如.part）的文件，刷盘完成后才会移除临时后缀：
  • 仅统计无临时后缀的文件，这类文件是最终完成刷盘的产物。
  • 查看文件的LastModified时间，确认是否符合你配置的10分钟间隔规则，排查是否存在延迟生成的文件。
• 核对偏移量提交记录
  Kafka Connect的任务偏移量存储在内部主题connect-offsets中，通过Kafka命令行工具消费该主题：

  kafka-console-consumer.sh --bootstrap-server <kafka-broker>:9092 --topic connect-offsets --from-beginning --property print.key=true
  

  找到对应连接器的偏移量记录，其中的timestamp字段对应刷盘完成后的偏移提交时间，若该时间与预期刷盘时间匹配，说明刷盘已完成。

二、刷盘延迟的排查与处理 #

• 优先查看连接器运行日志，搜索是否存在S3连接超时、Kafka分区消费滞后、内存不足等报错信息，定位延迟根源。
• 通过指标s3-sink-record-write-latency查看单条记录写入S3的耗时，判断延迟来自Kafka消费环节还是S3写入环节：
  • 若消费滞后，可增加连接器的任务并行度（调整tasks.max配置）；
  • 若S3写入慢，可检查S3区域网络链路，或临时调大batch.size减少请求频次。
• 若延迟导致某周期的刷盘未按时完成，可通过S3文件的生成时间和偏移量记录，确认该周期的数据是否最终完成写入，避免遗漏。

三、查看刷盘历史与上次完成时间 #

• 从S3文件列表追溯：将S3目标路径下的文件按LastModified降序排序，最后一个无临时后缀的文件的时间即为上次刷盘完成时间；若需要完整历史，可导出文件列表的时间信息做统计。
• 从连接器日志提取：连接器日志中会输出类似Completed file s3://<bucket>/<path>/<file-name>的日志条目，搜索该关键词可获取每次刷盘的完成时间和对应文件。
• 从偏移量主题解析：消费connect-offsets主题中对应连接器的记录，每条记录的timestamp代表该次刷盘完成后的偏移提交时间，按时间排序即可得到完整的刷盘历史时间线。

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