title: File filter not working in Spark StreamingContext.fileStream(...) API
body: <p>I'm building a Spark Streaming application where my requirement is to read all pre-existing files in a directory being monitored.</p>
  
  <p>I'm using <code>StreamingContext.fileStream(...)</code> API for this. This API needs one to pass the filter function. In my case I'm always returning <code>true</code> from this as I need to read all the files.
  Also the <code>newFilesOnly</code> flag in <code>StreamingContext.fileStream(...)</code> is set to <code>false</code>.</p>
  
  <p>[Here's API <a href="https://spark.apache.org/docs/1.6.0/api/java/org/apache/spark/streaming/StreamingContext.html#fileStream(java.lang.String,%20scala.Function1,%20boolean,%20scala.reflect.ClassTag,%20scala.reflect.ClassTag,%20scala.reflect.ClassTag)" rel="nofollow">doc]
  </a></p>
  
  <p>But, no matter what filter function returns or <code>newFilesOnly</code> flag is set to, the RDDs created in corresponding DStream are empty.</p>
  
  <p><strong>Here's the code snippet:</strong></p>
  
  <pre><code>val ssc = new StreamingContext(sparkConf, Seconds(30))
  val filterF = new Function[Path, Boolean] {
      def apply(x: Path): Boolean = {
        println("In File " + x.toString) //Prints exisitng file's path as expected
        true
      }
  }
  val strm = ssc.fileStream[LongWritable, Text, TextInputFormat]("s3n://&lt;bucket&gt;/", filterF, false).map(_._2.toString)
  strm.print()     //DOESN'T PRINT ANYTHING
  </code></pre>
  
  <p>I have tried different combinations of return values from filter function and newFilesOnly flag, nothing worked.</p>
  
  <p>If I use <code>StreamingContext.textFileStream(...)</code> instead, it works fine, but reads only new files which is expected behavior from this API.</p>
  
  <p>Am I missing something here? Any help will be appreciated. Thanks in advance!</p>