如何更改记录的时间戳？

Question

我正在使用FluentD（v.12最后一个稳定版本）向Kafka发送消息。但是FluentD正在使用一个旧的KafkaProducer，所以记录时间戳总是设置为-1。因此，我必须使用WallclockTimestampExtrator将记录的时间戳设置为消息到达kafka时的时间点。

是否有特定于Kafka Streams的解决方案？

我真的感兴趣的时间戳，是由Fluentd在消息中发送的：

“时间戳”：“1507885936”，“主机”：“V.X.Y.Z.”

Kafka的记录表示：

offset=0，timestamp=-1，key=null，value={“timestamp”：“1507885936”，“host”：“V.X.Y.Z.”

我想在Kafka有这样的记录：

offset=0，timestamp=1507885936，key=null，value={“timestamp”：“1507885936”，“host”：“V.X.Y.Z.”

我的解决方案如下：

>

写一个消费者提取时间戳（https://kafka.apache.org/0110/javadoc/org/apache/kafka/streams/processor/TimestampExtractor.html）

编写一个生产者，用时间戳集生成一个新记录（ProducerRecord（字符串主题、整数分区、长时间戳、K键、V值）

如果有的话，我更喜欢Kafka斯特伦的解决方案。

Answer 1

您可以编写一个非常简单的Kafka Streams应用程序，如：

KStreamBuilder builder = new KStreamBuilder();
builder.stream("input-topic").to("output-topic");

并使用自定义的时间戳提取器配置应用程序，从记录中提取时间戳并返回它。

Kafka流将在将记录写回Kafka时使用返回的时间戳。

注意：如果你有无序数据——也就是说，时间戳没有严格排序——结果也会包含无序时间戳。Kafka Streams使用返回的时间戳写回Kafka（也就是说，无论提取器返回什么，都被用作记录元数据时间戳）。注意，在写入时，当前处理的输入记录中的时间戳用于所有生成的输出记录——这在1.0版中保持不变，但在未来的版本中可能会更改。）。

更新：

通常，可以通过处理器API修改时间戳。调用上下文。forward（）您可以通过将输出记录时间戳设置为。全部（）。带时间戳（…）作为forward（）的参数。