minor fix

Author: carsonwang

sql/core/src/main/scala/org/apache/spark/sql/SQLContext.scala

@@ -1285,6 +1285,10 @@ object SQLContext {
     Option(instantiatedContext.get())
   }
 
+  private[sql] def clearSqlListener(listener: SQLListener): Unit = {
+    sqlListener.compareAndSet(listener, null)
+  }
+
   /**
    * Changes the SQLContext that will be returned in this thread and its children when
    * SQLContext.getOrCreate() is called. This can be used to ensure that a given thread receives
@@ -1335,9 +1339,10 @@ object SQLContext {
   private[sql] def createListenerAndUI(sc: SparkContext): SQLListener = {
     if (sqlListener.get() == null) {
       val listener = new SQLListener(sc.conf)
-      sqlListener.compareAndSet(null, listener)
-      sc.addSparkListener(listener)
-      sc.ui.foreach(new SQLTab(listener, _))
+      if (sqlListener.compareAndSet(null, listener)) {
+        sc.addSparkListener(listener)
+        sc.ui.foreach(new SQLTab(listener, _))
+      }
     }
     sqlListener.get()
   }

sql/core/src/main/scala/org/apache/spark/sql/execution/metric/SQLMetrics.scala

@@ -79,7 +79,7 @@ private[sql] class LongSQLMetricValue(private var _value : Long) extends SQLMetr
  * A specialized long Accumulable to avoid boxing and unboxing when using Accumulator's
  * `+=` and `add`.
  */
-private[sql] class LongSQLMetric private[metric] (name: String, param: LongSQLMetricParam)
+private[sql] class LongSQLMetric private[metric](name: String, param: LongSQLMetricParam)
   extends SQLMetric[LongSQLMetricValue, Long](name, param) {
 
   override def +=(term: Long): Unit = {

sql/core/src/main/scala/org/apache/spark/sql/execution/ui/SQLListener.scala

@@ -21,6 +21,7 @@ import scala.collection.mutable
 
 import org.apache.spark.annotation.DeveloperApi
 import org.apache.spark.scheduler._
+import org.apache.spark.sql.execution.SQLExecution
 import org.apache.spark.sql.execution.SparkPlanInfo
 import org.apache.spark.sql.execution.metric.{LongSQLMetricValue, SQLMetricValue, SQLMetricParam}
 import org.apache.spark.{JobExecutionStatus, Logging, SparkConf}
@@ -59,8 +60,6 @@ private[sql] class SQLEventRegister extends SparkListenerEventRegister {
 
 private[sql] class SQLListener(conf: SparkConf) extends SparkListener with Logging {
 
-  val EXECUTION_ID_KEY = "spark.sql.execution.id"
-
   private val retainedExecutions = conf.getInt("spark.sql.ui.retainedExecutions", 1000)
 
   private val activeExecutions = mutable.HashMap[Long, SQLExecutionUIData]()
@@ -112,7 +111,7 @@ private[sql] class SQLListener(conf: SparkConf) extends SparkListener with Loggi
   }
 
   override def onJobStart(jobStart: SparkListenerJobStart): Unit = {
-    val executionIdString = jobStart.properties.getProperty(EXECUTION_ID_KEY)
+    val executionIdString = jobStart.properties.getProperty(SQLExecution.EXECUTION_ID_KEY)
     if (executionIdString == null) {
       // This is not a job created by SQL
       return

sql/core/src/test/scala/org/apache/spark/sql/test/SharedSQLContext.scala

@@ -56,6 +56,7 @@ trait SharedSQLContext extends SQLTestUtils {
     try {
       if (_ctx != null) {
         _ctx.sparkContext.stop()
+        SQLContext.clearSqlListener(_ctx.listener)
         _ctx = null
       }
     } finally {