Synchronize access to _stageIdToStageMetrics

tashoyan · tashoyan · commit 3ecce56f084e · 2017-11-10T11:19:38.000+03:00
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/ui/SQLListener.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/ui/SQLListener.scala
@@ -210,7 +210,7 @@ class SQLListener(conf: SparkConf) extends SparkListener with Logging {
     }
   }
 
-  override def onStageCompleted(stageCompleted: SparkListenerStageCompleted): Unit = {
+  override def onStageCompleted(stageCompleted: SparkListenerStageCompleted): Unit = synchronized {
     val extraStages = _stageIdToStageMetrics.size - retainedStages
     if (extraStages > 0) {
       val toRemove = _stageIdToStageMetrics.take(extraStages).keys

Original file line number	Diff line number	Diff line change
`@@ -210,7 +210,7 @@ class SQLListener(conf: SparkConf) extends SparkListener with Logging {`
`210`	`210`	`}`
`211`	`211`	`}`
`212`	`212`
`213`		`- override def onStageCompleted(stageCompleted: SparkListenerStageCompleted): Unit = {`
	`213`	`+ override def onStageCompleted(stageCompleted: SparkListenerStageCompleted): Unit = synchronized {`
`214`	`214`	`val extraStages = _stageIdToStageMetrics.size - retainedStages`
`215`	`215`	`if (extraStages > 0) {`
`216`	`216`	`val toRemove = _stageIdToStageMetrics.take(extraStages).keys`