spark-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From andrewo...@apache.org
Subject spark git commit: [SPARK-10564] ThreadingSuite: assertion failures in threads don't fail the test (round 2)
Date Mon, 14 Sep 2015 22:09:54 GMT
Repository: spark
Updated Branches:
  refs/heads/branch-1.5 eb0cb25bb -> 5db51f911


[SPARK-10564] ThreadingSuite: assertion failures in threads don't fail the test (round 2)

This is a follow-up patch to #8723. I missed one case there.

Author: Andrew Or <andrew@databricks.com>

Closes #8727 from andrewor14/fix-threading-suite.

(cherry picked from commit 7b6c856367b9c36348e80e83959150da9656c4dd)
Signed-off-by: Andrew Or <andrew@databricks.com>


Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/5db51f91
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/5db51f91
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/5db51f91

Branch: refs/heads/branch-1.5
Commit: 5db51f91131e867fd27cb6b0457a2698925cd920
Parents: eb0cb25
Author: Andrew Or <andrew@databricks.com>
Authored: Mon Sep 14 15:09:43 2015 -0700
Committer: Andrew Or <andrew@databricks.com>
Committed: Mon Sep 14 15:09:50 2015 -0700

----------------------------------------------------------------------
 .../scala/org/apache/spark/ThreadingSuite.scala | 23 +++++++++++++-------
 1 file changed, 15 insertions(+), 8 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/spark/blob/5db51f91/core/src/test/scala/org/apache/spark/ThreadingSuite.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/org/apache/spark/ThreadingSuite.scala b/core/src/test/scala/org/apache/spark/ThreadingSuite.scala
index cda2b24..a96a4ce 100644
--- a/core/src/test/scala/org/apache/spark/ThreadingSuite.scala
+++ b/core/src/test/scala/org/apache/spark/ThreadingSuite.scala
@@ -147,12 +147,12 @@ class ThreadingSuite extends SparkFunSuite with LocalSparkContext with
Logging {
       }.start()
     }
     sem.acquire(2)
+    throwable.foreach { t => throw t }
     if (ThreadingSuiteState.failed.get()) {
       logError("Waited 1 second without seeing runningThreads = 4 (it was " +
                 ThreadingSuiteState.runningThreads.get() + "); failing test")
       fail("One or more threads didn't see runningThreads = 4")
     }
-    throwable.foreach { t => throw t }
   }
 
   test("set local properties in different thread") {
@@ -178,8 +178,8 @@ class ThreadingSuite extends SparkFunSuite with LocalSparkContext with
Logging {
     threads.foreach(_.start())
 
     sem.acquire(5)
-    assert(sc.getLocalProperty("test") === null)
     throwable.foreach { t => throw t }
+    assert(sc.getLocalProperty("test") === null)
   }
 
   test("set and get local properties in parent-children thread") {
@@ -207,15 +207,16 @@ class ThreadingSuite extends SparkFunSuite with LocalSparkContext with
Logging {
     threads.foreach(_.start())
 
     sem.acquire(5)
+    throwable.foreach { t => throw t }
     assert(sc.getLocalProperty("test") === "parent")
     assert(sc.getLocalProperty("Foo") === null)
-    throwable.foreach { t => throw t }
   }
 
   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
+    var throwable: Option[Throwable] = None
 
     sc = new SparkContext("local", "test")
     sc.setJobGroup("originalJobGroupId", "description")
@@ -232,14 +233,19 @@ 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 = {
-        assert(sc.getLocalProperty(SparkContext.SPARK_JOB_GROUP_ID) === "originalJobGroupId")
-        // Sleeps for a total of 10 seconds, but allows cancellation to interrupt the task
         try {
-          sc.parallelize(1 to 100).foreach { x =>
-            Thread.sleep(100)
+          assert(sc.getLocalProperty(SparkContext.SPARK_JOB_GROUP_ID) === "originalJobGroupId")
+          // Sleeps for a total of 10 seconds, but allows cancellation to interrupt the task
+          try {
+            sc.parallelize(1 to 100).foreach { x =>
+              Thread.sleep(100)
+            }
+          } catch {
+            case s: SparkException => // ignored so that we don't print noise in test
logs
           }
         } catch {
-          case s: SparkException => // ignored so that we don't print noise in test logs
+          case t: Throwable =>
+            throwable = Some(t)
         }
       }
     }
@@ -252,6 +258,7 @@ class ThreadingSuite extends SparkFunSuite with LocalSparkContext with
Logging {
     // modification of the properties object should not affect the properties of running
jobs
     sc.cancelJobGroup("originalJobGroupId")
     jobEnded.tryAcquire(10, TimeUnit.SECONDS)
+    throwable.foreach { t => throw t }
     assert(jobResult.isInstanceOf[JobFailed])
   }
 }


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org


Mime
View raw message