[GitHub] spark pull request #22221: [SPARK-25231] : Fix synchronization of executor h...

[xuanyuanking <gi...@git.apache.org>]

Github user xuanyuanking commented on a diff in the pull request:

    https://github.com/apache/spark/pull/22221#discussion_r214508181
  
    --- Diff: core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala ---
    @@ -517,10 +517,10 @@ private[spark] class TaskSchedulerImpl(
           accumUpdates: Array[(Long, Seq[AccumulatorV2[_, _]])],
           blockManagerId: BlockManagerId): Boolean = {
         // (taskId, stageId, stageAttemptId, accumUpdates)
    -    val accumUpdatesWithTaskIds: Array[(Long, Int, Int, Seq[AccumulableInfo])] = synchronized {
    +    val accumUpdatesWithTaskIds: Array[(Long, Int, Int, Seq[AccumulableInfo])] = {
           accumUpdates.flatMap { case (id, updates) =>
             val accInfos = updates.map(acc => acc.toInfo(Some(acc.value), None))
    -        taskIdToTaskSetManager.get(id).map { taskSetMgr =>
    +        Option(taskIdToTaskSetManager.get(id)).map { taskSetMgr =>
    --- End diff --
    
    Just leave a small concern here, original code locked hole scope of ids in `accumUpdates`, after this changing, maybe some id could be found originally but can't find now, because `taskIdToTaskSetManager` can be changed by `removeExecutor` or `statusUpdate`. Its not big problem if executor has been removed.


---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org