[SPARK-28025][SS] Fix FileContextBasedCheckpointFileManager leaking crc files #25488

HeartSaVioR · 2019-08-18T21:44:24Z

What changes were proposed in this pull request?

This PR fixes the leak of crc files from CheckpointFileManager when FileContextBasedCheckpointFileManager is being used.

Spark hits the Hadoop bug, HADOOP-16255 which seems to be a long-standing issue.

This is there're two renameInternal methods:

public void renameInternal(Path src, Path dst)
public void renameInternal(final Path src, final Path dst, boolean overwrite)

which should be overridden to handle all cases but ChecksumFs only overrides method with 2 params, so when latter is called FilterFs.renameInternal(...) is called instead, and it will do rename with RawLocalFs as underlying filesystem.

The bug is related to FileContext, so FileSystemBasedCheckpointFileManager is not affected.

SPARK-17475 took a workaround for this bug, but SPARK-23966 seemed to bring regression.

This PR deletes crc file as "best-effort" when renaming, as failing to delete crc file is not that critical to fail the task.

Why are the changes needed?

This PR prevents crc files not being cleaned up even purging batches. Too many files in same directory often hurts performance, as well as each crc file occupies more space than its own size so possible to occupy nontrivial amount of space when batches go up to 100000+.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Some unit tests are modified to check leakage of crc files.

HeartSaVioR · 2019-08-18T21:45:51Z

sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/HDFSMetadataLogSuite.scala

      // This check also tests for regressions of SPARK-17475
-      val allFiles = new File(metadataLog.metadataPath.toString).listFiles()
-        .filter(!_.getName.startsWith(".")).toSeq


This filter broke the check of regression on SPARK-17475, hence regression came in.

HeartSaVioR · 2019-08-18T21:48:18Z

sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/HDFSMetadataLogSuite.scala

