Merge branch 'master' of github.com:apache/spark into concurrent-sql-…

…executions

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

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

@@ -162,10 +162,15 @@ class ThreadingSuite extends SparkFunSuite with LocalSparkContext with Logging {
     val threads = (1 to 5).map { i =>
       new Thread() {
         override def run() {
-          // TODO: these assertion failures don't actually fail the test...
-          sc.setLocalProperty("test", i.toString)
-          assert(sc.getLocalProperty("test") === i.toString)
-          sem.release()
+          try {
+            sc.setLocalProperty("test", i.toString)
+            assert(sc.getLocalProperty("test") === i.toString)
+          } catch {
+            case t: Throwable =>
+              throwable = Some(t)
+          } finally {
+            sem.release()
+          }
         }
       }
     }
@@ -185,11 +190,16 @@ class ThreadingSuite extends SparkFunSuite with LocalSparkContext with Logging {
     val threads = (1 to 5).map { i =>
       new Thread() {
         override def run() {
-          // TODO: these assertion failures don't actually fail the test...
-          assert(sc.getLocalProperty("test") === "parent")
-          sc.setLocalProperty("test", i.toString)
-          assert(sc.getLocalProperty("test") === i.toString)
-          sem.release()
+          try {
+            assert(sc.getLocalProperty("test") === "parent")
+            sc.setLocalProperty("test", i.toString)
+            assert(sc.getLocalProperty("test") === i.toString)
+          } catch {
+            case t: Throwable =>
+              throwable = Some(t)
+          } finally {
+            sem.release()
+          }
         }
       }
     }
@@ -246,7 +256,6 @@ class ThreadingSuite extends SparkFunSuite with LocalSparkContext with Logging {
     // Create a new thread which will inherit the current thread's properties
     val thread = new Thread() {
       override def run(): Unit = {
-        // TODO: these assertion failures don't actually fail the test...
         assert(sc.getLocalProperty(SparkContext.SPARK_JOB_GROUP_ID) === "originalJobGroupId")
         // Sleeps for a total of 10 seconds, but allows cancellation to interrupt the task
         try {