[SPARK-45579][CORE] Catch errors for FallbackStorage.copy

[ukby1234]

What changes were proposed in this pull request?

As documented in the JIRA ticket, FallbackStorage.copy sometimes will throw FileNotFoundException even though we check for file that exists. This will cause the BlockManagerDecommissioner to be stuck in endless loops and prevent executors from exiting.
We should ignore any FileNotFoundException in this case, and set keepRunning to false for all other exceptions for retries.

Why are the changes needed?

Fix a bug documented in the JIRA ticket

Does this PR introduce any user-facing change?

No

How was this patch tested?

Tests weren't added due to difficulty to replicate the race condition.

Was this patch authored or co-authored using generative AI tooling?

No

[dongjoon-hyun]

Thank you for making a PR, @ukby1234 .

[dongjoon-hyun]

Do you think we can have a test coverage here?

spark/core/src/test/scala/org/apache/spark/storage/FallbackStorageSuite.scala

Line 43 in f1ae56b

class FallbackStorageSuite extends SparkFunSuite with LocalSparkContext {

[ukby1234]

Do you think we can have a test coverage here?

spark/core/src/test/scala/org/apache/spark/storage/FallbackStorageSuite.scala

Line 43 in f1ae56b

class FallbackStorageSuite extends SparkFunSuite with LocalSparkContext {

Added a unit test coverage.

[mridulm]

Drop this ? The existing NonFatal block at the end does this currently.

[ukby1234]

This is different from the existing NonFatal block because it will retry the failed blocks but the existing one is really a catch-all and leave some blocks not retried.

[mridulm]

It was not clear from the PR description that this behavior change was being made.
+CC @dongjoon-hyun as you know this part more.

[ukby1234]

There isn't a behavior change. If we remove the added NonFatal block, this section won't get executed. This means there are shuffle blocks that never trigger numMigratedShuffles.incrementAndGet() and the decommissioner will loop forever because the numMigratedShuffles is always less than migratingShuffles.

[dongjoon-hyun]

Is this true?

If we remove the added NonFatal block, this section won't get executed.

We have line 166, doesn't it?

spark/core/src/main/scala/org/apache/spark/storage/BlockManagerDecommissioner.scala

Lines 166 to 168 in 2ab7aa8

	case e: Exception =>
	logError(s"Error occurred during migrating $shuffleBlockInfo", e)
	keepRunning = false

Do you think you can provide a test case as the evidence for your claim, @ukby1234 ?

[ukby1234]

Well this exception is thrown in this catch block, so this line 166 won't get executed.
And updated tests "SPARK-45579: abort for other errors" to show this situation.

[ukby1234]

hmm looks like the SQL test just timed out and I retried a couple times already. cc @dongjoon-hyun

[steveloughran]

1. Does this happen with any fs client other than the s3a one?
2. Does anyone know why it happens?
3. There's a pr up to turn off use of the AWS SDK for its uploads, which will switch back to the classic sequential block read/upload algorithm of everything else. Reviews encouraged HADOOP-18925. S3A: option "fs.s3a.optimized.copy.from.local.enabled" to control CopyFromLocalOperation hadoop#6163

[ukby1234]

I think I can answer 2). It seems shuffle blocks are deleted in between the fs.exists and fs.copyFromLocal calls. From the stack trace linked in the jira ticket, it fails inside the org.apache.hadoop.fs.s3a.impl.CopyFromLocalOperation.checkSource.

[steveloughran]

@ukby1234 thanks

[ukby1234]

@dongjoon-hyun friendly bump

[github-actions]

We're closing this PR because it hasn't been updated in a while. This isn't a judgement on the merit of the PR in any way. It's just a way of keeping the PR queue manageable.
If you'd like to revive this PR, please reopen it and ask a committer to remove the Stale tag!