[jira] [Created] (SPARK-45858) Consistent FetchFailed/NoSuchFileExceptions when decommissioning is enabled

["Alibi Yeslambek (Jira)" <ji...@apache.org>]

Alibi Yeslambek created SPARK-45858:
---------------------------------------

             Summary: Consistent FetchFailed/NoSuchFileExceptions when decommissioning is enabled
                 Key: SPARK-45858
                 URL: https://issues.apache.org/jira/browse/SPARK-45858
             Project: Spark
          Issue Type: Bug
          Components: Spark Core
    Affects Versions: 3.5.0
            Reporter: Alibi Yeslambek


Decommissioning causes FetchFailures with NoSuchFileException due to multiple tasks on the same partition from different stage attempts sharing a single MapStatus object. Is there any workaround/config flag that I’m missing that will fix the issue or is this rather a bug?

*Example*
Here are same tasks from different stage attempts for the same partition:
{code:java}
INFO [2023-11-07T17:50:03.399091Z] org.apache.spark.scheduler.TaskSetManager: Starting task 16.0 in stage 11.1 (TID 1810) (10.0.158.211, executor 5, partition 81, PROCESS_LOCAL, 4743 bytes) taskResourceAssignments Map()
INFO [2023-11-07T17:51:20.229168Z] org.apache.spark.scheduler.TaskSetManager: Starting task 13.0 in stage 11.2 (TID 1836) (10.0.187.67, executor 6, partition 81, PROCESS_LOCAL, 4743 bytes) taskResourceAssignments Map() {code}
The latest mapStatus.location for partition 81 will be the latest succeeded task (exec-6) , i.e:
{code:java}
mapStatus(81).location = BlockManagerId(6, 10.0.187.67, 7079, None){code}
Which means that multiple MapIDs point to the same MapIndex and share one MapStatus object. In this example:

 
{code:java}
mapIdToMapIndex(1810) = 81
mapIdToMapIndex(1836) = 81 
{code}
Now if we decommission exec-5, all of its blocks (including 1810) will be migrated and driver mapStatuses will updated.
{code:java}
INFO [2023-11-07T17:57:23.545274Z] org.apache.spark.ShuffleStatus: Updating map output for 1810 to BlockManagerId(4, 10.0.153.179, 7079, None){code}
Which updates mapStatus.location for partition 81 to exec-4:
{code:java}
mapStatus(81).location = BlockManagerId(4, 10.0.153.179, 7079, None){code}
And when a task from different stage tries to fetch block for {{MapId: 1836}} , driver will return it’s location as exec-4, whereas in fact it is still on exec-6. The task will fail with FetchFailure caused by NoSuchFileException, because the actual block is in exec-6.
{code:java}
WARN [2023-11-07T17:58:40.008602Z] org.apache.spark.scheduler.TaskSetManager: Lost task 14.0 in stage 16.0 (TID 1988) (10.0.156.83 executor 24): FetchFailed(BlockManagerId(4, 10.0.153.179, 7079, None), shuffleId=5, mapIndex=81, mapId=1836, reduceId=84, message=
org.apache.spark.shuffle.FetchFailedException
at org.apache.spark.storage.ShuffleBlockFetcherIterator.throwFetchFailedException(ShuffleBlockFetcherIterator.scala:1167)
at org.apache.spark.storage.ShuffleBlockFetcherIterator.next(ShuffleBlockFetcherIterator.scala:903)
at org.apache.spark.storage.ShuffleBlockFetcherIterator.next(ShuffleBlockFetcherIterator.scala:84)
at org.apache.spark.util.CompletionIterator.next(CompletionIterator.scala:29)
at scala.collection.Iterator$$anon$11.nextCur(Iterator.scala:486)
at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:492)
at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460)
at org.apache.spark.util.CompletionIterator.hasNext(CompletionIterator.scala:31)
at org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37)
at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460)
at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage10.sort_addToSorter_0$(generated.java:31)
at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage10.processNext(generated.java:43)
at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
at org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$1.hasNext(WholeStageCodegenExec.scala:776)
at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage12.smj_findNextJoinRows_0$(generated.java:40)
at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage12.processNext(generated.java:101)
at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
at org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$2.hasNext(WholeStageCodegenExec.scala:795)
at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460)
at org.apache.spark.shuffle.sort.BypassMergeSortShuffleWriter.write(BypassMergeSortShuffleWriter.java:140)
at org.apache.spark.shuffle.ShuffleWriteProcessor.write(ShuffleWriteProcessor.scala:59)
at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:99)
at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:52)
at org.apache.spark.scheduler.Task.run(Task.scala:131)
at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$1(Executor.scala:516)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1465)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:519)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
at java.base/java.lang.Thread.run(Thread.java:840)
Caused by: java.lang.RuntimeException: java.nio.file.NoSuchFileException: /var/data/spark-local-dir/blockmgr-53a043ab-1d2a-4d88-abd5-f6040ebf409d/2f/shuffle_5_1836_0.index
at java.base/sun.nio.fs.UnixException.translateToIOException(UnixException.java:92)
at java.base/sun.nio.fs.UnixException.rethrowAsIOException(UnixException.java:106)
at java.base/sun.nio.fs.UnixException.rethrowAsIOException(UnixException.java:111)
at java.base/sun.nio.fs.UnixFileSystemProvider.newByteChannel(UnixFileSystemProvider.java:218)
at java.base/java.nio.file.Files.newByteChannel(Files.java:380)
at java.base/java.nio.file.Files.newByteChannel(Files.java:432)
at org.apache.spark.shuffle.IndexShuffleBlockResolver.getBlockData(IndexShuffleBlockResolver.scala:574)
at org.apache.spark.storage.BlockManager.getLocalBlockData(BlockManager.scala:682)
at org.apache.spark.network.netty.NettyBlockRpcServer.$anonfun$receive$5(NettyBlockRpcServer.scala:85)
at org.apache.spark.network.netty.NettyBlockRpcServer.$anonfun$receive$5$adapted(NettyBlockRpcServer.scala:71)
at scala.collection.TraversableLike.$anonfun$flatMap$1(TraversableLike.scala:293){code}

The comment for MapStatuses says that “we choose to ignore it because it’s rare”, but it happens very frequently if decommission is enabled ([link|https://github.com/apache/spark/blob/06d8cbe073499ff16bca3165e2de1192daad3984/core/src/main/scala/org/apache/spark/MapOutputTracker.scala#L85-L94])
{quote}_Even though in theory a task may run multiple times (due to speculation, stage retries, etc.), in practice the likelihood of a map output being available at multiple locations is so small that we choose to ignore that case and store only a single location for each output._{quote}

*Potential Solutions*
I am not 100% sure if we need to keep track of all MapIds per MapIndex, or only MapId from the latest stage attempt. That’s why there are two potential solutions depending on the answer:
 # If we only need to track MapIds from the latest stage attempt: Clear previous {{{}MapId{}}}s for requested {{mapIndex}} from {{mapIdToMapIndex}} when a new map output is added [here|https://github.com/apache/spark/blob/1e93c408e19f4ce8cec8220fd5eb6c1cb76468ff/core/src/main/scala/org/apache/spark/MapOutputTracker.scala#L161-L168].
 # In case we want to track all mapIds per MapIndex: We change {{mapStatuses}} from tracking one status per partition into array of statuses per partition. ([src|https://github.com/apache/spark/blob/1e93c408e19f4ce8cec8220fd5eb6c1cb76468ff/core/src/main/scala/org/apache/spark/MapOutputTracker.scala#L94])

I am happy to contribute a fix, just want to make sure I am on the right track and not missing any extra context.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)

---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org