wineternity commented on code in PR #38702:

@@ -645,8 +645,11 @@ private[spark] class AppStatusListener(
   override def onTaskEnd(event: SparkListenerTaskEnd): Unit = {
-    // TODO: can this really happen?
-    if (event.taskInfo == null) {
+    // TODO: can taskInfo null really happen?
+    // For resubmitted tasks caused by ExecutorLost, the SparkListenerTaskEnd 
is useless and
+    // will make activeTask in stage to be negative, this will cause stage not 
be removed in
+    // liveStages, and finally cause executor not removed in deadExecutors
+    if (event.taskInfo == null ||  event.reason == Resubmitted) {

Review Comment:
   1. First in TaskSetManager.scala, we handle the executorLost
   ``` scala
     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) {
         for ((tid, info) <- taskInfos if info.executorId == execId) {
           val index = info.index
           // We may have a running task whose partition has been marked as 
           // this partition has another task completed in another stage 
           // We treat it as a running task and will call handleFailedTask 
           if (successful(index) && !info.running && 
!killedByOtherAttempt.contains(tid)) {
             successful(index) = false
             copiesRunning(index) -= 1
             tasksSuccessful -= 1
             // Tell the DAGScheduler that this task was resubmitted so that it 
doesn't think our
             // stage finishes when a total of tasks.size tasks finish.
               tasks(index), Resubmitted, null, Seq.empty, Array.empty, info)
   2. the successful task is added to queue to re-execute in `addPendingTask`
   3. `sched.dagScheduler.taskEnded` is called to tell DAGScheduler to handle 
this, this will post a `CompletionEvent` to eventProcessLoop  
   4. the function `handleTaskCompletion` in DATScheduler.scala will handle 
this message, as the reason `Resubmitted` is kind of a failure, it will call 
the function `handleResubmittedFailure`,  which add the task partitionId to 
ShuffleMapStage's pendingPartitions. Thus this stage will wait this resubmitted 
task to finish again.

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:

For queries about this service, please contact Infrastructure at:

To unsubscribe, e-mail:
For additional commands, e-mail:

Reply via email to