Kafka Streams多Topic流转场景下，如何获取消息发布到Topic L的时间戳？

嗨，我来帮你搞定这个问题！你现在拿到的是Topic A的时间戳，核心原因是Kafka默认的消息时间戳是创建时间（CreateTime）——也就是消息第一次被发送到Topic A时的时间，这个时间会跟着消息在各个Topic间流转，所以不管消费哪个Topic，默认拿到的都是最初的那个时间。

下面给你两种可行的解决方案，根据你的场景选就行：

方案一：手动添加自定义到达时间戳（推荐） #

如果能控制中间的Kafka Streams处理逻辑，这是最灵活精准的方式：在消息准备发送到Topic L的瞬间，把当前时间作为自定义字段（可以放在消息headers或者value里）一起发送，这样消费Topic L时直接读取这个字段就是到达时间了。

举个代码例子：

在Streams处理器中添加时间戳到headers #

// 假设这是你处理完消息、准备发送到Topic L的逻辑
public void process(String key, String value) {
    // 记录当前时间作为到达Topic L的时间
    long arrivalTimeAtL = System.currentTimeMillis();
    // 创建headers并添加时间戳
    RecordHeaders headers = new RecordHeaders();
    headers.add("arrival-time-at-L", String.valueOf(arrivalTimeAtL).getBytes());
    // 发送到Topic L时带上headers
    context.forward(key, value, To.child("Topic-L").withHeaders(headers));
}

消费Topic L时读取时间戳 #

while (iterator.hasNext()){
    ConsumerRecord<String,String> consumerRecord = iterator.next();
    // 获取Topic A的创建时间（原来的时间戳）
    long createTimeAtA = consumerRecord.timestamp();
    // 获取自定义的到达Topic L的时间
    Header arrivalHeader = consumerRecord.headers().lastHeader("arrival-time-at-L");
    if (arrivalHeader != null) {
        long arrivalTimeAtL = Long.parseLong(new String(arrivalHeader.value()));
        // 计算总耗时
        long totalDuration = arrivalTimeAtL - createTimeAtA;
        // 这里就可以统计耗时啦
        map.put(consumerRecord.key(), totalDuration);
    }
}

方案二：修改Topic L的时间戳类型为LogAppendTime #

如果没法修改代码，也可以让Kafka Broker自动给Topic L的消息打上“写入时间戳”：把Topic L的message.timestamp.type配置改成LogAppendTime，这样消息被Broker写入Topic L日志时，会自动用当前Broker时间作为消息的timestamp，消费时consumerRecord.timestamp()拿到的就是这个时间。

你可以用Kafka的命令行工具修改Topic配置：

kafka-topics.sh --alter --topic Topic-L --bootstrap-server your-broker-address:9092 --config message.timestamp.type=LogAppendTime

⚠️ 注意：这个时间是Broker写入日志的时间，和消息实际被发送到L的时间会有一点点延迟，但大多数场景下足够满足统计需求。

内容的提问来源于stack exchange，提问作者Karunakar Reddy L