+      val allFiles = new File(metadataLog.metadataPath.toString).listFiles().toSeq
+      assert(allFiles.size <= 2)
+      assert(allFiles.exists(_.getName == "2"))
+      if (allFiles.size == 2) {


FileSystemBasedCheckpointFileManager is the case of crc file being left: it's not a leak because the origin file exists. It correctly deletes crc file when origin file is requested to be deleted, hence crc files for 0 and 1 don't exist.

HeartSaVioR · 2019-08-18T21:52:59Z

cc. @zsxwing @gaborgsomogyi
Also cc. @skonto as @skonto put efforts to analyze this issue in JIRA issue before, so may be able to point out if I'm missing anything.

HeartSaVioR · 2019-08-18T22:04:24Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/CheckpointFileManager.scala

@@ -327,12 +327,14 @@ class FileContextBasedCheckpointFileManager(path: Path, hadoopConf: Configuratio
  override def renameTempFile(srcPath: Path, dstPath: Path, overwriteIfPossible: Boolean): Unit = {
    import Options.Rename._
    fc.rename(srcPath, dstPath, if (overwriteIfPossible) OVERWRITE else NONE)
+    mayRemoveCrcFile(srcPath)


The ideal behavior would be renaming crc file to be matched with dstPath, but in SPARK-17475 seems like we just decided to remove the crc file, so I guess this is OK.

Is it possible to detect whether checksum is enabled? If it's disabled, we can save an exists call.

NVM. Just saw your latest comment.

Yeah, please feel free to let me know if we prefer alternative approach - I can make a change.

zsxwing · 2019-08-19T00:06:03Z

I remember this Hadoop issue but forget the ticket. Do you find the Hadoop ticket for this? Has it been fixed in newer Hadoop version?

HeartSaVioR · 2019-08-19T00:38:08Z

I haven't found Hadoop ticket for this, but if my understanding of SPARK-17475 is correct, SPARK-17475 was needed because rename/delete on Hadoop API didn't care of checksum file, which seems to be resolved as the fix is not needed for FileSystemBasedCheckpointFileManager to pass the modified test.

HeartSaVioR · 2019-08-19T00:44:34Z

And I'm not 100% sure of intention, but if we intended to disable crc via below line, it doesn't seem to work:

fc.create(path, EnumSet.of(CREATE, OVERWRITE), CreateOpts.checksumParam(ChecksumOpt.createDisabled()))

@skonto already analyzed relevant part but didn't figure out any Hadoop JIRA issue:

https://issues.apache.org/jira/browse/SPARK-28025?focusedCommentId=16862339&page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel#comment-16862339

SparkQA · 2019-08-19T01:19:15Z

Test build #109302 has finished for PR 25488 at commit e27baa9.

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

HeartSaVioR · 2019-08-19T01:26:08Z

As the issue only occurs from temp file, another possible approach would be removing crc file for temp file when FSOutputSummer is being used as OutputStream:

underlyingStream match {
  case e: FSDataOutputStream if e.getWrappedStream.isInstanceOf[FSOutputSummer] =>
    val checksumFile = new Path(tempPath.getParent, s".${tempPath.getName}.crc")
    fm.delete(checksumFile)
  case _ =>
}

adding above block after underlyingStream.close() would also work. This change will touch RenameBasedFSDataOutputStream.

As the annotations of the class describe, FSOutputSummer is Unstable and LimitedPrivate so not sure we would like to rely on this though.

…rc files

SparkQA · 2019-08-19T23:27:55Z

Test build #109365 has finished for PR 25488 at commit c30b045.

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

zsxwing

Overall look goods. Just some minor comments.

@HeartSaVioR could you create an upstream HDFS ticket for this ChecksumFs issue? It would be great that the upstream can fix this so that we can remove this hack in future.

zsxwing · 2019-08-22T20:11:20Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/CheckpointFileManager.scala

  }


  override def delete(path: Path): Unit = {
    try {
      fc.delete(path, true)
+      mayRemoveCrcFile(path)


Looks like org.apache.hadoop.fs.ChecksumFs.delete handles the file deletion correctly, so we don't need this change.

Ah, thanks for pointing out.

I just realized I analyzed incorrectly. I confused classes in stack trace hence confused the root reason of bug. I thought FileContext doesn't seem to pick LocalFs (extends ChecksumFs), but turned out that's not the case. The root reason is, there're two renameInternal methods:

public void renameInternal(Path src, Path dst) public void renameInternal(final Path src, final Path dst, boolean overwrite)

which should be overridden to handle all cases but ChecksumFs only overrides method with 2 params, so when latter is called FilterFs.renameInternal(...) is called instead, and it will do rename with RawLocalFs as underlying filesystem.

I'll update the description of PR and remove this line.

zsxwing · 2019-08-22T22:19:57Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/CheckpointFileManager.scala

@@ -343,5 +345,17 @@ class FileContextBasedCheckpointFileManager(path: Path, hadoopConf: Configuratio
    case _: LocalFs | _: RawLocalFs => true // LocalFs = RawLocalFs + ChecksumFs
    case _ => false
  }
+
+  private def mayRemoveCrcFile(path: Path): Unit = {
+    val checksumFile = new Path(path.getParent, s".${path.getName}.crc")


I just found one bug in this Path API: https://issues.apache.org/jira/browse/HDFS-14762
How about wrap the whole block like this

try { val checksumFile = new Path(path.getParent, s".${path.getName}.crc") if (exists(checksumFile)) { // checksum file exists, deleting it delete(checksumFile) } } catch { case NonFatal(_) => ... }

Nice finding! HDFS-14762 looks odd but we have to live with it. The suggestion looks great.

HeartSaVioR · 2019-08-23T01:43:32Z

Addressed. HADOOP JIRA issue is included in both code comment and PR description. Please take a look again. Thanks in advance!

SparkQA · 2019-08-23T05:10:49Z

Test build #109604 has finished for PR 25488 at commit 376b7eb.

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

zsxwing · 2019-08-23T06:11:24Z

@HeartSaVioR Thanks! I merged to master. Could you submit a backport PR for branch-2.4?

HeartSaVioR · 2019-08-23T06:43:47Z

Thanks for the quick reviewing and merging! I'll raise PR for branch-2.4 soon.

…rc files This PR fixes the leak of crc files from CheckpointFileManager when FileContextBasedCheckpointFileManager is being used. Spark hits the Hadoop bug, [HADOOP-16255](https://issues.apache.org/jira/browse/HADOOP-16255) which seems to be a long-standing issue. This is there're two `renameInternal` methods: ``` public void renameInternal(Path src, Path dst) public void renameInternal(final Path src, final Path dst, boolean overwrite) ``` which should be overridden to handle all cases but ChecksumFs only overrides method with 2 params, so when latter is called FilterFs.renameInternal(...) is called instead, and it will do rename with RawLocalFs as underlying filesystem. The bug is related to FileContext, so FileSystemBasedCheckpointFileManager is not affected. [SPARK-17475](https://issues.apache.org/jira/browse/SPARK-17475) took a workaround for this bug, but [SPARK-23966](https://issues.apache.org/jira/browse/SPARK-23966) seemed to bring regression. This PR deletes crc file as "best-effort" when renaming, as failing to delete crc file is not that critical to fail the task. This PR prevents crc files not being cleaned up even purging batches. Too many files in same directory often hurts performance, as well as each crc file occupies more space than its own size so possible to occupy nontrivial amount of space when batches go up to 100000+. No. Some unit tests are modified to check leakage of crc files. Closes apache#25488 from HeartSaVioR/SPARK-28025. Authored-by: Jungtaek Lim (HeartSaVioR) <kabhwan@gmail.com> Signed-off-by: Shixiong Zhu <zsxwing@gmail.com>

HeartSaVioR commented Aug 18, 2019

View reviewed changes

dongjoon-hyun added the STRUCTURED STREAMING label Aug 19, 2019

[SPARK-28025][SS] Fix FileContextBasedCheckpointFileManager leaking c…

c30b045

…rc files

HeartSaVioR force-pushed the SPARK-28025 branch from e27baa9 to c30b045 Compare August 19, 2019 19:44

zsxwing requested changes Aug 22, 2019

View reviewed changes

Address review comments from zsxwing

376b7eb

asfgit closed this in 406c533 Aug 23, 2019

zsxwing mentioned this pull request Aug 23, 2019

HDFSLogStore leaks crc files delta-io/delta#138

Closed

HeartSaVioR mentioned this pull request Aug 23, 2019

[SPARK-28025][SS][2.4] Fix FileContextBasedCheckpointFileManager leaking crc files #25565

Closed

HeartSaVioR deleted the SPARK-28025 branch August 23, 2019 06:47

holdenk mentioned this pull request Aug 30, 2019

[SPARK-28784][SS] Use CheckpointFileManager in StreamExecution/StreamingQueryManager for checkpoint dirs #25514

Closed

[SPARK-28025][SS] Fix FileContextBasedCheckpointFileManager leaking crc files #25488

[SPARK-28025][SS] Fix FileContextBasedCheckpointFileManager leaking crc files #25488

Uh oh!

Conversation

HeartSaVioR commented Aug 18, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Uh oh!

HeartSaVioR Aug 18, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

HeartSaVioR commented Aug 18, 2019

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

zsxwing commented Aug 19, 2019

Uh oh!

HeartSaVioR commented Aug 19, 2019

Uh oh!

HeartSaVioR commented Aug 19, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

SparkQA commented Aug 19, 2019

Uh oh!

HeartSaVioR commented Aug 19, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

SparkQA commented Aug 19, 2019

Uh oh!

zsxwing left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

HeartSaVioR Aug 23, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

HeartSaVioR commented Aug 23, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

SparkQA commented Aug 23, 2019

Uh oh!

zsxwing commented Aug 23, 2019

Uh oh!

HeartSaVioR commented Aug 23, 2019

Uh oh!

Uh oh!

HeartSaVioR commented Aug 18, 2019 •

edited

Loading

HeartSaVioR Aug 18, 2019 •

edited

Loading

HeartSaVioR commented Aug 19, 2019 •

edited

Loading

HeartSaVioR commented Aug 19, 2019 •

edited

Loading

HeartSaVioR Aug 23, 2019 •

edited

Loading

HeartSaVioR commented Aug 23, 2019 •

edited

Loading