ZequnLin created SPARK-57994:
--------------------------------
Summary: 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
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: [email protected]
For additional commands, e-mail: [email protected]
