カフカに関するすべてのメッセージを消費して切断します。

Question

1日に1回トリガーされるバッチジョブがあります。

1. が
2. プロセスのメッセージ処理が正常に完了した場合
3. 、オフセットをコミット、その時点でカフカトピックで使用可能なすべてのメッセージを消費する必要があります。

現在、ConsumerRecords.isEmpty（）がtrueになるまでwhileループ内のメッセージをポーリングします。 ConsumerRecords.isEmpty（）がtrueの場合、その時点でTopicで使用可能なすべてのレコードが消費されたと見なします。アプリケーションはオフセットを維持し、カフカの消費者を閉じます。

メッセージの処理が完了して正常に完了すると、私は新しいKafkaConsumerを作成し、アプリケーションによって維持されるオフセットをコミットします。

注意最初にメッセージを読むために使用したKafkaConsumerを閉じ、コンシューマの再バランスの例外を避けるために別のKafkaConsumerインスタンスを使用してオフセットをコミットします。

トピックに対して最大5kのメッセージが必要です。トピックは分割され、複製されます。

トピックの特定の時点ですべてのメッセージを消費する良い方法はありますか？行方不明、または世話をする必要があるものはありますか？私はループ内のメッセージをpoll（）してからポーリングが完了した後にメッセージを処理するので、消費者のリバランスを考慮する必要はないと思います。

私はjava kafka client v0.9を使用しており、上記のシナリオで役立つ場合はv0.10に変更できます。

更新

ありがとう：

AtomicBoolean flag = new AtomicBoolean(); 
flag.set(true); 

while(flag.get()) { 

ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(timeout); 

if(consumerRecords.isEmpty()) { 
    flag.set(false); 
    continue; 
} 
    //if the ConsumerRecords is not empty process the messages and continue to poll() 
} 
kafkaConsumer.close(); 

Answer 1

あなたは）（ポーリングする呼び出しの後、あなたが原因max.pollにその瞬間のトピックで利用可能なすべてのメッセージを読んでいると仮定することはできません消費者の.records設定パラメータ。これは、単一のpoll（）によって返されるレコードの最大数で、デフォルト値は500です。つまり、その瞬間にトピック内に600個のメッセージがある場合は、すべてのメッセージを読むためにpoll（）を2回呼び出す必要があります（ただし、他のメッセージが届く可能性があると考えてください）。 私が理解していないもう一つの理由は、異なる消費者を使用してオフセットをコミットする理由です。あなたが話している消費者の再バランスの例外は何ですか？