[SPARK-20204][SQL][Followup] SQLConf should react to change in default timezone settings

core/src/main/scala/org/apache/spark/internal/config/ConfigBuilder.scala

@@ -147,6 +147,14 @@ private[spark] class TypedConfigBuilder[T](
     }
   }
 
+  /** Creates a [[ConfigEntry]] with a function has a default value */
+  def createWithDefaultFunction(defaultFunc: () => T): ConfigEntry[T] = {
+    val entry = new ConfigEntryWithDefaultFunction[T](parent.key, defaultFunc, converter,
+      stringConverter, parent._doc, parent._public)
+    parent._onCreate.foreach(_ (entry))
+    entry
+  }
+
   /**
    * Creates a [[ConfigEntry]] that has a default value. The default value is provided as a
    * [[String]] and must be a valid value for the entry.

core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala

@@ -78,7 +78,24 @@ private class ConfigEntryWithDefault[T] (
   def readFrom(reader: ConfigReader): T = {
     reader.get(key).map(valueConverter).getOrElse(_defaultValue)
   }
+}
+
+private class ConfigEntryWithDefaultFunction[T] (
+     key: String,
+     _defaultFunction: () => T,
+     valueConverter: String => T,
+     stringConverter: T => String,
+     doc: String,
+     isPublic: Boolean)
+  extends ConfigEntry(key, valueConverter, stringConverter, doc, isPublic) {
+
+  override def defaultValue: Option[T] = Some(_defaultFunction())
 
+  override def defaultValueString: String = stringConverter(_defaultFunction())
+
+  def readFrom(reader: ConfigReader): T = {
+    reader.get(key).map(valueConverter).getOrElse(_defaultFunction())
+  }
 }
 
 private class ConfigEntryWithDefaultString[T] (

core/src/test/scala/org/apache/spark/internal/config/ConfigEntrySuite.scala

@@ -17,6 +17,7 @@
 
 package org.apache.spark.internal.config
 
+import java.util.TimeZone
 import java.util.concurrent.TimeUnit
 
 import org.apache.spark.{SparkConf, SparkFunSuite}
@@ -51,6 +52,26 @@ class ConfigEntrySuite extends SparkFunSuite {
     assert(conf.get(dConf) === 20.0)
   }
 
+  test("conf entry: timezone") {
+    val tzStart = TimeZone.getDefault().getID()
+    val conf = new SparkConf()
+    val dConf = ConfigBuilder(testKey("tz"))
+      .stringConf
+      .createWithDefaultFunction(() => TimeZone.getDefault().getID())
+
+    val tzConf = conf.get(dConf)
+    assert(tzStart === tzConf)
+
+    // Pick a timezone which is not the current timezone
+    val availableTzs: Seq[String] = TimeZone.getAvailableIDs();
+    val newTz = availableTzs.find(_ != tzStart).getOrElse(tzStart)
+    TimeZone.setDefault(TimeZone.getTimeZone(newTz))
+
+    val tzChanged = conf.get(dConf)
+    assert(tzChanged === newTz)
+    TimeZone.setDefault(TimeZone.getTimeZone(tzStart))
+  }
+
   test("conf entry: boolean") {
     val conf = new SparkConf()
     val bConf = ConfigBuilder(testKey("boolean")).booleanConf.createWithDefault(false)

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

@@ -752,7 +752,7 @@ object SQLConf {
     buildConf("spark.sql.session.timeZone")
       .doc("""The ID of session local timezone, e.g. "GMT", "America/Los_Angeles", etc.""")
       .stringConf
-      .createWithDefault(TimeZone.getDefault().getID())
+      .createWithDefaultFunction(() => TimeZone.getDefault().getID())
 
   val WINDOW_EXEC_BUFFER_SPILL_THRESHOLD =
     buildConf("spark.sql.windowExec.buffer.spill.threshold")