Apache Kafkaで追加されたパーティションを消費する

Question

私はKafkaConsumerを持っており、手動でパーティションを割り当てています。パーティションを配布するには、追加のパーティションを検出するために一定の間隔でconsumer.partitionsFor(topicId)を使用します。ジョブは永遠に実行され、このケースをサポートしたいからです。ただし、コンシューマを再起動しない限り、これは常にパーティションの初期リストを返します。

消費者から追加されたパーティションを検出する方法はありますか？何をポーリングするのか聞きますか？

Answer 1

KafkaConsumerは、configuration property "metadata.max.age.ms"（デフォルトは5分）です。つまり、consumer.partitionsForを呼び出すたびに、キャッシュされたコピーが返され、新しいメタデータだけが取得されます。

プロパティを0に設定するたびに、新しいメタデータがフェッチされます。

Answer 2

assignment()メソッドでクエリを実行すると、これがパーティションの再割り当てのプロセスにある可能性があります。 5秒間スリープ状態にしてから、これをチェックしてテストすることができます。

現在、このコンシューマに割り当てられているパーティションのセットを取得します。 assign（Collection）を使用してパーティションを直接割り当てることによってサブスクリプションが発生した場合、割り当てられたパーティションと同じパーティションが返されます。 トピックサブスクリプションを使用した場合、現在コンシューマに割り当てられているトピックパーティションのセットが表示されます（割り当てがまだ行われていない場合、またはパーティションが再割り当て中の場合はありません）。