KafkaConsumer : erro on consumer.subscribe(Arrays.asList(topic)) error

Question

I'm trying to config my KafkaConsumer to read data from KafkaProducer by the 'kafkatopic'.

My scala code is :

    package com.sundogsoftware.sparkstreaming

    import java.util
    import java.util.Properties
    import org.apache.kafka.clients.consumer.KafkaConsumer
    import scala.collection.JavaConverters._
    import org.apache.kafka.clients.consumer.{ ConsumerRecords, KafkaConsumer }
    import org.apache.kafka.clients.producer.{ KafkaProducer, Producer, ProducerConfig, ProducerRecord }
    import java.util.Arrays;


object ConsumerExample extends App {
  val props = new Properties()
  props.put("bootstrap.servers", "localhost:9092");
  props.put("group.id", "test-group");
  props.put("enable.auto.commit", "true");
  props.put("auto.commit.interval.ms", "1000");
  props.put("session.timeout.ms", "30000");
  props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
  props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
  props.put("partition.assignment.strategy", "range");

  val consumer = new KafkaConsumer[String, String](props)

  consumer.subscribe(Arrays.asList("kafkatopic"))

  while (true) {
    val records = consumer.poll(100)
    println(consumer)
    println(records)
    for (record <- records.asScala) {
      println(record)
    }
  }
}

The error is on consumer.subscribe(Arrays.asList("kafkatopic")):

overloaded method value subscribe with alternatives: (x$1: org.apache.kafka.common.TopicPartition*)Unit (x$1: String*)Unit cannot be applied to (java.util.List[String])

overloaded method value subscribe with alternatives: (x$1: org.apache.kafka.common.TopicPartition*)Unit <and> (x$1: String*)Unit cannot be applied to (Array[String]) — Thaise
Exception in thread "main" java.lang.NullPointerException at scala.collection.convert.Wrappers$JMapWrapperLike$$anon$2.<init>(Wrappers.scala:275) at ... — Thaise
@Thaise Did you able to figure out the problem ? i am getting the same issue with subscribe(). — Akan

streetturtle streetturtle · Accepted Answer · 2017-05-19T18:10:41

In Scala usually it's done following way:

consumer.subscribe(Collections.singletonList("kafkatopic"));

KafkaConsumer : erro on consumer.subscribe(Arrays.asList(topic)) error

1 Answers