Add JIRA reference; move clone into DAGScheduler

Author: JoshRosen

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

@@ -34,8 +34,6 @@ import scala.reflect.{ClassTag, classTag}
 
 import akka.actor.Props
 
-import org.apache.commons.lang3.SerializationUtils
-
 import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.fs.Path
 import org.apache.hadoop.io.{ArrayWritable, BooleanWritable, BytesWritable, DoubleWritable,
@@ -1491,7 +1489,7 @@ class SparkContext(config: SparkConf) extends Logging with ExecutorAllocationCli
       logInfo("RDD's recursive dependencies:\n" + rdd.toDebugString)
     }
     dagScheduler.runJob(rdd, cleanedFunc, partitions, callSite, allowLocal,
-      resultHandler, SerializationUtils.clone(localProperties.get))
+      resultHandler, localProperties.get)
     progressBar.foreach(_.finishAll())
     rdd.doCheckpoint()
   }
@@ -1577,7 +1575,7 @@ class SparkContext(config: SparkConf) extends Logging with ExecutorAllocationCli
     logInfo("Starting job: " + callSite.shortForm)
     val start = System.nanoTime
     val result = dagScheduler.runApproximateJob(rdd, func, evaluator, callSite, timeout,
-      SerializationUtils.clone(localProperties.get))
+      localProperties.get)
     logInfo(
       "Job finished: " + callSite.shortForm + ", took " + (System.nanoTime - start) / 1e9 + " s")
     result
@@ -1605,7 +1603,7 @@ class SparkContext(config: SparkConf) extends Logging with ExecutorAllocationCli
       callSite,
       allowLocal = false,
       resultHandler,
-      SerializationUtils.clone(localProperties.get))
+      localProperties.get)
     new SimpleFutureAction(waiter, resultFunc)
   }
 

core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala

@@ -31,6 +31,8 @@ import scala.util.control.NonFatal
 import akka.pattern.ask
 import akka.util.Timeout
 
+import org.apache.commons.lang3.SerializationUtils
+
 import org.apache.spark._
 import org.apache.spark.broadcast.Broadcast
 import org.apache.spark.executor.TaskMetrics
@@ -493,7 +495,8 @@ class DAGScheduler(
     val func2 = func.asInstanceOf[(TaskContext, Iterator[_]) => _]
     val waiter = new JobWaiter(this, jobId, partitions.size, resultHandler)
     eventProcessLoop.post(JobSubmitted(
-      jobId, rdd, func2, partitions.toArray, allowLocal, callSite, waiter, properties))
+      jobId, rdd, func2, partitions.toArray, allowLocal, callSite, waiter,
+      SerializationUtils.clone(properties)))
     waiter
   }
 
@@ -534,7 +537,8 @@ class DAGScheduler(
     val partitions = (0 until rdd.partitions.size).toArray
     val jobId = nextJobId.getAndIncrement()
     eventProcessLoop.post(JobSubmitted(
-      jobId, rdd, func2, partitions, allowLocal = false, callSite, listener, properties))
+      jobId, rdd, func2, partitions, allowLocal = false, callSite, listener,
+      SerializationUtils.clone(properties)))
     listener.awaitResult()    // Will throw an exception if the job fails
   }
 

core/src/test/scala/org/apache/spark/JobCancellationSuite.scala

@@ -141,7 +141,7 @@ class JobCancellationSuite extends FunSuite with Matchers with BeforeAndAfter
     assert(jobB.get() === 100)
   }
 
-  test("inherited job group") {
+  test("inherited job group (SPARK-6629)") {
     sc = new SparkContext("local[2]", "test")
 
     // Add a listener to release the semaphore once any tasks are launched.

core/src/test/scala/org/apache/spark/ThreadingSuite.scala

@@ -191,7 +191,7 @@ class ThreadingSuite extends FunSuite with LocalSparkContext {
     assert(sc.getLocalProperty("Foo") === null)
   }
 
-  test("mutations to local properties should not affect submitted jobs") {
+  test("mutations to local properties should not affect submitted jobs (SPARK-6629)") {
     val jobStarted = new Semaphore(0)
     val jobEnded = new Semaphore(0)
     @volatile var jobResult: JobResult = null