Github user aarondav commented on a diff in the pull request:
https://github.com/apache/spark/pull/7014#discussion_r35280447
--- Diff: core/src/test/scala/org/apache/spark/FailureSuite.scala ---
@@ -141,5 +141,73 @@ class FailureSuite extends SparkFunSuite with
LocalSparkContext {
FailureSuiteState.clear()
}
+ // Run a 3-task map job in which task 1 always fails with a exception
message that
+ // depends on the failure number, and check that we get the last failure.
+ test("last failure cause is sent back to driver") {
+ sc = new SparkContext("local[1,2]", "test")
+ val data = sc.makeRDD(1 to 3, 3).map { x =>
+ FailureSuiteState.synchronized {
+ FailureSuiteState.tasksRun += 1
+ if (x == 3) {
+ FailureSuiteState.tasksFailed += 1
+ throw new UserException("oops",
+ new IllegalArgumentException("failed=" +
FailureSuiteState.tasksFailed))
+ }
+ }
+ x * x
+ }
+ val thrown = intercept[SparkException] {
+ data.collect()
+ }
+ FailureSuiteState.synchronized {
+ assert(FailureSuiteState.tasksRun === 4)
+ }
+ assert(thrown.getClass === classOf[SparkException])
+ assert(thrown.getCause.getClass === classOf[UserException])
+ assert(thrown.getCause.getMessage === "oops")
+ assert(thrown.getCause.getCause.getClass ===
classOf[IllegalArgumentException])
+ assert(thrown.getCause.getCause.getMessage === "failed=2")
+ FailureSuiteState.clear()
+ }
+
+ test("failure cause stacktrace is sent back to driver if exception is
not serializable") {
+ sc = new SparkContext("local", "test")
+ val data = sc.makeRDD(1 to 3).map(x => { throw new
NonSerializableUserException; (x,
+ x) }).groupByKey(3)
+ val thrown = intercept[SparkException] {
+ data.collect()
+ }
+ assert(thrown.getClass === classOf[SparkException])
+ assert(thrown.getCause === null)
+ assert(thrown.getMessage.contains("NonSerializableUserException"))
+ FailureSuiteState.clear()
+ }
+
+ test("failure cause stacktrace is sent back to driver if exception is
not deserializable") {
+ sc = new SparkContext("local", "test")
+ val data = sc.makeRDD(1 to 3).map(x => { throw new
NonDeserializableUserException; (x,
+ x) }).groupByKey(3)
+ val thrown = intercept[SparkException] {
+ data.collect()
+ }
+ assert(thrown.getClass === classOf[SparkException])
+ assert(thrown.getCause === null)
+ assert(thrown.getMessage.contains("NonDeserializableUserException"))
+ FailureSuiteState.clear()
+ }
+
// TODO: Need to add tests with shuffle fetch failures.
}
+
+class UserException(message: String, cause: Throwable)
+ extends RuntimeException(message, cause)
+
+class NonSerializableUserException extends RuntimeException {
+ val nonSerializableInstanceVariable = new NonSerializable
+}
+
+class NonDeserializableUserException extends RuntimeException {
+ private def readObject(in: ObjectInputStream): Unit = {
+ throw new IOException("Intentional exception during deserialization.");
--- End diff --
nit: semicolon
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
