title: Apache Spark-Kafka.TaskCompletionListenerException & KafkaRDD$KafkaRDDIterator.close NPE on local cluster(Client Mode)
body: <p>My spark-streaming code works seamlessly on eclipse IDE.
  But when i run it on a local spark cluster it gives the org.apache.spark.util.TaskCompletionListenerException.</p>
  
  <p>Also on spark-submit "Client Mode" code runs fine, until i start my kafka producer but as i start producer it gives the following error.</p>
  
  <p>I start the local cluster with command sh SPARK_HOME/sbin/start-all.sh</p>
  
  <p>and call spark-submit with this <strong>script</strong>.</p>
  
  <pre><code>#!/bin/sh
  
  SP_SUBMIT=/home/user/spark/bin/spark-submit
  DEP_MODE=client
  
  
  $SP_SUBMIT \
  --deploy-mode $DEP_MODE \
  --class com.alind.sparkStream.Test \
  --master spark://clstr:7077 \
  --name alind\
  /home/user/jar/com.alind-0.0.1-SNAPSHOT.jar \
  </code></pre>
  
  <p>And i am getting this <strong>error</strong> when the spark stream starts receiving messages.</p>
  
  <pre><code>2015-06-29 16:13:56 ERROR JobScheduler:96 - Error running job streaming job 1435574590600 ms.3
  org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 306.0 failed 1 times, most recent failure: Lost task 0.0 in stage 306.0 (TID 164, localhost): org.apache.spark.util.TaskCompletionListenerException
      at org.apache.spark.TaskContextImpl.markTaskCompleted(TaskContextImpl.scala:76)
      at org.apache.spark.scheduler.Task.run(Task.scala:58)
      at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:200)
      at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
      at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
      at java.lang.Thread.run(Thread.java:745)
  
  Driver stacktrace:
      at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1214)
      at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1203)
      at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1202)
      at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
      at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47)
      at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:1202)
      at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:696)
      at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:696)
      at scala.Option.foreach(Option.scala:236)
      at org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:696)
      at org.apache.spark.scheduler.DAGSchedulerEventProcessActor$$anonfun$receive$2.applyOrElse(DAGScheduler.scala:1420)
      at akka.actor.Actor$class.aroundReceive(Actor.scala:465)
      at org.apache.spark.scheduler.DAGSchedulerEventProcessActor.aroundReceive(DAGScheduler.scala:1375)
      at akka.actor.ActorCell.receiveMessage(ActorCell.scala:516)
      at akka.actor.ActorCell.invoke(ActorCell.scala:487)
      at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:238)
      at akka.dispatch.Mailbox.run(Mailbox.scala:220)
      at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:393)
      at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
      at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
      at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
      at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
  2015-06-29 16:13:56 WARN  JobProgressListener:71 - Task start for unknown stage 307
  2015-06-29 16:13:56 WARN  JobProgressListener:71 - Task start for unknown stage 308
  2015-06-29 16:13:56 WARN  JobProgressListener:71 - Task start for unknown stage 309
  2015-06-29 16:13:56 INFO  SparkContext:59 - Starting job: foreach at Test.java:428
  2015-06-29 16:13:56 INFO  MapOutputTrackerMaster:59 - Size of output statuses for shuffle 34 is 84 bytes
  2015-06-29 16:13:56 INFO  MapOutputTrackerMaster:59 - Size of output statuses for shuffle 35 is 84 bytes
  2015-06-29 16:13:56 INFO  DAGScheduler:59 - Got job 94 (foreach at Test.java:428) with 2 output partitions (allowLocal=false)
  2015-06-29 16:13:56 INFO  DAGScheduler:59 - Final stage: Stage 327(foreach at Test.java:428)
  2015-06-29 16:13:56 INFO  DAGScheduler:59 - Parents of final stage: List(Stage 320, Stage 317, Stage 324, Stage 321, Stage 318, Stage 325, Stage 322, Stage 326, Stage 323, Stage 319)
  2015-06-29 16:13:56 INFO  ShuffledDStream:59 - Slicing from 1435574619500 ms to 1435574620400 ms (aligned to 1435574619500 ms and 1435574620400 ms)
  2015-06-29 16:13:56 INFO  DAGScheduler:59 - Missing parents: List(Stage 320, Stage 317, Stage 318, Stage 319)
  2015-06-29 16:13:56 INFO  DAGScheduler:59 - Submitting Stage 317 (MappedRDD[234] at mapToPair at Test.java:157), which has no missing parents
  2015-06-29 16:13:56 INFO  MemoryStore:59 - ensureFreeSpace(4024) called with curMem=386851, maxMem=278302556
  2015-06-29 16:13:56 INFO  MemoryStore:59 - Block broadcast_129 stored as values in memory (estimated size 3.9 KB, free 265.0 MB)
  2015-06-29 16:13:56 INFO  MemoryStore:59 - ensureFreeSpace(2230) called with curMem=390875, maxMem=278302556
  2015-06-29 16:13:56 INFO  MemoryStore:59 - Block broadcast_129_piece0 stored as bytes in memory (estimated size 2.2 KB, free 265.0 MB)
  2015-06-29 16:13:56 INFO  BlockManagerInfo:59 - Added broadcast_129_piece0 in memory on localhost:42836 (size: 2.2 KB, free: 265.3 MB)
  2015-06-29 16:13:56 INFO  BlockManagerMaster:59 - Updated info of block broadcast_129_piece0
  2015-06-29 16:13:56 INFO  SparkContext:59 - Created broadcast 129 from getCallSite at DStream.scala:294
  2015-06-29 16:13:56 INFO  DAGScheduler:59 - Submitting 1 missing tasks from Stage 317 (MappedRDD[234] at mapToPair at Test.java:157)
  2015-06-29 16:13:56 INFO  TaskSchedulerImpl:59 - Adding task set 317.0 with 1 tasks
  2015-06-29 16:13:56 INFO  TaskSetManager:59 - Starting task 0.0 in stage 317.0 (TID 168, localhost, NODE_LOCAL, 7642 bytes)
  2015-06-29 16:13:56 INFO  Executor:59 - Running task 0.0 in stage 317.0 (TID 168)
  2015-06-29 16:13:56 INFO  KafkaRDD:103 - Computing topic test, partition 0 offsets 252661 -&gt; 253192
  2015-06-29 16:13:56 INFO  VerifiableProperties:68 - Verifying properties
  2015-06-29 16:13:56 INFO  VerifiableProperties:68 - Property group.id is overridden to 
  2015-06-29 16:13:56 INFO  VerifiableProperties:68 - Property zookeeper.connect is overridden to 
  2015-06-29 16:13:56 ERROR TaskContextImpl:96 - Error in TaskCompletionListener
  java.lang.NullPointerException
      at org.apache.spark.streaming.kafka.KafkaRDD$KafkaRDDIterator.close(KafkaRDD.scala:158)
      at org.apache.spark.util.NextIterator.closeIfNeeded(NextIterator.scala:63)
      at org.apache.spark.streaming.kafka.KafkaRDD$KafkaRDDIterator$$anonfun$1.apply(KafkaRDD.scala:101)
      at org.apache.spark.streaming.kafka.KafkaRDD$KafkaRDDIterator$$anonfun$1.apply(KafkaRDD.scala:101)
      at org.apache.spark.TaskContextImpl$$anon$1.onTaskCompletion(TaskContextImpl.scala:49)
      at org.apache.spark.TaskContextImpl$$anonfun$markTaskCompleted$1.apply(TaskContextImpl.scala:68)
      at org.apache.spark.TaskContextImpl$$anonfun$markTaskCompleted$1.apply(TaskContextImpl.scala:66)
      at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
      at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47)
      at org.apache.spark.TaskContextImpl.markTaskCompleted(TaskContextImpl.scala:66)
      at org.apache.spark.scheduler.Task.run(Task.scala:58)
      at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:200)
      at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
      at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
      at java.lang.Thread.run(Thread.java:745)
  2015-06-29 16:13:56 ERROR Executor:96 - Exception in task 0.0 in stage 317.0 (TID 168)
  org.apache.spark.util.TaskCompletionListenerException
      at org.apache.spark.TaskContextImpl.markTaskCompleted(TaskContextImpl.scala:76)
      at org.apache.spark.scheduler.Task.run(Task.scala:58)
      at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:200)
      at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
      at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
      at java.lang.Thread.run(Thread.java:745)
  2015-06-29 16:13:56 WARN  TaskSetManager:71 - Lost task 0.0 in stage 317.0 (TID 168, localhost): org.apache.spark.util.TaskCompletionListenerException
      at org.apache.spark.TaskContextImpl.markTaskCompleted(TaskContextImpl.scala:76)
      at org.apache.spark.scheduler.Task.run(Task.scala:58)
      at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:200)
      at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
      at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
      at java.lang.Thread.run(Thread.java:745)
  </code></pre>
  
  <p>My Pom.xml looks like this.</p>
  
  <pre><code>&lt;project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
      xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"&gt;
      &lt;modelVersion&gt;4.0.0&lt;/modelVersion&gt;
      &lt;groupId&gt;alinds&lt;/groupId&gt;
      &lt;artifactId&gt;alind&lt;/artifactId&gt;
      &lt;version&gt;0.0.1-SNAPSHOT&lt;/version&gt;
  
      &lt;build&gt;
          &lt;sourceDirectory&gt;src&lt;/sourceDirectory&gt;
          &lt;plugins&gt;
  
              &lt;plugin&gt;
                  &lt;groupId&gt;org.apache.maven.plugins&lt;/groupId&gt;
                  &lt;artifactId&gt;maven-shade-plugin&lt;/artifactId&gt;
                  &lt;version&gt;2.1&lt;/version&gt;
                  &lt;executions&gt;
                      &lt;execution&gt;
                          &lt;phase&gt;package&lt;/phase&gt;
                          &lt;goals&gt;
                              &lt;goal&gt;shade&lt;/goal&gt;
                          &lt;/goals&gt;
                          &lt;configuration&gt;
                              &lt;filters&gt;
                                  &lt;filter&gt;
                                      &lt;artifact&gt;*:*&lt;/artifact&gt;
                                      &lt;excludes&gt;
                                          &lt;exclude&gt;META-INF/*.SF&lt;/exclude&gt;
                                          &lt;exclude&gt;META-INF/*.DSA&lt;/exclude&gt;
                                          &lt;exclude&gt;META-INF/*.RSA&lt;/exclude&gt;
                                      &lt;/excludes&gt;
                                  &lt;/filter&gt;
                              &lt;/filters&gt;
                              &lt;transformers&gt;
                                  &lt;transformer
                                      implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer"&gt;
                                      &lt;resource&gt;reference.conf&lt;/resource&gt;
                                  &lt;/transformer&gt;
                                  &lt;transformer
                                      implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer"&gt;
                                      &lt;mainClass&gt;
                                        com.alind.sparkStream.Test
                                      &lt;/mainClass&gt;
                                  &lt;/transformer&gt;
                              &lt;/transformers&gt;
                          &lt;/configuration&gt;
                      &lt;/execution&gt;
                  &lt;/executions&gt;
              &lt;/plugin&gt;
  
              &lt;plugin&gt;
                  &lt;groupId&gt;org.apache.maven.plugins&lt;/groupId&gt;
                  &lt;artifactId&gt;maven-compiler-plugin&lt;/artifactId&gt;
                  &lt;version&gt;2.3.2&lt;/version&gt;
                  &lt;configuration&gt;
                      &lt;source&gt;${jdk.version}&lt;/source&gt;
                      &lt;target&gt;${jdk.version}&lt;/target&gt;
                  &lt;/configuration&gt;
              &lt;/plugin&gt;
  
          &lt;/plugins&gt;
      &lt;/build&gt;
      &lt;dependencies&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;1.3.1&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;1.3.1&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-kafka_2.10&lt;/artifactId&gt;
              &lt;version&gt;1.3.1&lt;/version&gt;
  
          &lt;/dependency&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;0.8.2.1&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.8.2.0&lt;/version&gt;
              &lt;scope&gt;provided&lt;/scope&gt;
          &lt;/dependency&gt;
  
          &lt;dependency&gt;
              &lt;groupId&gt;MyOtherProject&lt;/groupId&gt;
              &lt;version&gt;1.0&lt;/version&gt;
  
          &lt;/dependency&gt;
  
      &lt;/dependencies&gt;
      &lt;repositories&gt;
          &lt;repository&gt;
              &lt;id&gt;Spark repository&lt;/id&gt;
              &lt;url&gt;http://www.sparkjava.com/nexus/content/repositories/spark/&lt;/url&gt;
          &lt;/repository&gt;
      &lt;/repositories&gt;
  &lt;/project&gt;
  </code></pre>
  
  <p>And the Spark Driver looks like this...</p>
  
  <pre><code>public class Test {
  
  
  
   static Logger log = Logger.getLogger(Test.class.getName());
  
  
  
      public static void main(String[] args) {
  
          System.setProperty("spark.serializer",
                  "org.apache.spark.serializer.KryoSerializer");
  
  
          SparkConf sparkConf = new SparkConf();
  
          sparkConf.setMaster("spark://clstr:7077");
         // when i run this code from eclipse i change setMaster value to ("local[2]")
          sparkConf.setAppName("alind");
          JavaSparkContext javaSparkContext = new JavaSparkContext(sparkConf);
          JavaStreamingContext javaStreamingContext = new JavaStreamingContext(
                  javaSparkContext, new Duration(100));
  
          Set&lt;String&gt; topics = new HashSet&lt;String&gt;();
          topics.add("test");
  
          Map&lt;String, String&gt; kafkaParams = new HashMap&lt;String, String&gt;();
          kafkaParams.put("metadata.broker.list", "10.20.3.14:9092");
      // Tested this metadata.broker.list with localhost:9092 as well, its not working on cluster with any of these.
          JavaPairInputDStream&lt;String, String&gt; stream = KafkaUtils
                  .createDirectStream(javaStreamingContext, String.class,
                          String.class, StringDecoder.class, StringDecoder.class,
                          kafkaParams, topics);
          stream.print();
  
  
          javaStreamingContext.start();
          javaStreamingContext.awaitTermination();
  
      }
  }
  </code></pre>
  
  <p>If you could give me any idea whats wrong with the local cluster, i will be grateful.
  Seems like something at the kafka end is wrong.</p>