title: NullPointerException running a Spark job
body: <p>I am running a job on Spark in standalone mode, version 1.2.0</p>
  
  <p>The first operation I am doing is taking an RDD of folder paths, and generating an RDD of file names, composed of the files reside in each folder:</p>
  
  <pre><code>JavaRDD&lt;String&gt; filePaths = paths.mapPartitions(new FoldersToFiles()).repartition(defaultPartitions);
  </code></pre>
  
  <p>Where the inner implementation of FoldersToFiles class is:</p>
  
  <pre><code>@Override
  public Iterable&lt;String&gt; call(Iterator&lt;String&gt; pathsIterator) throws Exception {
      List&lt;String&gt; filesPath = new ArrayList&lt;String&gt;();
      if (pathsIterator != null) {
          while (pathsIterator.hasNext()) {
              try {
                  String currFolder = pathsIterator.next();
                  Path currPath = new Path(currFolder);
                  FileSystem fs = FileSystem.get(currPath.toUri(), new Configuration(true));
                  FileStatus[] files = fs.listStatus(currPath);
                  List&lt;FileStatus&gt; filesList = Arrays.asList(files);
                  List&lt;String&gt; filesPathsStr = new Utils().convertFileStatusToPath(filesList);
                  filesPath.addAll(filesPathsStr);
              } catch(Exception e) {
                  log.error("Error during file names extraction: " + e.getMessage());
              }
          }
      }
      if(filesPath == null || filesPath.isEmpty()) {
          log.error("Warning: files path list is null or empty!! Given Path Iterator is: " + pathsIterator.toString());
      }
      return filesPath;
  }
  </code></pre>
  
  <p>When running the job on cluster, I get the following error:</p>
  
  <pre><code>520983 [task-result-getter-1] WARN org.apache.spark.scheduler.TaskSetManager  - Lost task 33.0 in stage 1.0 (TID 1033, hadoop-w-8.c.taboola-qa-01.internal): java.lang.NullPointerException
  at org.apache.spark.api.java.JavaRDDLike$$anonfun$fn$4$1.apply(JavaRDDLike.scala:140)
  at org.apache.spark.api.java.JavaRDDLike$$anonfun$fn$4$1.apply(JavaRDDLike.scala:140)
  at org.apache.spark.rdd.RDD$$anonfun$13.apply(RDD.scala:601)
  at org.apache.spark.rdd.RDD$$anonfun$13.apply(RDD.scala:601)
  at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:35)
  at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:263)
  at org.apache.spark.rdd.RDD.iterator(RDD.scala:230)
  at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:61)
  at org.apache.spark.scheduler.Task.run(Task.scala:56)
  at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:196)
  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>So the error is not directly inside my code. However, looking at the relevant line in Spark code:</p>
  
  <pre><code>  /**
     * Return a new RDD by applying a function to each partition of this RDD.
     */
    def mapPartitions[U](f: FlatMapFunction[java.util.Iterator[T], U]): JavaRDD[U] = {
      def fn = (x: Iterator[T]) =&gt; asScalaIterator(f.call(asJavaIterator(x)).iterator())
      JavaRDD.fromRDD(rdd.mapPartitions(fn)(fakeClassTag[U]))(fakeClassTag[U])
    }
  </code></pre>
  
  <p>(line 140 in which the exception happens is the first one)</p>
  
  <p>It probably related to (and this is actually the first mapPartitions in my job, so it makes sense) the code I mentioned above, however I cannot understand why.</p>