dongjoon-hyun commented on PR #58402:
URL: https://github.com/apache/spark/pull/58402#issuecomment-5459152468
Thanks for the fix! I went through the two patched failure paths and the
surrounding rollback machinery. The core approach looks right to me.
Reusing `markAsRollingBack()` here is a good choice: since
`ignoreOldTaskAttempts` compares with `>=`, only attempts up to the failed one
are suppressed and the replacement attempt's results still land. And because
the same flag also gates `notifyPartitionCompletion()`, it covers the "blocked
slot" case (where the replacement task set gets a partition pre-marked complete
and never launches), which a `findMissingPartitions()`-side fix would not. Both
patched sites are behind the existing `latestInfo.attemptNumber() !=
task.stageAttemptId` guards, so they always apply to the latest attempt.
A few things I'd like your take on.
**1. Interaction with the `alreadyRollingBack` dedup guard**
`maxAttemptIdToIgnore` is also used as the dedup key for rollback:
```scala
val alreadyRollingBack =
sms.maxAttemptIdToIgnore.contains(sms.latestInfo.attemptNumber())
if (sms.getNextAttemptId > 0 && !alreadyRollingBack) {
rollbackShuffleMapStage(sms, ...)
```
(`rollbackSucceedingStages`, and reached from `submitMissingTasks`.)
By setting `maxAttemptIdToIgnore = N` at barrier-failure time, a subsequent
indeterminate rollback *while the stage is still at attempt N* now skips
`rollbackShuffleMapStage()`. The `submitMissingTasks` rollback block runs
before `makeNewStageAttempt()`, so `latestInfo.attemptNumber()` is still `N`
there and it is reachable -- e.g. a barrier stage that is also statically
indeterminate (`rdd.repartition(n).barrier()`).
What gets skipped is `newShuffleMergeState()`, which the comment right above
that call describes as "creating a new shuffle merge state for the upcoming
retry".
The practical impact today looks small: `canShuffleMergeBeEnabled()`
disables push-based shuffle for barrier stages (`!rdd.isBarrier()`), so the
only real effect is a missed `_shuffleMergeId` bump, which is inert there. But
`maxAttemptIdToIgnore` now carries two meanings -- "invalidation marker" and
"rollback dedup key" -- and this guard silently changes behavior as a result.
Could you add a note about it? If SPARK-35547 ever enables push shuffle for
barrier stages this becomes a real bug.
**2. The producer-side barrier path is left unprotected**
The other barrier branch in the `FetchFailed` handler is unchanged:
```scala
if (mapStage.rdd.isBarrier()) {
mapOutputTracker.unregisterAllMapAndMergeOutput(shuffleId)
```
A zombie/speculative task of the completed barrier producer whose success
arrives after this `unregisterAll` re-registers that partition, and the
following retry's `findMissingPartitions()` drops it -- the same class of
barrier hang.
I agree with the reasoning in the PR description for not marking it (you
must not invalidate a producer retry that is already running), and the
regression test for it is a good addition. Since this leaves a pre-existing
hole though, could you call it out explicitly as a remaining gap in the
description, or file a follow-up JIRA? It also seems narrowable -- marking only
when `mapStage` has not yet moved to a new attempt would cover the straggler
case without touching a running retry.
**3. Minor**
- `Stage.markAsRollingBack()`'s `/** Mark the latest attempt as rollback */`
could use the same update as the field scaladoc, since it now also covers
barrier invalidation.
- The new comment replaced `// Mark all the map as broken in the map stage,
to ensure retry all the tasks on resubmitted stage attempt.`, but that line
explains the `unregisterAllMapAndMergeOutput()` call below it, while the new
one explains `markAsRollingBack()`. I'd keep both.
Test coverage looks thorough -- before/after resubmission x 1-2 failed
attempts, the `barrier=true/false` contrast confirming determinate non-barrier
output is still reused, and the producer-retry regression. Using
`scheduler.handleTaskCompletion()` directly to avoid the 200ms
`scheduleResubmit()` timer event makes sense and there is precedent in the
suite.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
