カフカストリームで消費されるメッセージごとにオフセットを取得する方法はありますか？

Question

KAFKA STREAMSが殺されたときに処理され、コミットされなかったメッセージを読むことを避けるために、キーと値とともに各メッセージのオフセットを取得して、どこかに保存して使用できるようにしたい既に処理されたメッセージの再処理を避けます。

Answer 1

はい、可能です。 FAQの項目http://docs.confluent.io/current/streams/faq.html#accessing-record-metadata-such-as-topic-partition-and-offset-informationを参照してください。そのような話題、パーティションとして

へのアクセス記録のメタデータ、および情報をオフセット：

は、私は、以下のキー情報をコピー＆ペーストうか？

レコードメタデータはProcessor APIからアクセスできます。 Processor API integrationのおかげで、DSLによって間接的にアクセスすることもできます。

プロセッサAPIでは、 ProcessorContextからレコードメタデータにアクセスできます。 Processor#init()の間にプロセッサの インスタンスフィールドにコンテキストの参照を格納してから、内のプロセッサのコンテキストを問い合わせる 、たとえば （Transformerと同じ）をクエリできます。現在処理されているレコードが に一致するようにコンテキストが自動的に更新されます。つまり、のようなメソッド は常に最新の レコードのメタデータを返します。プロセッサ のコンテキストをpunctuate()に呼び出すときには、いくつかの注意事項が適用されます。詳細については、Javadocsを参照してください。

カスタムTransformerと組み合わせるDSLを使用する場合は、例えば、 あなたもパーティション を含めるとメタデータを相殺するために入力レコードの値を変えることができ、そのようなmapや filterとして、その後のDSL操作は、この情報を活用することができ。