[SPARK-19779][SS]Delete needless tmp file after restart structured streaming job

[gf53520]

What changes were proposed in this pull request?

SPARK-19779

The PR (#17012) fixed restart a Structured Streaming application using hdfs as fileSystem, but also exist a problem that a tmp file of delta file is still reserved in hdfs. And Structured Streaming don't delete the tmp file generated when restart streaming job in future.

How was this patch tested?

unit tests

[srowen]

If the file exists, it is deleted, but no new file is renamed to it -- is that right?

[gf53520]

when restart streaming job , thefinalDeltaFile generated by the first batch is same to a finalDeltaFile generated by the last batch of streaming job before restart. So here don't need rename to create an same file.

[srowen]

I guess my point is, after this change, the file may not exist after this executes. Before, it always existed after this block. I wasn't sure that was the intended behavior change because the purpose seems to be to delete the temp file.

[gf53520]

Yes. This pr just want to delete the needless temp file, and the delta file need exist.

[SparkQA]

Test build #3589 has finished for PR 17124 at commit 5600776.

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

[zsxwing]

Thanks for doing this. Just some minor issues.

[zsxwing]

This comment is not 100% correct, this may also happen in a speculation task.

This PR is just a follow up to delete the temp file that #17012 forgot to do it. IMO, not need to add a comment for it.

[zsxwing]

Instead of adding a new test, I prefer to just add several lines to the above SPARK-19677: Committing a delta file atop an existing one should not fail on HDFS. E.g.

  test("SPARK-19677: Committing a delta file atop an existing one should not fail on HDFS") {
    val conf = new Configuration()
    conf.set("fs.fake.impl", classOf[RenameLikeHDFSFileSystem].getName)
    conf.set("fs.default.name", "fake:///")

    val provider = newStoreProvider(hadoopConf = conf)
    provider.getStore(0).commit()
    provider.getStore(0).commit()

    // Verify we don't leak temp files
    val tempFiles = FileUtils.listFiles(new File(provider.id.checkpointLocation), null, true)
      .asScala.filter(_.getName.contains("temp-"))
    assert(tempFiles.isEmpty)
  }

[gf53520]

@zsxwing I have rewritten the test case.

[zsxwing]

retest this please

[SparkQA]

Test build #73786 has finished for PR 17124 at commit db3f4db.

• This patch fails Scala style tests.
• This patch merges cleanly.
• This patch adds no public classes.

[gf53520]

retest this please

[zsxwing]

ok to test

[SparkQA]

Test build #73803 has finished for PR 17124 at commit c5895e2.

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

[zsxwing]

LGTM. Merging to master, 2.1 and 2.0. Thanks!