registerSeekCallback（）を使用してSpring Kafkaでシーク操作を行う

Question

org.springframework.kafka.listener.ConcurrentMessageListenerContainerには、トピックのN個のパーティションをリスンするN MessageListenerがあります。場合によっては、最後にコミットされたオフセットとは異なるオフセットでアプリケーションを開始する必要があります。 私はリスナーをConsumerSeekAwareとし、registerSeekCallback()を使用しました。異なるオフセットにリセットするのは、アプリケーションが起動しているときではなく、起動しているときでなければならないため、registerSeekCallback()というコールバックを保存する必要はありません。 ドキュメントごとに、このregisterSeekCallback()は、リスナーコンテナが開始されたときに自動的に呼び出されます。だから私はregisterSeekCallbackメソッド内でcallback.seek(topic,partition,offset)メソッドを呼び出すつもりです。

質問：リスナーに割り当てられているパーティションを取得するにはどうすればよいですか。 seekメソッドには引数としてtopic、partition、およびoffsetが必要です。実行時にパーティションが割り当てられるため、正しいパーティション番号を取得してシークオペレーションを行う方法がわかりません。

Answer 1

registerSeekCallbackは、実行時（コンテナが開始され、パーティションが割り当てられた後の任意の時点）に任意に検索するために使用されます。

代わりに、あなたにパーティションを与えるConsumerSeekAware.onPartitionsAssigned()でシークする必要があります。