[SPARK-21357][DStreams] FileInputDStream not remove out of date RDD

[shaofei007]

What changes were proposed in this pull request?

         class FileInputDStream

 [line 162]   protected[streaming] override def clearMetadata(time: Time) {
    batchTimeToSelectedFiles.synchronized {
      val oldFiles = batchTimeToSelectedFiles.filter(_._1 < (time - rememberDuration))
      batchTimeToSelectedFiles --= oldFiles.keys

The above code does not remove the old generatedRDDs. "super.clearMetadata(time)" was added to the beginning of clearMetadata to remove the old generatedRDDs.

How was this patch tested?

At the end of clearMetadata, the testing code (print the number of generatedRDDs) was added to check the old RDDS were removed manually.

[srowen]

Seems reasonable, but, I am sort of not sure why it hasn't come up before. @tdas?

[SparkQA]

Test build #3848 has finished for PR 18718 at commit 5a34428.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[shaofei007]

@tdas
how do you think about this PR please?

@asfgit
Can we close it please?

[srowen]

@shaofei007 I actually think your fix is correct, so I wouldn't close it. The comment says: /** Clear the old time-to-files mappings along with old RDDs */ But only the superclass method clears the old RDDs. In the absence of any comments to the contrary, I'd merge this, at least to master

[srowen]

Merged to master