[SPARK-20342] DAGScheduler sends SparkListenerTaskEnd before updating task's accumulators

• Type: Bug
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 2.2.0
• Fix Version/s: 2.2.1, 2.3.0
• Component/s: Spark Core
• Assignee: Marcelo Masiero Vanzin
• Reporter: Marcelo Masiero Vanzin
• Created: 14/Apr/17 22:17
• Updated: 08/Jul/17 16:26
• Resolved: 08/Jul/17 16:25

Description

Hit this on 2.2, but probably has been there forever. This is similar in spirit to SPARK-20205.

Event is sent here, around L1154:

    listenerBus.post(SparkListenerTaskEnd(
       stageId, task.stageAttemptId, taskType, event.reason, event.taskInfo, taskMetrics))

Accumulators are updated later, around L1173:

    val stage = stageIdToStage(task.stageId)
    event.reason match {
      case Success =>
        task match {
          case rt: ResultTask[_, _] =>
            // Cast to ResultStage here because it's part of the ResultTask
            // TODO Refactor this out to a function that accepts a ResultStage
            val resultStage = stage.asInstanceOf[ResultStage]
            resultStage.activeJob match {
              case Some(job) =>
                if (!job.finished(rt.outputId)) {
                  updateAccumulators(event)

Same thing applies here; UI shows correct info because it's pointing at the mutable TaskInfo structure. But the event log, for example, may record the wrong information.

Issue Links

• blocks: SPARK-18085 SPIP: Better History Server scalability for many / large applications
• is duplicated by: SPARK-21009 SparkListenerTaskEnd.taskInfo.accumulables might not be accurate
• links to: [Github] Pull Request #18233 (vanzin)
• links to: [Github] Pull Request #18393 (vanzin)