Kafka流：如何更改记录的时间戳（0.11.0）？

Question

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

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

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

Kafka的唱片表现：

偏移量=0，时间戳=-1，键=空，值={“时间戳”：“1507885936”，“主机”：“V.X.Y.Z.”}

我的解决方法类似于：-编写一个使用者来提取时间戳(https://kafka.apache.org/0110/javadoc/org/apache/kafka/streams/processor/timestampextractor.html)

• 编写一个生成器，以生成一个设置了时间戳的新记录（ProducerRecord（字符串主题、整数分区、长时间戳、K键、V值）

我更喜欢KafkaStreams的解决方案，如果有的话。

Answer 1

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

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

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

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

注意：如果您有无序数据--即时间戳没有严格排序--结果也会包含无序的时间戳。Kafka Streams使用返回的时间戳写回Kafka（即，提取器返回的任何时间戳都用作记录元数据时间戳）。请注意，在写入时，来自当前处理的输入记录的时间戳将用于所有生成的输出记录--这对1.0版是有效的，但在以后的版本中可能会改变。）。