Updated testsuites.

Author: tdas

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

@@ -43,7 +43,7 @@ class Checkpoint(@transient ssc: StreamingContext, val checkpointTime: Time)
   val delaySeconds = MetadataCleaner.getDelaySeconds(ssc.conf)
   val sparkConfPairs = ssc.conf.getAll
 
-  def sparkConf: SparkConf = {
+  def createSparkConf(): SparkConf = {
     val newSparkConf = new SparkConf(loadDefaults = false).setAll(sparkConfPairs)
       .remove("spark.driver.host")
       .remove("spark.driver.port")

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

@@ -116,7 +116,7 @@ class StreamingContext private[streaming] (
 
   private[streaming] val sc: SparkContext = {
     if (isCheckpointPresent) {
-      new SparkContext(cp_.sparkConf)
+      new SparkContext(cp_.createSparkConf())
     } else {
       sc_
     }

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

@@ -146,15 +146,15 @@ class CheckpointSuite extends TestSuiteBase {
 
   // This tests whether spark conf persists through checkpoints, and certain
   // configs gets scrubbed
-  test("persistence of conf through checkpoints") {
+  test("recovery of conf through checkpoints") {
     val key = "spark.mykey"
     val value = "myvalue"
     System.setProperty(key, value)
     ssc = new StreamingContext(master, framework, batchDuration)
     val originalConf = ssc.conf
 
     val cp = new Checkpoint(ssc, Time(1000))
-    val cpConf = cp.sparkConf
+    val cpConf = cp.createSparkConf()
     assert(cpConf.get("spark.master") === originalConf.get("spark.master"))
     assert(cpConf.get("spark.app.name") === originalConf.get("spark.app.name"))
     assert(cpConf.get(key) === value)
@@ -163,7 +163,8 @@ class CheckpointSuite extends TestSuiteBase {
     // Serialize/deserialize to simulate write to storage and reading it back
     val newCp = Utils.deserialize[Checkpoint](Utils.serialize(cp))
 
-    val newCpConf = newCp.sparkConf
+    // Verify new SparkConf has all the previous properties
+    val newCpConf = newCp.createSparkConf()
     assert(newCpConf.get("spark.master") === originalConf.get("spark.master"))
     assert(newCpConf.get("spark.app.name") === originalConf.get("spark.app.name"))
     assert(newCpConf.get(key) === value)
@@ -174,6 +175,20 @@ class CheckpointSuite extends TestSuiteBase {
     ssc = new StreamingContext(null, newCp, null)
     val restoredConf = ssc.conf
     assert(restoredConf.get(key) === value)
+    ssc.stop()
+
+    // Verify new SparkConf picks up new master url if it is set in the properties. See SPARK-6331.
+    try {
+      val newMaster = "local[100]"
+      System.setProperty("spark.master", newMaster)
+      val newCpConf = newCp.createSparkConf()
+      assert(newCpConf.get("spark.master") === newMaster)
+      assert(newCpConf.get("spark.app.name") === originalConf.get("spark.app.name"))
+      ssc = new StreamingContext(null, newCp, null)
+      assert(ssc.sparkContext.master === newMaster)
+    } finally {
+      System.clearProperty("spark.master")
+    }
   }
 
 

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

@@ -91,36 +91,18 @@ class StreamingContextSuite extends FunSuite with BeforeAndAfter with Timeouts w
   }
 
   test("from checkpoint") {
-    // Create Checkpoint and verify whether SparkConf configuration is captured
-    val myConf = SparkContext.updatedConf(new SparkConf(loadDefaults = false), master, appName)
+    val myConf = SparkContext.updatedConf(new SparkConf(false), master, appName)
     myConf.set("spark.cleaner.ttl", "10")
-    ssc = new StreamingContext(myConf, batchDuration)
-    addInputStream(ssc).register()
-    ssc.start()
-    val cp = new Checkpoint(ssc, Time(1000))
-    ssc.stop()
+    val ssc1 = new StreamingContext(myConf, batchDuration)
+    addInputStream(ssc1).register()
+    ssc1.start()
+    val cp = new Checkpoint(ssc1, Time(1000))
     assert(cp.sparkConfPairs.toMap.getOrElse("spark.cleaner.ttl", "-1") === "10")
-
-    // Verify SparkConf recreated from Checkpoint has the same configuration
+    ssc1.stop()
     val newCp = Utils.deserialize[Checkpoint](Utils.serialize(cp))
-    assert(newCp.sparkConf.getInt("spark.cleaner.ttl", -1) === 10)
+    assert(newCp.createSparkConf().getInt("spark.cleaner.ttl", -1) === 10)
     ssc = new StreamingContext(null, newCp, null)
     assert(ssc.conf.getInt("spark.cleaner.ttl", -1) === 10)
-    assert(ssc.conf.get("spark.master") === master)
-    ssc.stop()
-
-    // Verify SparkConf recreated from Checkpoint picks up new master
-    try {
-      val newMaster = "local[100]"
-      System.setProperty("spark.master", newMaster)
-      val anotherNewCp = Utils.deserialize[Checkpoint](Utils.serialize(cp))
-      ssc = new StreamingContext(null, anotherNewCp, null)
-      assert(ssc.conf.getInt("spark.cleaner.ttl", -1) === 10)
-      assert(ssc.conf.get("spark.master") === newMaster)
-      ssc.stop()
-    } finally {
-      System.clearProperty("spark.master")
-    }
   }
 
   test("start and stop state check") {