parthshyara commented on code in PR #39011:
URL: https://github.com/apache/spark/pull/39011#discussion_r1573887387
##########
core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala:
##########
@@ -1046,17 +1048,45 @@ private[spark] class TaskSetManager(
/** Called by TaskScheduler when an executor is lost so we can re-enqueue
our tasks */
override def executorLost(execId: String, host: String, reason:
ExecutorLossReason): Unit = {
- // Re-enqueue any tasks that ran on the failed executor if this is a
shuffle map stage,
- // and we are not using an external shuffle server which could serve the
shuffle outputs.
- // The reason is the next stage wouldn't be able to fetch the data from
this dead executor
- // so we would need to rerun these tasks on other executors.
- if (isShuffleMapTasks && !env.blockManager.externalShuffleServiceEnabled
&& !isZombie) {
+ // Re-enqueue any tasks with potential shuffle data loss that ran on the
failed executor
+ // if this is a shuffle map stage, and we are not using an external
shuffle server which
+ // could serve the shuffle outputs or the executor lost is caused by
decommission (which
+ // can destroy the whole host). The reason is the next stage wouldn't be
able to fetch the
+ // data from this dead executor so we would need to rerun these tasks on
other executors.
+ val maybeShuffleMapOutputLoss = isShuffleMapTasks &&
+ (reason.isInstanceOf[ExecutorDecommission] ||
!env.blockManager.externalShuffleServiceEnabled)
Review Comment:
@Ngone51 @mridulm Is the above issue being tracked elsewhere?
--
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]
