Kafka Consumer Get Latest 10 messages Example

KafkaConsumer.scala

import collection.JavaConverters._
object KafkaConsumer {

  val topicName = "yourTopicName"
  val partitionNumber = 1
  val offsetOnPoll = 10
  val topicPartition = new TopicPartition(topicName, partitionNumber)

  def main(args: Array[String]): Unit = {
    val properties = new Properties()
    properties.put("bootstrap.servers", "localhost:9092")
    properties.put("group.id", "consumer-group-1")
    properties.put("key.deserializer", classOf[StringDeserializer])
    properties.put("value.deserializer", classOf[StringDeserializer])

    val kafkaConsumer = new KafkaConsumer[String, String](properties)
    kafkaConsumer.subscribe(List(topicPartition.topic).asJava)
    val getLatestOffsets = kafkaConsumer.endOffsets(List(topicPartition).asJava)

    kafkaConsumer.seek(topicPartition, getLatestOffsets.asScala.head._2 - offsetOnPoll)
    val results = kafkaConsumer.poll(3000).asScala

    //iterate over the last `offsetOnPoll` number of results
    for (consumerRecord <- results) {
      val valueOf = consumerRecord.value()
      // Do your stuff here
    }

  }

}