Added a test for SPARK-571

Author: willb

core/src/test/scala/org/apache/spark/util/ClosureCleanerSuite.scala

@@ -21,6 +21,7 @@ import org.scalatest.FunSuite
 
 import org.apache.spark.LocalSparkContext._
 import org.apache.spark.SparkContext
+import org.apache.spark.SparkException
 
 class ClosureCleanerSuite extends FunSuite {
   test("closures inside an object") {
@@ -50,6 +51,14 @@ class ClosureCleanerSuite extends FunSuite {
     val obj = new TestClassWithNesting(1)
     assert(obj.run() === 96) // 4 * (1+2+3+4) + 4 * (1+2+3+4) + 16 * 1
   }
+  
+  test("return statements in closures are identified at cleaning time") {
+    val ex = intercept[SparkException] {
+      TestObjectWithBogusReturns.run()
+    }
+    
+    assert(ex.getMessage.contains("Return statements aren't allowed in Spark closures"))
+  }
 }
 
 // A non-serializable class we create in closures to make sure that we aren't
@@ -108,6 +117,20 @@ class TestClassWithoutFieldAccess {
   }
 }
 
+object TestObjectWithBogusReturns {
+  def badClosureWithReturn(v: org.apache.spark.rdd.RDD[Int]): Int = {
+     v.map {x => return 1 ; x * 2}
+     1
+  }
+  
+  def run(): Int = {
+    withSpark(new SparkContext("local", "test")) { sc =>
+      val nums = sc.parallelize(Array(1, 2, 3, 4))
+      badClosureWithReturn(nums)
+    }
+  }
+}
+
 
 object TestObjectWithNesting {
   def run(): Int = {