Кафка потребительская в Скале - PullRequest
Новая
       6

Кафка потребительская в Скале

0 голосов
/ 19 октября 2018

Я пытаюсь написать потребительский код Kafka, используя Scala.

У меня есть кластер Kafka (сервер A) с некоторыми сообщениями, уже опубликованными в теме (test-topic), и я использую его с другого сервера (скажем, B), который имеет соединение, установленное с сервера B на A(Я использую спарк-оболочку для кодирования).

Ниже приведен фрагмент кода, который я пытаюсь выполнить на сервере B, но я получаю исключение пустого указателя при выполнении кода. 

import java.util._
import java.util.Properties
import org.apache.kafka.clients.consumer.KafkaConsumer
import org.apache.kafka.clients.consumer.ConsumerConfig
import scala.collection.JavaConverters._
import org.apache.kafka.common.TopicPartition
val  props = new Properties()
props.put("group.id","group-test")
props.put("auto.commit.interval.ms","1000")
props.put("auto.offset.reset","earliest")
props.put("bootstrap.servers", "<kafka server ip>:9092")
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","org.apache.kafka.clients.consumer.RangeAssignor")
val consumer = new KafkaConsumer[String, String](props)
consumer.subscribe("test-topic")
 while(true){
   val records=consumer.poll(100)
   println(consumer)
   println(records)
   for (record<-records.asScala){
    println(record)
   }
 }



Error --
org.apache.kafka.clients.consumer.KafkaConsumer@29451e22
null
java.lang.NullPointerException
  at scala.collection.convert.Wrappers$JMapWrapperLike$$anon$2.<init>(Wrappers.scala:275)
  at scala.collection.convert.Wrappers$JMapWrapperLike$class.iterator(Wrappers.scala:274)
  at scala.collection.convert.Wrappers$JMapWrapper.iterator(Wrappers.scala:292)
  at scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
  at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
  ... 59 elided

Пожалуйста, помогите решить проблему.

1 Ответ

0 голосов
/ 19 октября 2018

consumer.poll (100) возвращает ноль.

Увеличьте время ожидания и проверьте записи! = Null, прежде чем их зацикливать.

...