SpringBoot项目中Kafka Streams提取消息时间戳方法咨询

方法1：使用带RecordContext的map操作 #

Kafka Streams的KStream提供了支持访问记录上下文的map/flatMap重载方法，直接从中就能获取消息时间戳，无需依赖原生Consumer的方法。

假设你当前的基础转储逻辑是：

KStream<String, String> inputStream = streamsBuilder.stream("input-topic");
inputStream.to("output-topic");

修改为带上下文的map操作，将时间戳注入到JSON中：

import org.apache.kafka.streams.kstream.ValueMapperWithContext;
import org.apache.kafka.streams.processor.RecordContext;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;

ObjectMapper objectMapper = new ObjectMapper();

KStream<String, String> inputStream = streamsBuilder.stream("input-topic");

inputStream.map((key, value, context) -> {
    // 获取毫秒级消息时间戳
    long timestamp = context.timestamp();
    
    try {
        // 解析原始JSON，注入时间戳后重新序列化
        ObjectNode jsonNode = (ObjectNode) objectMapper.readTree(value);
        jsonNode.put("messageTimestamp", timestamp);
        return KeyValue.pair(key, objectMapper.writeValueAsString(jsonNode));
    } catch (Exception e) {
        // JSON解析失败时返回原始消息，避免流任务中断
        return KeyValue.pair(key, value);
    }
}).to("output-topic");

方法2：自定义Processor（复杂场景适用） #

如果你的消息处理逻辑更复杂，比如需要多步骤转换，可以用Processor API自定义处理器，直接从ProcessorContext中获取时间戳：

1. 实现自定义Processor：

import org.apache.kafka.streams.processor.Processor;
import org.apache.kafka.streams.processor.ProcessorContext;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;

public class TimestampInjectProcessor implements Processor<String, String> {
    private ProcessorContext context;
    private ObjectMapper objectMapper;

    @Override
    public void init(ProcessorContext context) {
        this.context = context;
        this.objectMapper = new ObjectMapper();
    }

    @Override
    public void process(String key, String value) {
        long timestamp = context.timestamp();
        try {
            ObjectNode jsonNode = (ObjectNode) objectMapper.readTree(value);
            jsonNode.put("messageTimestamp", timestamp);
            // 将处理后的消息转发到下游
            context.forward(key, objectMapper.writeValueAsString(jsonNode));
        } catch (Exception e) {
            // 异常场景下转发原始消息
            context.forward(key, value);
        }
    }

    @Override
    public void close() {
        // 可在此清理资源，比如关闭ObjectMapper（如果需要）
    }
}

2. 在StreamsBuilder中集成该Processor：

streamsBuilder.stream("input-topic")
    .process(() -> new TimestampInjectProcessor())
    .to("output-topic");

注意事项 #

• 时间戳格式：context.timestamp()返回的是毫秒级Unix时间戳，如果需要格式化的时间字符串（如yyyy-MM-dd HH:mm:ss），可以在注入时用DateTimeFormatter做转换。
• 时间戳来源：默认是生产者发送时间，若Topic配置了使用Broker接收时间，这里拿到的就是Broker接收消息的时间。
• 异常防护：必须处理JSON解析、序列化失败的情况，避免单个坏消息导致整个流任务崩溃。

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