如何正确配置Kafka Connect S3 Sink的TimeBasedPartitioner？

首先，我来帮你梳理配置里的核心问题，给出正确的配置方案，再分享实用的调试方法。

一、核心配置修正 #

你想要用记录中的local_timestamp（UNIX时间戳）按小时分区，有几个关键配置需要调整和确认：

1. 时间戳单位的必加配置（如果是秒级时间戳） #

如果local_timestamp是秒级UNIX时间戳（比如1699999200对应2023-11-14 02:00:00），必须添加时间戳转换器配置——否则连接器会默认把它当成毫秒级时间戳，计算出的时间完全错误，分区路径自然不符合预期。添加以下两项：

"timestamp.convertor.class": "io.confluent.connect.storage.partitioner.TimestampConverter",
"timestamp.convertor.type": "Seconds"

如果你的时间戳已经是毫秒级，可以跳过这一步，但建议先确认字段值的单位。

2. 保留flush.size，它不干扰分区逻辑 #

你之前移除flush.size导致报错是因为它是必填项（无默认值），而且它和分区逻辑完全不冲突：

• flush.size控制单个文件刷新到S3的记录数
• 当时间跨越分区边界时（比如从23:59到00:00），即使当前文件的记录数没到flush.size，连接器也会自动关闭当前文件，切换到新的分区路径

3. 可选但推荐的配置 #

添加rotate.schedule.interval.ms并设置为和partition.duration.ms相同的值（3600000，即1小时），确保即使没有新数据流入，连接器也会按时间周期滚动文件，避免单个分区下的文件无限延迟写入。

二、完整的正确配置示例 #

{
  "name":"s3-sink",
  "config":{
    "connector.class":"io.confluent.connect.s3.S3SinkConnector",
    "tasks.max":"1",
    "topics":"xxxxx",
    "s3.region":"yyyyyy",
    "s3.bucket.name":"zzzzzzz",
    "s3.part.size":"5242880",
    "flush.size":"1000",
    "storage.class":"io.confluent.connect.storage.S3Storage",
    "format.class":"io.confluent.connect.s3.format.avro.AvroFormat",
    "schema.generator.class":"io.confluent.connect.storage.hive.schema.DefaultSchemaGenerator",
    "partitioner.class":"io.confluent.connect.storage.partitioner.TimeBasedPartitioner",
    "timestamp.extractor":"Record",
    "timestamp.field":"local_timestamp",
    "timestamp.convertor.class": "io.confluent.connect.storage.partitioner.TimestampConverter",
    "timestamp.convertor.type": "Seconds",
    "path.format":"YYYY-MM-dd-HH",
    "partition.duration.ms":"3600000",
    "schema.compatibility":"NONE",
    "rotate.schedule.interval.ms":"3600000"
  }
}

三、调试方法：排查问题的实用步骤 #

当分区不符合预期时，可以通过以下方式定位原因：

1. 查看连接器日志 #

Connect的日志通常在部署目录的logs/connect.log下（Docker部署用docker logs <connect-container>查看），搜索这些关键字：

• TimeBasedPartitioner：查看分区路径的生成日志，比如会输出类似Writing records to partition path: s3://zzzzzzz/xxxxx/2023-11-14-02/的内容，确认路径是否符合预期
• timestamp：查看日志中对local_timestamp字段的解析值，确认时间转换是否正确
• error：检查是否有时间戳解析失败的报错，比如字段类型不匹配（比如local_timestamp是字符串而非long）

2. 用REST API验证状态和配置 #

• 查看连接器运行状态：

  curl http://localhost:8083/connectors/s3-sink/status
  

  确认任务状态是RUNNING，没有报错信息
• 查看当前生效的配置：

  curl http://localhost:8083/connectors/s3-sink/config
  

  确保所有配置项都正确应用，没有被意外覆盖

3. 测试单条明确的测试数据 #

生产一条带有已知local_timestamp的测试Avro记录，比如对应2023-11-14-02小时的时间戳（秒级是1699999200，毫秒级是1699999200000），然后观察S3中是否生成对应的分区路径，快速验证配置是否生效。

4. 检查Avro Schema #

确认local_timestamp字段的类型是long（UNIX时间戳通常用long类型存储），如果是字符串或其他类型，时间戳提取器可能无法正确解析，导致分区失败。

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