[SPARK-32165][SQL] Ensure Spark only initiates SharedState once across SparkSessions

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

@@ -132,7 +132,7 @@ class SparkSession private(
   @Unstable
   @transient
   lazy val sharedState: SharedState = {
-    existingSharedState.getOrElse(new SharedState(sparkContext, initialSessionOptions))
+    existingSharedState.getOrElse(SharedState.get(sparkContext, initialSessionOptions))
   }
 
   /**

sql/core/src/main/scala/org/apache/spark/sql/internal/SharedState.scala

@@ -171,6 +171,19 @@ private[sql] class SharedState(
 
 object SharedState extends Logging {
   @volatile private var fsUrlStreamHandlerFactoryInitialized = false
+  @volatile private var _sharedState: SharedState = _
+
+  def get(sparkContext: SparkContext, initialConfigs: scala.collection.Map[String, String])
+    : SharedState = {
+    if (_sharedState == null) {
+      synchronized {
+        if (_sharedState == null) {
+          _sharedState = new SharedState(sparkContext, initialConfigs)
+        }
+      }
+    }
+    _sharedState
+  }
 
   private def setFsUrlStreamHandlerFactory(conf: SparkConf, hadoopConf: Configuration): Unit = {
     if (!fsUrlStreamHandlerFactoryInitialized &&

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

@@ -17,6 +17,8 @@
 
 package org.apache.spark.sql
 
+import scala.collection.JavaConverters._
+
 import org.scalatest.BeforeAndAfterEach
 
 import org.apache.spark.{SparkConf, SparkContext, SparkException, SparkFunSuite}
@@ -196,6 +198,71 @@ class SparkSessionBuilderSuite extends SparkFunSuite with BeforeAndAfterEach {
     assert(postFirstCreation == postSecondCreation)
   }
 
+  test("SPARK-32165: SparkContext only register one SQLAppStatusListener") {
+    val conf = new SparkConf()
+      .setMaster("local")
+      .setAppName("test-register-one-SQLAppStatusListener")
+    val context = new SparkContext(conf)
+    SparkSession
+      .builder()
+      .sparkContext(context)
+      .master("local")
+      .getOrCreate()
+      // Touches the sessionState so it initiate SQLAppStatusListener and ExecutionListenerBus
+      .sessionState
+
+    assert(countListener("SQLAppStatusListener", context) === 1)
+    assert(countListener("ExecutionListenerBus", context) === 1)
+    val postFirstCreation = context.listenerBus.listeners.size()
+    SparkSession.clearActiveSession()
+    SparkSession.clearDefaultSession()
+
+    SparkSession
+      .builder()
+      .sparkContext(context)
+      .master("local")
+      .getOrCreate()
+      .sessionState
+    assert(countListener("SQLAppStatusListener", context) === 1)
+    assert(countListener("ExecutionListenerBus", context) === 2)
+    val postSecondCreation = context.listenerBus.listeners.size()
+    SparkSession.clearActiveSession()
+    SparkSession.clearDefaultSession()
+    // Minus 1 because the listener `ExecutionListenerBus` is created per SparkSession
+    assert(postFirstCreation == postSecondCreation - 1)
+    context.stop()
+  }
+
+  test("SPARK-32165: Ensure only initiates SharedState once") {
+    val conf = new SparkConf()
+      .setMaster("local")
+      .setAppName("test-initiates-one-shared-state")
+    val context = new SparkContext(conf)
+    val sharedState1 = SparkSession
+      .builder()
+      .sparkContext(context)
+      .master("local")
+      .getOrCreate()
+      .sharedState
+
+    SparkSession.clearActiveSession()
+    SparkSession.clearDefaultSession()
+
+    val sharedState2 = SparkSession
+      .builder()
+      .sparkContext(context)
+      .master("local")
+      .getOrCreate()
+      .sharedState
+
+    assert(sharedState1.eq(sharedState2))
+  }
+
+  private def countListener(listener: String, context: SparkContext): Int = {
+    val listeners = context.listenerBus.listeners.asScala
+    listeners.count(_.getClass.getSimpleName === listener)
+  }
+
   test("SPARK-31532: should not propagate static sql configs to the existing" +
     " active/default SparkSession") {
     val session = SparkSession.builder()