5

我正在使用 Kafka Streams 开发 PoC。现在我需要获取流消费者中的偏移值，并使用它(topic-offset)->hash为每条消息生成一个唯一键。原因是：生产者是 syslog，只有少数几个有 ID。我无法在消费者中生成 UUID，因为在重新处理的情况下我需要重新生成相同的密钥。

我的问题是：org.apache.kafka.streams.processor.ProcessorContext该类公开了一个.offset()返回值的方法，但是我使用的是 KStream 而不是处理器，并且找不到返回相同内容的方法。

有人知道如何从 KStream 中提取每一行的消费者价值吗？提前致谢。

4

1 回答 1

6

process(...)您可以通过、transform(...)和使用混合匹配的 DSL 和处理器 API transformValues(...)。

它允许您访问类似于普通处理器 API 的当前记录偏移量。在您的情况下，您似乎想使用KStream#transform(...).

于 2016-11-28T06:03:31.180 回答