如何让AWS MSK连接器从过去指定时间戳开始读取消息？

完全可以实现类似Kinesis AT_TIMESTAMP的初始消费位置功能，具体配置方式取决于你使用的MSK S3 Parquet连接器类型，且需在Terraform的连接器配置块中设置相关参数。

核心配置参数 #

针对MSK的S3 Sink Connector（无论是Confluent版本还是AWS官方版本），需要配置两个关键参数来实现指定时间戳启动消费：

• auto.offset.reset：设置为timestamp，告诉连接器以时间戳作为初始偏移量的判断依据
• offset.reset.timestamp：传入毫秒级Unix时间戳，指定连接器开始消费的具体时间点

Terraform中的配置位置 #

在Terraform定义aws_mskconnect_connector资源的configuration块中添加这两个参数即可，示例配置片段如下：

resource "aws_mskconnect_connector" "s3_parquet_sink" {
  name               = "s3-parquet-converter"
  kafka_cluster {
    apache_kafka_cluster {
      bootstrap_servers = aws_msk_cluster.my_msk_cluster.bootstrap_brokers_sasl_ssl
      vpc {
        security_groups = [aws_security_group.msk_connect_sg.id]
        subnets         = aws_subnet.private.*.id
      }
    }
  }

  kafka_connect_version = "2.7.1"
  capacity {
    auto_scaling {
      mcu_count       = 1
      min_capacity    = 1
      max_capacity    = 2
      scale_in_policy {
        cpu_utilization_percentage = 20
      }
      scale_out_policy {
        cpu_utilization_percentage = 80
      }
    }
  }

  configuration = jsonencode({
    "connector.class"          = "io.confluent.connect.s3.S3SinkConnector"
    "tasks.max"                = "1"
    "topics"                   = "your_target_topic"
    "s3.bucket.name"           = "your-s3-bucket-name"
    "s3.region"                = "us-east-1"
    "format.class"             = "io.confluent.connect.s3.format.parquet.ParquetFormat"
    "storage.class"            = "io.confluent.connect.s3.storage.S3Storage"
    "key.converter"            = "org.apache.kafka.connect.storage.StringConverter"
    "value.converter"          = "io.confluent.connect.avro.AvroConverter"
    "value.converter.schema.registry.url" = "http://your-schema-registry:8081"
    # 关键配置：指定从目标时间戳开始消费
    "auto.offset.reset"        = "timestamp"
    "offset.reset.timestamp"   = "1620000000000" # 替换为你需要的毫秒级Unix时间戳
  })

  service_execution_role_arn = aws_iam_role.msk_connect_exec_role.arn
}

注意事项 #

1. 版本兼容性：确保使用的连接器版本支持该特性，Confluent S3 Connector 5.4.0及以上版本开始支持auto.offset.reset=timestamp参数
2. 偏移量重置：如果连接器已创建并存在历史消费组偏移量，需先重置消费组偏移量或删除原有消费组，否则新配置不会生效。可通过Kafka命令行工具执行：

   kafka-consumer-groups.sh --bootstrap-server <msk-bootstrap-servers> --reset-offsets --to-timestamp <target-timestamp-ms> --topic <your-topic> --group <connector-consumer-group> --execute
   

3. 时间戳格式：必须传入毫秒级时间戳，若你手中是秒级时间戳，需要乘以1000转换

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