title: Kafka Spark Streaming Consumer will not receive any messages from Kafka Console Producer?
body: <p>I'm trying to integrate spark and Kafka for consuming the messages from Kafka. I have producer code also to send messages on "temp" topic. Also, I'm using Kafka's Console Producer to producer the messages on "temp" topic.</p>
  
  <p>I have created below code to consume the messages from the same "temp" topic but it will not receive single message also.</p>
  
  <p><strong>Program:</strong></p>
  
  <pre><code>import java.util.Arrays;
  import java.util.Map;
  import java.util.HashMap;
  import static org.apache.commons.lang3.StringUtils.SPACE;
  
  import org.apache.spark.SparkConf;
  import org.apache.spark.streaming.Duration;
  import org.apache.spark.streaming.api.java.JavaDStream;
  import org.apache.spark.streaming.api.java.JavaPairDStream;
  import org.apache.spark.streaming.api.java.JavaPairReceiverInputDStream;
  import org.apache.spark.streaming.api.java.JavaStreamingContext;
  import org.apache.spark.streaming.kafka.KafkaUtils;
  import scala.Tuple2;
  import org.apache.log4j.Logger;
  import org.apache.spark.api.java.JavaSparkContext;
  import scala.collection.immutable.ListSet;
  import scala.collection.immutable.Set;
  
  public class ConsumerDemo {
  
      public void main() {
          String zkGroup = "localhost:2181";
          String group = "test";
          String[] topics = {"temp"};
          int numThreads = 1;
  
          SparkConf sparkConf = new SparkConf().setAppName("JavaKafkaWordCount").setMaster("local[4]").set("spark.ui.port‌​", "7077").set("spark.executor.memory", "1g");
          JavaStreamingContext jssc = new JavaStreamingContext(sparkConf, new Duration(2000));
          Map&lt;String, Integer&gt; topicMap = new HashMap&lt;&gt;();
          for (String topic : topics) {
              topicMap.put(topic, numThreads);
          }
          System.out.println("topics : " + Arrays.toString(topics));
          JavaPairReceiverInputDStream&lt;String, String&gt; messages
                  = KafkaUtils.createStream(jssc, zkGroup, group, topicMap);
  
          messages.print();
  
          JavaDStream&lt;String&gt; lines = messages.map(Tuple2::_2);
  
          //lines.print();
          JavaDStream&lt;String&gt; words = lines.flatMap(x -&gt; Arrays.asList(SPACE.split(x)).iterator());
  
          JavaPairDStream&lt;String, Integer&gt; wordCounts = words.mapToPair(s -&gt; new Tuple2&lt;&gt;(s, 1))
                  .reduceByKey((i1, i2) -&gt; i1 + i2);
  
          //wordCounts.print();
          jssc.start();
          jssc.awaitTermination();
      }
  
      public static void main(String[] args) {
          System.out.println("Started...");
          new ConsumerDemo().main();
          System.out.println("Ended...");
      }
  }
  </code></pre>
  
  <p>I added following dependencies in the pom.xml file:</p>
  
  <pre><code>    &lt;dependency&gt;
          &lt;groupId&gt;org.apache.kafka&lt;/groupId&gt;
          &lt;artifactId&gt;kafka_2.10&lt;/artifactId&gt;
          &lt;version&gt;0.9.0.0&lt;/version&gt;
      &lt;/dependency&gt;
      &lt;dependency&gt;
          &lt;groupId&gt;org.apache.kafka&lt;/groupId&gt;
          &lt;artifactId&gt;kafka-clients&lt;/artifactId&gt;
          &lt;version&gt;0.11.0.0&lt;/version&gt;
      &lt;/dependency&gt;
  
      &lt;dependency&gt;
          &lt;groupId&gt;org.apache.spark&lt;/groupId&gt;
          &lt;artifactId&gt;spark-core_2.11&lt;/artifactId&gt;
          &lt;version&gt;2.2.0&lt;/version&gt;
          &lt;scope&gt;provided&lt;/scope&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;0.9.0-incubating&lt;/version&gt;
          &lt;type&gt;jar&lt;/type&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;1.6.3&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;1.6.3&lt;/version&gt;
          &lt;type&gt;jar&lt;/type&gt;
      &lt;/dependency&gt;
  
      &lt;dependency&gt;
          &lt;groupId&gt;log4j&lt;/groupId&gt;
          &lt;artifactId&gt;log4j&lt;/artifactId&gt;
          &lt;version&gt;1.2.17&lt;/version&gt;
      &lt;/dependency&gt;
  
      &lt;dependency&gt;
          &lt;groupId&gt;org.anarres.lzo&lt;/groupId&gt;
          &lt;artifactId&gt;lzo-core&lt;/artifactId&gt;
          &lt;version&gt;1.0.5&lt;/version&gt;
          &lt;type&gt;jar&lt;/type&gt;
      &lt;/dependency&gt;
  
      &lt;dependency&gt; 
          &lt;groupId&gt;com.fasterxml.jackson.core&lt;/groupId&gt; 
          &lt;artifactId&gt;jackson-databind&lt;/artifactId&gt; 
          &lt;version&gt;2.8.2&lt;/version&gt; 
      &lt;/dependency&gt; 
      &lt;dependency&gt; 
          &lt;groupId&gt;com.fasterxml.jackson.module&lt;/groupId&gt; 
          &lt;artifactId&gt;jackson-module-scala_2.10&lt;/artifactId&gt; 
          &lt;version&gt;2.8.2&lt;/version&gt; 
      &lt;/dependency&gt;
      &lt;dependency&gt;
          &lt;groupId&gt;com.msiops.footing&lt;/groupId&gt;
          &lt;artifactId&gt;footing-tuple&lt;/artifactId&gt;
          &lt;version&gt;0.2&lt;/version&gt;
      &lt;/dependency&gt;
  </code></pre>
  
  <p>Is I'm missing some dependency or issue is in code? Why this code will not receive any messages?</p>