Looking at KafkaCluster#getLeaderOffsets(): respMap.get(tp).foreach { por: PartitionOffsetsResponse => if (por.error == ErrorMapping.NoError) { ... } else { errs.append(ErrorMapping.exceptionFor(por.error)) } There should be some error other than "Couldn't find leader offsets for Set()"
Can you check again ? Thanks On Thu, Mar 19, 2015 at 12:10 PM, Alberto Rodriguez <ardl...@gmail.com> wrote: > Hi all, > > I am trying to make the new kafka and spark streaming integration work > (direct > approach "no receivers" > <http://spark.apache.org/docs/1.3.0/streaming-kafka-integration.html>). I > have created an unit test where I configure and start both zookeeper and > kafka. > > When I try to create the InputDStream using the createDirectStream method > of the KafkaUtils class I am getting the following error: > > org.apache.spark.SparkException:* Couldn't find leader offsets for Set()* > org.apache.spark.SparkException: org.apache.spark.SparkException: Couldn't > find leader offsets for Set() > at > > org.apache.spark.streaming.kafka.KafkaUtils$$anonfun$createDirectStream$2.apply(KafkaUtils.scala:413) > > Following is the code that tries to create the DStream: > > val messages: InputDStream[(String, String)] = > KafkaUtils.createDirectStream[String, String, StringDecoder, > StringDecoder]( > ssc, kafkaParams, topics) > > Does anyone faced this problem? > > Thank you in advance. > > Kind regards, > > Alberto >