Chao Sun created SPARK-59094:
--------------------------------
Summary: Ignore late successes from invalidated barrier shuffle
attempts
Key: SPARK-59094
URL: https://issues.apache.org/jira/browse/SPARK-59094
Project: Spark
Issue Type: Bug
Components: Spark Core
Affects Versions: 5.0.0
Reporter: Chao Sun
A deterministic barrier shuffle stage can accept a late successful task result
after failure recovery has invalidated the entire stage attempt. The old result
can remove a participant from the replacement barrier stage or mark a
replacement partition complete before it launches.
The [barrier execution
contract|https://spark.apache.org/docs/latest/api/java/org/apache/spark/rdd/RDD.html#barrier()]
requires all tasks in a barrier stage to be relaunched when one task fails.
Deterministic output does not make it safe to omit one of those participants.
Upstream draft PR:
[apache/spark#58402|https://github.com/apache/spark/pull/58402].
h3. Reproduction ordering
# Run a deterministic two-partition {{RDD.barrier().mapPartitions(...)}}
followed by a normal shuffle and {{collect()}}, using {{SortShuffleManager}}.
Both tasks call {{BarrierTaskContext.barrier()}}. Let partition 0 finish its
shuffle output, but delay delivery of its successful completion to the driver.
# Fail partition 1 once. Spark fails the barrier stage and unregisters all of
its map outputs so that both partitions should run in the next attempt.
# Deliver partition 0's old success either before resubmission, or after the
replacement task set is submitted but before its tasks launch.
A native local-mode reproducer uses an executor plugin's {{onTaskSucceeded}}
hook to hold only the completed task's success delivery. The peer failure,
scheduler recovery, shuffle files, and downstream read use normal Spark
execution; no constructed map statuses or scheduler-state writes are used. For
the second ordering, an unrelated one-task job temporarily occupies one of the
two execution slots.
h3. Reproduction results
Reproduced on a fresh public Spark {{5.0.0-SNAPSHOT}} build at upstream commit
{{024120dc960517021a887b51865220f544340362}}, using Scala {{2.13.18}}. The
native runs used Java {{21.0.12}} and {{SortShuffleManager}}; the RDD reports
{{DETERMINATE}} output. The probe configures a three-second barrier timeout.
|| Delivery ordering || Unpatched behavior ||
| After failure, before resubmission | The two-task attempt is retried with
only one task, whose context still expects two participants. Barrier
synchronization times out. A two-attempt limit aborts the job; a four-attempt
limit allows an additional complete retry and successful readback. |
| After resubmission, with one slot temporarily occupied | The replacement has
{{successful=[true,false]}} and no running tasks. It does not launch after the
slot is released and before the 20-second job watchdog explicitly cancels it. |
The watchdog deadline is measured from the main job's start, not from slot
release; the latter precedes cancellation by about 17 seconds in this run. This
is a bounded stall followed by probe cancellation, not an observed permanent
hang or a Spark-triggered abort.
Late success after resubmission without deliberately blocked slots, a no-delay
barrier control, and an ordinary non-barrier control all succeed. The unblocked
case releases success on stage submission; it does not force delivery to wait
until both replacement tasks have launched.
With [candidate fix
daef30b0e585|https://github.com/sunchao/spark/commit/daef30b0e585c80786c2cc1f9304ce81204a190f],
all six matched native scenarios succeed and assert {{Seq((0, 0), (1, 1))}}.
Every barrier retry contains both tasks; the blocked case retains
{{successful=[false,false]}} until both tasks launch after slot release. All
completed scenarios satisfy their probe preconditions.
Separately, direct ScalaTest runs on JDK 17 give six new regression failures
and 225 passes in the unpatched {{DAGSchedulerSuite}}, then 352 passes across
the patched {{DAGSchedulerSuite}} and {{TaskSchedulerImplSuite}}. This
validates the tested development snapshot, not a released version or a backport.
h3. Cause and proposed scope
In [DAGScheduler at upstream commit
024120dc960517021a887b51865220f544340362|https://github.com/apache/spark/blob/024120dc960517021a887b51865220f544340362/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala#L3334],
accepted old successes can notify the task scheduler that a partition is
complete and register its map output. A barrier task failure, including a fetch
failure within a barrier shuffle stage, clears that failed stage's outputs
without marking the invalidated attempt's results to be ignored. Before
resubmission, a late registration changes the missing-partition set; after
resubmission, the completion notification can suppress a required participant.
The proposed change is limited to the two paths that clear the outputs of a
barrier shuffle stage that has itself failed:
* An ordinary barrier task failure.
* A fetch failure within that barrier shuffle stage.
Use the existing stage-attempt rollback marker to invalidate successful
completions from that failed attempt, including the interval before a new
attempt exists. Preserve valid replacement-attempt results and Spark's existing
reuse of old deterministic results for ordinary non-barrier stages. Do not
change the separate upstream barrier-producer invalidation path: a fetch
failure reported by a downstream consumer must not invalidate an
already-running retry of its barrier producer merely because that producer's
outputs are cleared.
This is related to
[SPARK-45182|https://issues.apache.org/jira/browse/SPARK-45182],
[SPARK-54556|https://issues.apache.org/jira/browse/SPARK-54556], and
[SPARK-54956|https://issues.apache.org/jira/browse/SPARK-54956], which address
old results during indeterminate or checksum-driven rollback.
[SPARK-25161|https://issues.apache.org/jira/browse/SPARK-25161] covers barrier
failure handling, including old failure events. This report concerns successful
completions from a deterministic barrier attempt that has been invalidated.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
