使用Confluent Kafka .NET Client时如何自定义消息发送的时间戳

Confluent.Kafka 客户端的 Message<TKey, TValue> 原生支持自定义时间戳配置，直接修改发送代码新增Timestamp属性赋值即可，修改后示例如下：

var kp = new kafkaDatapoint
{
   value = p.Value,
   tags = p.tags
};
// 此处替换为你需要的自定义时间，支持传入DateTime、DateTimeOffset或Unix时间戳毫秒数
var customTimestamp = new Timestamp(DateTimeOffset.Now.AddDays(-1));
await _producer.ProduceAsync(p.Variable, new Message<string, kafkaDatapoint>()
                  {
                      Key = p.Variable,
                      Value = kp,
                      Timestamp = customTimestamp // 新增此行覆盖默认自动生成的时间戳
                  });

注意事项 #

• 若自定义时间戳不生效，请检查对应Kafka Topic的配置项message.timestamp.type：
  • 配置为CreateTime时，自定义时间戳会正常生效（大部分场景默认为此配置
  • 配置为LogAppendTime时，消息时间戳会被Broker强制覆盖为写入日志的时间，自定义赋值不会生效
• Timestamp构造函数支持三种入参形式：DateTime、DateTimeOffset、Unix时间戳毫秒数，可按需选择使用

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