What is the alternative to deprecated poll() method in a Kafka Consumer in Scala?

user_7119501 0
Viewed 1134

I am trying to create a Scala consumer as below:

import java.util.Properties
import org.apache.kafka.clients.consumer.{ConsumerRecords, KafkaConsumer}
import org.apache.kafka.common.TopicPartition
import java.util.ArrayList
import scala.concurrent.duration._

object ScalaConsumer {

  def subscribePartitions() = {
    val props = new Properties()
    props.put("bootstrap.servers", "localhost:9092")
    props.put("key.serializer", "org.apache.kafka.common.serialization.StringDeSerializer")
    props.put("value.serializer", "org.apache.kafka.common.serialization.StringDeSerializer")

    val consumer = new KafkaConsumer[String, String](props)
    val partitionList:ArrayList[TopicPartition] = new ArrayList[TopicPartition]()
    val topicPartition1 = new TopicPartition("topicr1p3", 0)
    val topicPartition2 = new TopicPartition("othertopicr1p3", 2)
    partitionList.add(topicPartition1)
    partitionList.add(topicPartition2)
    consumer.assign(partitionList)
    try {
        val records:ConsumerRecords[String, String] = consumer.poll(10)
    } catch {
       case e:Exception => e.printStackTrace()
    }
  }

  def main(args: Array[String]): Unit = {

  }
}

The above example is incomplete because the method poll() used in

    val records:ConsumerRecords[String, String] = consumer.poll(10)

is deprecated in Scala.

enter image description here

It says the message: Symbol poll is deprecated. So I use the alternate for poll which is marked in the image below: enter image description here

in the code as:

val records:ConsumerRecords[String, String] = consumer.poll(Duration(1000, "millis"))

But this time, the error message is: Cannot resolve overloaded method 'poll' enter image description here

build.sbt contents:

name := "KafkaScalaImplementation"
version := "0.1"
scalaVersion := "2.12.10"
// https://mvnrepository.com/artifact/org.apache.kafka/kafka-clients
libraryDependencies += "org.apache.kafka" % "kafka-clients" % "2.4.0"

I am now confused a bit on what should I do here to use the poll(). Is there a new way of using poll method ? Could anyone let me know how can I fix the error in a proper way ?

kafka-consumer-api apache-kafka scala
1 Answers

The following worked for me:

import java.time.Duration
val records:ConsumerRecords[String, String] = consumer.poll(Duration.ofMillis(1000))

According to the Java Docs of Duration class you have the following options:

static Duration     of(long amount, TemporalUnit unit)
// Obtains a Duration representing an amount in the specified unit.

static Duration     ofDays(long days)
// Obtains a Duration representing a number of standard 24 hour days.

static Duration     ofHours(long hours)
// Obtains a Duration representing a number of standard hours.

static Duration     ofMillis(long millis)
// Obtains a Duration representing a number of milliseconds.

static Duration     ofMinutes(long minutes)
// Obtains a Duration representing a number of standard minutes.

static Duration     ofNanos(long nanos)
// Obtains a Duration representing a number of nanoseconds.

static Duration     ofSeconds(long seconds)
user_122089100
Related