title: Retrieve last n messages of Kafka consumer from a particular topic
body: <p>kafka version : 0.9.0.1</p>
  
  <p>If <code>n = 20</code>,
  I have to get last 20 messages of a topic.</p>
  
  <p>I tried with</p>
  
  <pre class="lang-java prettyprint-override"><code>kafkaConsumer.seekToBeginning();
  </code></pre>
  
  <p>But it retrieves all the messages. I need to get only the last 20 messages.</p>
  
  <p>This topic may have hundreds of thousands of records</p>
  
  <pre class="lang-java prettyprint-override"><code>public List&lt;JSONObject&gt; consumeMessages(String kafkaTopicName) {
    KafkaConsumer&lt;String, String&gt; kafkaConsumer = null;
    boolean flag = true;
    List&lt;JSONObject&gt; messagesFromKafka = new ArrayList&lt;&gt;();
    int recordCount = 0;
    int i = 0;
    int maxMessagesToReturn = 20;
  
    Properties props = new Properties();         
    props.put("bootstrap.servers", "localhost:9092");
    props.put("group.id", "project.group.id");
    props.put("max.partition.fetch.bytes", "1048576000");
    props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    kafkaConsumer = new KafkaConsumer&lt;&gt;(props);
  
    kafkaConsumer.subscribe(Arrays.asList(kafkaTopicName));
    TopicPartition topicPartition = new TopicPartition(kafkaTopicName, 0);
    LOGGER.info("Subscribed to topic " + kafkaConsumer.listTopics());
    while (flag) {
      // will consume all the messages and store in records
      ConsumerRecords&lt;String, String&gt; records = kafkaConsumer.poll(1000);
      kafkaConsumer.seekToBeginning(topicPartition);
  
      // getting total records count
      recordCount = records.count();
      LOGGER.info("recordCount " + recordCount);
      for (ConsumerRecord&lt;String, String&gt; record : records) {
        if(record.value() != null) {
          if (i &gt;= recordCount - maxMessagesToReturn) {
            // adding last 20 messages to messagesFromKafka
            LOGGER.info("kafkaMessage "+record.value());
            messagesFromKafka.add(new JSONObject(record.value()));
          }
          i++;
        }
      }
      if (recordCount &gt; 0) {
        flag = false;
      }
    }
    kafkaConsumer.close();
    return messagesFromKafka;
  }
  </code></pre>