title: Consume all messages on Kafka topic and disconnect
body: <p>I have a batch job which will be triggered once a day. The requirement is to </p>
  
  <ol>
  <li>consume all the messages available on the Kafka Topic at that point of time</li>
  <li>Process the messages</li>
  <li>If the process was successfully completed, commit the offsets.</li>
  </ol>
  
  <p>Currently I poll() the messages in while loop until ConsumerRecords.isEmpty() is true. When ConsumerRecords.isEmpty() is true I assume all the records available on Topic at the point of time has been consumed. The application maintains the offsets and closes the kafka consumer.</p>
  
  <p>When the processing on messages is done and successfully completed, I create a new KafkaConsumer and commit the offsets maintained by the application.</p>
  
  <p>Note I close the KafkaConsumer initially used to read the messages and use another KafkaConsumer instance to commit the offsets to avoid the consumer rebalance exception.</p>
  
  <p>I am expecting max of 5k messages on Topic. The topic is partitioned and replicated.</p>
  
  <p>Is there any better way to consume all messages on Topic at a specific point of time ? Is there anything I am missing or need to take care of ? I don't think I need to take care of consumer rebalancing since I poll() for the messages in loop and process the messages after the polling is done.</p>
  
  <p>I am using java kafka client v0.9 and can change to v0.10 if it helps in above scenario.</p>
  
  <p>Thanks</p>
  
  <p>Updated:</p>
  
  <pre><code>AtomicBoolean flag = new AtomicBoolean();
  flag.set(true);
  
  while(flag.get()) {
  
   ConsumerRecords&lt;String, String&gt; 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();
  </code></pre>