title: Spark Streaming Kafka Consumer
body: <p>I'm attempting to set up a Spark Streaming simple app that will read messages from a Kafka topic.</p>
  
  <p>After much work I am at this stage but get the exceptions shown below.</p>
  
  <p>Code:</p>
  
  <pre><code>public static void main(String[] args) throws Exception {
  
      String brokers = "my.kafka.broker" + ":" + "6667";
      String topics = "MyKafkaTopic";
  
      // Create context with a 2 seconds batch interval
      SparkConf sparkConf = new SparkConf().setAppName("StreamingE")
              .setMaster("local[1]")
              ;
      JavaStreamingContext jssc = new JavaStreamingContext(sparkConf, Durations.seconds(2));
  
      Set&lt;String&gt; topicsSet = new HashSet&lt;&gt;(Arrays.asList(topics.split(",")));
      Map&lt;String, String&gt; kafkaParams = new HashMap&lt;&gt;();
      kafkaParams.put("metadata.broker.list", brokers);
      System.out.println("Brokers: " + brokers);
  
      // Create direct kafka stream with brokers and topics
      JavaPairInputDStream&lt;String, String&gt; messages = KafkaUtils.createDirectStream(
              jssc,
              String.class,
              String.class,
              StringDecoder.class,
              StringDecoder.class,
              kafkaParams,
              topicsSet
      );
  
      System.out.println("Message received: " + messages);
  
      // Start the computation
      jssc.start();
      jssc.awaitTermination();
  
  }
  </code></pre>
  
  <p>Which throws:</p>
  
  <pre><code>[WARNING] 
  java.lang.reflect.InvocationTargetException
      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:498)
      at org.codehaus.mojo.exec.ExecJavaMojo$1.run(ExecJavaMojo.java:293)
      at java.lang.Thread.run(Thread.java:745)
  Caused by: java.lang.IllegalArgumentException: requirement failed: No output operations registered, so nothing to execute
      at scala.Predef$.require(Predef.scala:233)
      at org.apache.spark.streaming.DStreamGraph.validate(DStreamGraph.scala:161)
      at org.apache.spark.streaming.StreamingContext.validate(StreamingContext.scala:542)
      at org.apache.spark.streaming.StreamingContext.liftedTree1$1(StreamingContext.scala:601)
      at org.apache.spark.streaming.StreamingContext.start(StreamingContext.scala:600)
      at org.apache.spark.streaming.api.java.JavaStreamingContext.start(JavaStreamingContext.scala:624)
      at com.ncr.dataplatform.api.StreamingE.main(StreamingE.java:66)
  </code></pre>
  
  <p>Out of desperation I tried connecting to the Zookeeper:</p>
  
  <pre><code>String brokers = "my.kafka.zookeeper" + ":" + "2181";
  String topics = "MyKafkaTopic";
  </code></pre>
  
  <p>But that throws:</p>
  
  <pre><code>[WARNING] 
  java.lang.reflect.InvocationTargetException
      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:498)
      at org.codehaus.mojo.exec.ExecJavaMojo$1.run(ExecJavaMojo.java:293)
      at java.lang.Thread.run(Thread.java:745)
  Caused by: org.apache.spark.SparkException: java.io.EOFException: Received -1 when reading from channel, socket has likely been closed.
      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.apache.spark.streaming.kafka.KafkaUtils$.createDirectStream(KafkaUtils.scala:607)
      at org.apache.spark.streaming.kafka.KafkaUtils.createDirectStream(KafkaUtils.scala)
      at com.ncr.dataplatform.api.StreamingE.main(StreamingE.java:53)
  </code></pre>
  
  <p>The relevant dependencies are:</p>
  
  <pre><code>&lt;properties&gt;
    &lt;spark.version&gt;1.6.2&lt;/spark.version&gt;
    &lt;kafka.version&gt;0.8.2.1&lt;/kafka.version&gt;
  &lt;/properties&gt;
  
  &lt;dependency&gt;
    &lt;groupId&gt;org.apache.kafka&lt;/groupId&gt;
    &lt;artifactId&gt;kafka_2.10&lt;/artifactId&gt;
    &lt;version&gt;${kafka.version}&lt;/version&gt;
  &lt;/dependency&gt;
  
  &lt;dependency&gt;
    &lt;groupId&gt;org.apache.spark&lt;/groupId&gt;
    &lt;artifactId&gt;spark-streaming_2.10&lt;/artifactId&gt;
    &lt;version&gt;${spark.version}&lt;/version&gt;
  &lt;/dependency&gt;
  
  &lt;dependency&gt;
    &lt;groupId&gt;org.apache.spark&lt;/groupId&gt;
    &lt;artifactId&gt;spark-core_2.10&lt;/artifactId&gt;
    &lt;version&gt;${spark.version}&lt;/version&gt;
  &lt;/dependency&gt;
  
  &lt;dependency&gt;
    &lt;groupId&gt;org.apache.spark&lt;/groupId&gt;
    &lt;artifactId&gt;spark-streaming-kafka_2.10&lt;/artifactId&gt;
    &lt;version&gt;${spark.version}&lt;/version&gt;
  &lt;/dependency&gt;
  </code></pre>
  
  <p>I'd like to ask:</p>
  
  <p>Should I be connecting to the Kafka broker or the zookeeper servers?</p>
  
  <p>What am I doing wrong in my code to be unable to connect/listen to incoming messages?</p>