title: Spark streaming Kafka messages not consumed
body: <p>I want to receive messages from a topic in Kafka (broker v <strong>0.10.2.1</strong>) using Spark (1.6.2) Streaming.</p>
  
  <p>I'm using the <strong><code>Receiver</code></strong> approach. The code is as following code:</p>
  
  <pre><code>public static void main(String[] args) throws Exception
  {
      SparkConf sparkConf = new SparkConf().setAppName("SimpleStreamingApp");
      JavaStreamingContext javaStreamingContext = new JavaStreamingContext(sparkConf, new Duration(5000));
      //
      Map&lt;String, Integer&gt; topicMap = new HashMap&lt;&gt;();
      topicMap.put("myTopic", 1);
      //
      String zkQuorum = "host1:port1,host2:port2,host3:port3";
      //
      Map&lt;String, String&gt; kafkaParamsMap = new HashMap&lt;&gt;();
      kafkaParamsMap.put("bootstraps.server", zkQuorum);
      kafkaParamsMap.put("metadata.broker.list", zkQuorum);
      kafkaParamsMap.put("zookeeper.connect", zkQuorum);
      kafkaParamsMap.put("group.id", "group_name");
      kafkaParamsMap.put("security.protocol", "SASL_PLAINTEXT");
      kafkaParamsMap.put("security.mechanism", "GSSAPI");
      kafkaParamsMap.put("ssl.kerberos.service.name", "kafka");
      kafkaParamsMap.put("key.deserializer", "kafka.serializer.StringDecoder");
      kafkaParamsMap.put("value.deserializer", "kafka.serializer.DefaultDecoder");
      //
      JavaPairReceiverInputDStream&lt;byte[], byte[]&gt; stream = KafkaUtils.createStream(javaStreamingContext,
                              byte[].class, byte[].class,
                              DefaultDecoder.class, DefaultDecoder.class,
                              kafkaParamsMap,
                              topicMap,
                              StorageLevel.MEMORY_ONLY());
  
      VoidFunction&lt;JavaPairRDD&lt;byte[], byte[]&gt;&gt; voidFunc = new VoidFunction&lt;JavaPairRDD&lt;byte[], byte[]&gt;&gt; ()
      {
         public void call(JavaPairRDD&lt;byte[], byte[]&gt; rdd) throws Exception
         {
            List&lt;Tuple2&lt;byte[], byte[]&gt;&gt; all = rdd.collect();
            System.out.println("size of red: " + all.size());
         }
      }
  
      stream.forEach(voidFunc);
  
      javaStreamingContext.start();
      javaStreamingContext.awaitTermination();
  }
  </code></pre>
  
  <p>Access to Kafka is <strong>kerberized</strong>. When I launch</p>
  
  <p><code>spark-submit --verbose --conf "spark.executor.extraJavaOptions=-Djava.security.auth.login.config=jaas.conf"</code>
  <code>--files jaas.conf,privKey.der</code>
  <code>--principal &lt;accountName&gt;</code>
  <code>--keytab &lt;path to keytab file&gt;</code>
  <code>--master yarn</code>
  <code>--jars &lt;comma separated path to all jars&gt;</code>
  <code>--class &lt;fully qualified java main class&gt;</code>
  <code>&lt;path to jar file containing main class&gt;</code></p>
  
  <ol>
  <li><code>VerifiableProperties</code> class from Kafka logs <strong>warning</strong> messages for the properties included in the <code>kafkaParams</code> hashmap:</li>
  </ol>
  
  <blockquote>
  <pre><code>INFO KafkaReceiver: connecting to zookeeper: &lt;the correct zookeeper quorum provided in kafkaParams map&gt;
  
  VerifiableProperties: Property auto.offset.reset is overridden to largest
  VerifiableProperties: Property enable.auto.commit is not valid.
  VerifiableProperties: Property sasl.kerberos.service.name is not valid
  VerifiableProperties: Property key.deserializer is not valid
  ...
  VerifiableProperties: Property zookeeper.connect is overridden to ....
  </code></pre>
  </blockquote>
  
  <p>I think because these properties are not accepted, so it might be affecting the stream processing.</p>
  
  <p>** when I launch in the cluster mode <code>--master yarn</code>, then these warning messages don't appear**</p>
  
  <ol start="2">
  <li><p>Later, I see following logs repeated every 5 seconds as configured:</p>
  
  <p><code>INFO BlockRDD: Removing RDD 4 from persistence list</code></p>
  
  <p><code>INFO KafkaInputDStream: Removing blocks of RDD BlockRDD[4] at createStream at ...</code></p>
  
  <p><code>INFO ReceivedBlockTracker: Deleting batches ArrayBuffer()</code></p>
  
  <p><code>INFO ...  INFO BlockManager: Removing RDD 4</code></p></li>
  </ol>
  
  <p>However, I don't see any actual <strong>message</strong> getting printed on the console.</p>
  
  <p>Question: Why is my code not printing any actual messages?</p>
  
  <p>My gradle dependencies are:</p>
  
  <pre><code>compile group: 'org.apache.spark', name: 'spark-core_2.10', version: '1.6.2'
  compile group: 'org.apache.spark', name: 'spark-streaming_2.10', version: '1.6.2'
  compile group: 'org.apache.spark', name: 'spark-streaming-kafka_2.10', version: '1.6.2'
  </code></pre>