[
https://issues.apache.org/jira/browse/SPARK-59094?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Chao Sun resolved SPARK-59094.
------------------------------
Fix Version/s: 4.4.0
5.0.0
Resolution: Fixed
Issue resolved by pull request 58402
[https://github.com/apache/spark/pull/58402]
> 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
> Assignee: Chao Sun
> Priority: Major
> Labels: pull-request-available
> Fix For: 4.4.0, 5.0.0
>
>
> 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]
