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-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
