Dear Todd, By running bin/kafka-run-class.sh kafka.tools.GetOffsetShell --topic <topic_name> --broker-list localhost:9092 --time -1
I get the following current offset for <topic_name> <topic_name>:0:1760 But I guess this does not provide as much information. To answer your other question, onto how exactly do I track the offset - implicitly via Spark Streaming, i.e. using the default checkpoints. > On 07 Jun 2016, at 15:46, Todd Nist <tsind...@gmail.com> wrote: > > Hi Dominik, > > Right, and spark 1.6.x uses Kafka v0.8.2.x as I recall. However, it appears > as though the v.0.8 consumer is compatible with the Kafka v0.9.x broker, but > not the other way around; sorry for the confusion there. > > With the direct stream, simple consumer, offsets are tracked by Spark > Streaming within its checkpoints by default. You can also manage them > yourself if desired. How are you dealing with offsets ? > > Can you verify the offsets on the broker: > > kafka-run-class.sh kafka.tools.GetOffsetShell --topic <TOPIC> --broker-list > <BROKER-IP:PORT> --time -1 > > -Todd > > On Tue, Jun 7, 2016 at 8:17 AM, Dominik Safaric <dominiksafa...@gmail.com > <mailto:dominiksafa...@gmail.com>> wrote: > libraryDependencies += "org.apache.spark" % "spark-core_2.10" % "1.6.0" > libraryDependencies += "org.apache.spark" % "spark-streaming_2.10" % "1.6.0" > libraryDependencies += "org.apache.spark" % "spark-streaming-kafka_2.10" % > "1.6.1" > Please take a look at the SBT copy. > > I would rather think that the problem is related to the Zookeeper/Kafka > consumers. > > [2016-06-07 11:24:52,484] WARN Either no config or no quorum defined in > config, running in standalone mode > (org.apache.zookeeper.server.quorum.QuorumPeerMain) > > Any indication onto why the channel connection might be closed? Would it be > Kafka or Zookeeper related? > >> On 07 Jun 2016, at 14:07, Todd Nist <tsind...@gmail.com >> <mailto:tsind...@gmail.com>> wrote: >> >> What version of Spark are you using? I do not believe that 1.6.x is >> compatible with 0.9.0.1 due to changes in the kafka clients between 0.8.2.2 >> and 0.9.0.x. See this for more information: >> >> https://issues.apache.org/jira/browse/SPARK-12177 >> <https://issues.apache.org/jira/browse/SPARK-12177> >> >> -Todd >> >> On Tue, Jun 7, 2016 at 7:35 AM, Dominik Safaric <dominiksafa...@gmail.com >> <mailto:dominiksafa...@gmail.com>> wrote: >> Hi, >> >> Correct, I am using the 0.9.0.1 version. >> >> As already described, the topic contains messages. Those messages are >> produced using the Confluence REST API. >> >> However, what I’ve observed is that the problem is not in the Spark >> configuration, but rather Zookeeper or Kafka related. >> >> Take a look at the exception’s stack top item: >> >> org.apache.spark.SparkException: java.nio.channels.ClosedChannelException >> org.apache.spark.SparkException: Couldn't find leader offsets for >> Set([<topicname>,0]) >> at >> org.apache.spark.streaming.kafka.KafkaCluster$$anonfun$checkErrors$1.apply(KafkaCluster.scala:366) >> at >> org.apache.spark.streaming.kafka.KafkaCluster$$anonfun$checkErrors$1.apply(KafkaCluster.scala:366) >> at scala.util.Either.fold(Either.scala:97) >> at >> org.apache.spark.streaming.kafka.KafkaCluster$.checkErrors(KafkaCluster.scala:365) >> at >> org.apache.spark.streaming.kafka.KafkaUtils$.getFromOffsets(KafkaUtils.scala:222) >> at >> org.apache.spark.streaming.kafka.KafkaUtils$.createDirectStream(KafkaUtils.scala:484) >> at org.mediasoft.spark.Driver$.main(Driver.scala:22) >> at .<init>(<console>:11) >> at .<clinit>(<console>) >> at .<init>(<console>:7) >> >> By listing all active connections using netstat, I’ve also observed that >> both Zookeper and Kafka are running. Zookeeper on port 2181, while Kafka >> 9092. >> >> Furthermore, I am also able to retrieve all log messages using the console >> consumer. >> >> Any clue what might be going wrong? >> >>> On 07 Jun 2016, at 13:13, Jacek Laskowski <ja...@japila.pl >>> <mailto:ja...@japila.pl>> wrote: >>> >>> Hi, >>> >>> What's the version of Spark? You're using Kafka 0.9.0.1, ain't you? What's >>> the topic name? >>> >>> Jacek >>> >>> On 7 Jun 2016 11:06 a.m., "Dominik Safaric" <dominiksafa...@gmail.com >>> <mailto:dominiksafa...@gmail.com>> wrote: >>> As I am trying to integrate Kafka into Spark, the following exception >>> occurs: >>> >>> org.apache.spark.SparkException: java.nio.channels.ClosedChannelException >>> org.apache.spark.SparkException: Couldn't find leader offsets for >>> Set([*<topicName>*,0]) >>> at >>> org.apache.spark.streaming.kafka.KafkaCluster$$anonfun$checkErrors$1.apply(KafkaCluster.scala:366) >>> at >>> org.apache.spark.streaming.kafka.KafkaCluster$$anonfun$checkErrors$1.apply(KafkaCluster.scala:366) >>> at scala.util.Either.fold(Either.scala:97) >>> at >>> org.apache.spark.streaming.kafka.KafkaCluster$.checkErrors(KafkaCluster.scala:365) >>> at >>> org.apache.spark.streaming.kafka.KafkaUtils$.getFromOffsets(KafkaUtils.scala:222) >>> at >>> org.apache.spark.streaming.kafka.KafkaUtils$.createDirectStream(KafkaUtils.scala:484) >>> at org.mediasoft.spark.Driver$.main(Driver.scala:42) >>> at .<init>(<console>:11) >>> at .<clinit>(<console>) >>> at .<init>(<console>:7) >>> at .<clinit>(<console>) >>> at $print(<console>) >>> at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) >>> at >>> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) >>> at >>> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) >>> at java.lang.reflect.Method.invoke(Method.java:483) >>> at >>> scala.tools.nsc.interpreter.IMain$ReadEvalPrint.call(IMain.scala:734) >>> at >>> scala.tools.nsc.interpreter.IMain$Request.loadAndRun(IMain.scala:983) >>> at >>> scala.tools.nsc.interpreter.IMain.loadAndRunReq$1(IMain.scala:573) >>> at scala.tools.nsc.interpreter.IMain.interpret(IMain.scala:604) >>> at scala.tools.nsc.interpreter.IMain.interpret(IMain.scala:568) >>> at >>> scala.tools.nsc.interpreter.ILoop.reallyInterpret$1(ILoop.scala:760) >>> at >>> scala.tools.nsc.interpreter.ILoop.interpretStartingWith(ILoop.scala:805) >>> at scala.tools.nsc.interpreter.ILoop.command(ILoop.scala:717) >>> at scala.tools.nsc.interpreter.ILoop.processLine$1(ILoop.scala:581) >>> at scala.tools.nsc.interpreter.ILoop.innerLoop$1(ILoop.scala:588) >>> at scala.tools.nsc.interpreter.ILoop.loop(ILoop.scala:591) >>> at >>> scala.tools.nsc.interpreter.ILoop$$anonfun$process$1.apply$mcZ$sp(ILoop.scala:882) >>> at >>> scala.tools.nsc.interpreter.ILoop$$anonfun$process$1.apply(ILoop.scala:837) >>> at >>> scala.tools.nsc.interpreter.ILoop$$anonfun$process$1.apply(ILoop.scala:837) >>> at >>> scala.tools.nsc.util.ScalaClassLoader$.savingContextLoader(ScalaClassLoader.scala:135) >>> at scala.tools.nsc.interpreter.ILoop.process(ILoop.scala:837) >>> at scala.tools.nsc.interpreter.ILoop.main(ILoop.scala:904) >>> at >>> org.jetbrains.plugins.scala.compiler.rt.ConsoleRunner.main(ConsoleRunner.java:64) >>> at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) >>> at >>> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) >>> at >>> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) >>> at java.lang.reflect.Method.invoke(Method.java:483) >>> at >>> com.intellij.rt.execution.application.AppMain.main(AppMain.java:144) >>> >>> As for the Spark configuration: >>> >>> val conf: SparkConf = new >>> SparkConf().setAppName("AppName").setMaster("local[2]") >>> >>> val confParams: Map[String, String] = Map( >>> "metadata.broker.list" -> "<IP_ADDRESS>:9092", >>> "auto.offset.reset" -> "largest" >>> ) >>> >>> val topics: Set[String] = Set("<topic_name>") >>> >>> val context: StreamingContext = new StreamingContext(conf, Seconds(1)) >>> val kafkaStream = KafkaUtils.createDirectStream(context,confParams, >>> topics) >>> >>> kafkaStream.foreachRDD(rdd => { >>> rdd.collect().foreach(println) >>> }) >>> >>> context.awaitTermination() >>> context.start() >>> >>> The Kafka topic does exist, Kafka server is up and running and I am able to >>> produce messages to that particular topic using the Confluent REST API. >>> >>> What might the problem actually be? >>> >>> >>> >>> >>> -- >>> View this message in context: >>> http://apache-spark-user-list.1001560.n3.nabble.com/Apache-Spark-Kafka-Integration-org-apache-spark-SparkException-Couldn-t-find-leader-offsets-for-Set-tp27103.html >>> >>> <http://apache-spark-user-list.1001560.n3.nabble.com/Apache-Spark-Kafka-Integration-org-apache-spark-SparkException-Couldn-t-find-leader-offsets-for-Set-tp27103.html> >>> Sent from the Apache Spark User List mailing list archive at Nabble.com >>> <http://nabble.com/>. >>> >>> --------------------------------------------------------------------- >>> To unsubscribe, e-mail: user-unsubscr...@spark.apache.org >>> <mailto:user-unsubscr...@spark.apache.org> >>> For additional commands, e-mail: user-h...@spark.apache.org >>> <mailto:user-h...@spark.apache.org> >>> >> >> > >
