[SPARK-29046][SQL] Fix NPE in SQLConf.get when active SparkContext is stopping

# What changes were proposed in this pull request?

This patch fixes the bug regarding NPE in SQLConf.get, which is only possible when SparkContext._dagScheduler is null due to stopping SparkContext. The logic doesn't seem to consider active SparkContext could be in progress of stopping.

Note that it can't be encountered easily as `SparkContext.stop()` blocks the main thread, but there're many cases which SQLConf.get is accessed concurrently while SparkContext.stop() is executing - users run another threads, or listener is accessing SQLConf.get after dagScheduler is set to null (this is the case what I encountered.)

### Why are the changes needed?

The bug brings NPE.

### Does this PR introduce any user-facing change?

No.

### How was this patch tested?

Added new UT to verify NPE doesn't occur. Without patch, the test fails with throwing NPE.

Closes #25753 from HeartSaVioR/SPARK-29046.

Authored-by: Jungtaek Lim (HeartSaVioR) <[email protected]>
Signed-off-by: HyukjinKwon <[email protected]>

Author: HeartSaVioR

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

@@ -139,7 +139,8 @@ object SQLConf {
       }
     } else {
       val isSchedulerEventLoopThread = SparkContext.getActive
-        .map(_.dagScheduler.eventProcessLoop.eventThread)
+        .flatMap { sc => Option(sc.dagScheduler) }
+        .map(_.eventProcessLoop.eventThread)
         .exists(_.getId == Thread.currentThread().getId)
       if (isSchedulerEventLoopThread) {
         // DAGScheduler event loop thread does not have an active SparkSession, the `confGetter`

sql/core/src/test/scala/org/apache/spark/sql/internal/SQLConfSuite.scala

@@ -19,6 +19,7 @@ package org.apache.spark.sql.internal
 
 import org.apache.hadoop.fs.Path
 
+import org.apache.spark.{LocalSparkContext, SparkConf, SparkContext}
 import org.apache.spark.sql._
 import org.apache.spark.sql.internal.StaticSQLConf._
 import org.apache.spark.sql.test.{SharedSparkSession, TestSQLContext}
@@ -320,4 +321,22 @@ class SQLConfSuite extends QueryTest with SharedSparkSession {
     assert(e2.getMessage.contains("spark.sql.shuffle.partitions"))
   }
 
+  test("SPARK-29046: SQLConf.get shouldn't throw NPE when active SparkContext is stopping") {
+    // Logically, there's only one case SQLConf.get throws NPE: there's active SparkContext,
+    // but SparkContext is stopping - especially it sets dagScheduler as null.
+
+    val oldSparkContext = SparkContext.getActive
+    Utils.tryWithSafeFinally {
+      // this is necessary to set new SparkContext as active: it cleans up active SparkContext
+      oldSparkContext.foreach(_ => SparkContext.clearActiveContext())
+
+      val conf = new SparkConf().setAppName("test").setMaster("local")
+      LocalSparkContext.withSpark(new SparkContext(conf)) { sc =>
+        sc.dagScheduler = null
+        SQLConf.get
+      }
+    } {
+      oldSparkContext.orElse(Some(null)).foreach(SparkContext.setActiveContext)
+    }
+  }
 }