Address the comments

Author: jerryshao

core/src/main/scala/org/apache/spark/SparkContext.scala

@@ -94,11 +94,6 @@ class SparkContext(config: SparkConf) extends Logging with ExecutorAllocationCli
   // contains a map from hostname to a list of input format splits on the host.
   private[spark] var preferredNodeLocationData: Map[String, Set[SplitInfo]] = Map()
 
-  // This is used for Spark Streaming to check whether driver host and port are set by user,
-  // if these two configurations are set by user, so the recovery mechanism should not remove this.
-  private[spark] val isDriverHostSetByUser = config.contains("spark.driver.host")
-  private[spark] val isDriverPortSetByUser = config.contains("spark.driver.port")
-
   val startTime = System.currentTimeMillis()
 
   private val stopped: AtomicBoolean = new AtomicBoolean(false)

streaming/src/main/scala/org/apache/spark/streaming/Checkpoint.scala

@@ -41,9 +41,7 @@ class Checkpoint(@transient ssc: StreamingContext, val checkpointTime: Time)
   val checkpointDuration = ssc.checkpointDuration
   val pendingTimes = ssc.scheduler.getPendingTimes().toArray
   val delaySeconds = MetadataCleaner.getDelaySeconds(ssc.conf)
-  val sparkConfPairs = ssc.conf.getAll.filterNot { kv =>
-    (!ssc.sc.isDriverHostSetByUser && kv._1 == "spark.driver.host") ||
-      (!ssc.sc.isDriverPortSetByUser && kv._1 == "spark.driver.port") }
+  val sparkConfPairs = ssc.conf.getAll
 
   def createSparkConf(): SparkConf = {
 

streaming/src/test/scala/org/apache/spark/streaming/CheckpointSuite.scala

@@ -198,8 +198,6 @@ class CheckpointSuite extends TestSuiteBase {
     conf.foreach(kv => System.setProperty(kv._1, kv._2))
     ssc = new StreamingContext(master, framework, batchDuration)
     val originalConf = ssc.conf
-    assert(ssc.sc.isDriverHostSetByUser === true)
-    assert(ssc.sc.isDriverPortSetByUser === true)
     assert(originalConf.get("spark.driver.host") === "localhost")
     assert(originalConf.get("spark.driver.port") === "9999")
 
@@ -218,8 +216,6 @@ class CheckpointSuite extends TestSuiteBase {
     // Check if all the parameters have been restored
     ssc = new StreamingContext(null, newCp, null)
     val restoredConf = ssc.conf
-    assert(ssc.sc.isDriverHostSetByUser === true)
-    assert(ssc.sc.isDriverPortSetByUser === true)
     assert(restoredConf.get("spark.driver.host") === "localhost")
     assert(restoredConf.get("spark.driver.port") === "9999")
   }