有沒有一種方法可以獲得在kafka流中消費的每條消息的偏移量？

Question

爲了避免讀取KAFKA STREAMS被殺死時處理但未提交的消息，我希望獲得每個消息的偏移以及鍵和值，以便我可以將它存儲在某處並將其用於避免重新處理已處理的消息。

Answer 1

是的，這是可能的。請參閱http://docs.confluent.io/current/streams/faq.html#accessing-record-metadata-such-as-topic-partition-and-offset-information的FAQ條目。

我將在下面的關鍵信息複製粘貼：

訪問記錄元數據，如主題，分區和偏移信息？

記錄元數據可通過Processor API訪問。 它也可以通過DSL間接獲得，這要感謝它的 Processor API integration。

使用Processor API，您可以通過 ProcessorContext訪問記錄元數據。您可以在Processor#init()存儲在你的處理器的 實例字段的上下文的引用，然後 查詢中Processor#process()處理器方面，例如 （同爲Transformer）。上下文自動更新以匹配 當前正在處理的記錄，這意味着方法 （如ProcessorContext#partition()）總是返回當前的 記錄的元數據。在punctuate()內調用處理器 上下文時，有些注意事項適用，請參閱Javadocs以瞭解詳細信息。

如果使用DSL與自定義Transformer組合，例如， 你可以改變輸入記錄的值也包括分區 和偏移元數據，以及諸如map或 隨後的DSL業務filter可以再利用這些信息。