[jira] [Resolved] (SPARK-57994) Buffer shuffle-migration relocation that races map output registration

["Wenchen Fan (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/SPARK-57994?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Wenchen Fan resolved SPARK-57994.
---------------------------------
    Fix Version/s: 4.3.0
       Resolution: Fixed

Issue resolved by pull request 57075
[https://github.com/apache/spark/pull/57075]

> Buffer shuffle-migration relocation that races map output registration
> ----------------------------------------------------------------------
>
>                 Key: SPARK-57994
>                 URL: https://issues.apache.org/jira/browse/SPARK-57994
>             Project: Spark
>          Issue Type: Improvement
>          Components: Spark Core
>    Affects Versions: 5.0.0
>            Reporter: ZequnLin
>            Assignee: ZequnLin
>            Priority: Minor
>              Labels: pull-request-available
>             Fix For: 4.3.0
>
>
> h2. What's the problem?
> During executor decommission, a migrated shuffle block's driver-side location can be silently dropped, leaving the map output pointing at the dead origin executor. That entry is then nulled when the origin executor is removed, surfacing downstream as a null MapStatus (NPE on MapStatus.location()) or a fetch failure.
> h2. Root cause
> The problem is a race between two independently-threaded channels that both start when a shuffle map file is committed on the decommissioning executor:
>  * *Registration:* task completion -> DAG scheduler event loop -> {+}_registerMapOutput_{+}, recording the map output at the origin (decommissioning) executor.
>  * *Relocation:* the migrated block is uploaded to a peer; the peer reports it; the report is routed off the event loop to {+}_MapOutputTracker.updateMapOutput_{+}, relocating the map output to the peer.
> The two run on different threads with no ordering guarantee. When the relocation beats the registration, the map output is not tracked yet, so +_ShuffleStatus.updateMapOutput_+ hits the untracked branch, logs a warning, and drops the relocation with no retry. Registration then records the stale origin location; when the decommissioned executor is later removed ({+}_removeOutputsOnExecutor_{+}), that slot is nulled.
> h2. Fix
> Buffer the racing relocation (keyed by mapId, the specific task attempt) instead of dropping it, and replay it once the map output is registered, so the migrated (peer) location wins over the origin location. A later relocation for the same untracked mapId overwrites the pending entry (latest migration wins); a buffered entry that never gets a matching registration simply never replays and is freed when the shuffle unregisters.
> The behavior will be gated behind a new internal config +_spark.storage.decommission.shuffleBlocks.bufferRacingMigrations_+ (default true), which preserves the legacy log-and-drop behavior when disabled.



--
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