[SPARK-21009] SparkListenerTaskEnd.taskInfo.accumulables might not be accurate

• Type: Bug
• Status: Resolved
• Priority: Major
• Resolution: Duplicate
• Affects Version/s: 2.2.0
• Component/s: Spark Core
• Assignee: Unassigned
• Reporter: Bogdan Raducanu
• Created: 07/Jun/17 11:54
• Updated: 07/Jun/17 16:44
• Resolved: 07/Jun/17 15:33

Description

The following code reproduces it:

  test("test") {
    val foundMetrics = mutable.Set.empty[String]
    spark.sparkContext.addSparkListener(new SparkListener {
      override def onTaskEnd(taskEnd: SparkListenerTaskEnd): Unit = {
        taskEnd.taskInfo.accumulables.foreach { a =>
          if (a.name.isDefined) {
            foundMetrics.add(a.name.get)
          }
        }
      }
    })
    for (iter <- 0 until 100) {
      foundMetrics.clear()
      println(s"iter = $iter")
      spark.range(10).groupBy().agg("id" -> "sum").collect
      spark.sparkContext.listenerBus.waitUntilEmpty(3000)
      assert(foundMetrics.size > 0)
    }
  }

The problem comes from DAGScheduler.handleTaskCompletion.
The SparkListenerTaskEnd event is sent before updateAccumulators is called, so it might not be up to date.
The code there looks like it needs refactoring.

Issue Links

• duplicates: SPARK-20342 DAGScheduler sends SparkListenerTaskEnd before updating task's accumulators