初めてカフカ消費者がレコードを取得すると、多くのレコードのうちの1レコードしか取得できないのですか？

Question

私はspring-kafkaとspring-kafka-testバージョン1.0.2.RELEASEを使用しています。

私のテストでは、私のアプリケーションは、KafkaTemplateとほとんどのデフォルトの設定を使ってEmbeddedKafkaインスタンスの1つのTopicPartionに100レコードを連続して送ります。

私はKafkaインスタンスからレコードを取得し、それらがすべて送信されたことを確認するために、KafkaTestUtils.getRecords（コンシューマ）メソッドを使用します。

初めてgetRecordsを呼び出すと、1つのレコードしか受け取りません。私は再びそれを呼び出すと、私は明示的にTopicPartitionの先頭に消費者の位置を設定し、getRecordsを呼び出す場合、私はすべて100

はなぜgetRecordsは一つだけになるだろう得る他の99

を取得します最初に録音しますか？消費者に明示的にseekToBeginningを呼び出すことによって、100を一度に取得するより良い方法はありますか？

Answer 1

競合状態の可能性が最も高い - 消費者はpoll()に座っており、ブローカーは到着するとすぐに最初のメッセージを送信します。

kafka docsのプロパティfetch.min.bytesとfetch.max.wait.msを参照してください。

fetch.min.bytesは、デフォルトでは1です。あなたはまた、getRecords()を呼び出す前KafkaTemplateをINGのflush()を試みることができる

EDIT

。

しかし、テストでは、すべてのメッセージを1つのフェッチで取得することに本当に頼りすぎてはいけません。

Answer 2

タイミングの問題のように聞こえます。最初にpoll()を呼び出したときに、1つのメッセージしか利用できなかった可能性があります。この方法では、取り出されるメッセージの数は保証されません。コードを書くときには、一度にXレコードを受け取ると仮定してはいけません。テスト目的のために、すべてを100にポーリングするまで、受信ループを実行することができる、Kafka 0.10 max.poll.recordsのコンシューマープロパティがあります。