The exception that occurs is given below. Basically, the listStatus API get the list of files from the underlying file system and then calls the user-provided PathFilter object (CustomPathFilter in my case) to filter the necessary files. As part of the accepting the file, I try to fetch the mod time of the file. Between these two steps of listing the file names and getting the file mod time, the file may have been deleted. This race condition leads to this error. Unless files are being constantly deleted, the likelihood of this error occurring is low and the next attempt would fail. If files are deleted in bursts then this should be fine. 

A cleaner approach would have been to do the listing as an atomic operation from the POV of the filesystem. But that is beyond our control. So this retry is basically an optimistic attempt.

---------------------------------------------
java.io.FileNotFoundException: File does not exist: hdfs://ip-10-191-57-33.ec2.internal:9000/data/file-12-23-23-01-26-724-1-2378
  at org.apache.hadoop.hdfs.DistributedFileSystem.getFileStatus(DistributedFileSystem.java:517)
  at org.apache.spark.streaming.dstream.FileInputDStream$CustomPathFilter.accept(FileInputDStream.scala:212)
  at org.apache.hadoop.fs.FileSystem.listStatus(FileSystem.java:845)
  at org.apache.hadoop.fs.FileSystem.listStatus(FileSystem.java:867)
  at org.apache.spark.streaming.dstream.FileInputDStream.findNewFiles(FileInputDStream.scala:118)
  at org.apache.spark.streaming.dstream.FileInputDStream.compute(FileInputDStream.scala:79)
  at org.apache.spark.streaming.DStream.getOrCompute(DStream.scala:287)
  at org.apache.spark.streaming.dstream.MappedDStream.compute(MappedDStream.scala:35)
  at org.apache.spark.streaming.DStream.getOrCompute(DStream.scala:287)
  at org.apache.spark.streaming.dstream.FlatMappedDStream.compute(FlatMappedDStream.scala:35)
  at org.apache.spark.streaming.DStream.getOrCompute(DStream.scala:287)
  at org.apache.spark.streaming.dstream.MappedDStream.compute(MappedDStream.scala:35)
  at org.apache.spark.streaming.DStream.getOrCompute(DStream.scala:287)
  at org.apache.spark.streaming.dstream.ForEachDStream.generateJob(ForEachDStream.scala:38)
  at org.apache.spark.streaming.DStreamGraph$$anonfun$1.apply(DStreamGraph.scala:110)
  at org.apache.spark.streaming.DStreamGraph$$anonfun$1.apply(DStreamGraph.scala:110)
  at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:251)
  at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:251)
  at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
  at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47)
  at scala.collection.TraversableLike$class.flatMap(TraversableLike.scala:251)
  at scala.collection.AbstractTraversable.flatMap(Traversable.scala:105)
  at org.apache.spark.streaming.DStreamGraph.generateJobs(DStreamGraph.scala:110)
  at org.apache.spark.streaming.scheduler.JobGenerator.generateJobs(JobGenerator.scala:140)
  at org.apache.spark.streaming.scheduler.JobGenerator.org$apache$spark$streaming$scheduler$JobGenerator$$processEvent(JobGenerator.scala:87)
  at org.apache.spark.streaming.scheduler.JobGenerator$$anonfun$2$$anon$1$$anonfun$receive$1.applyOrElse(JobGenerator.scala:46)
  at akka.actor.ActorCell.receiveMessage(ActorCell.scala:498)
  at akka.actor.ActorCell.invoke(ActorCell.scala:456)
  at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:237)
  at akka.dispatch.Mailbox.run(Mailbox.scala:219)
  at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:386)
  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)