Wraps exceptions into RuntimeException objects as a workaround for the

case of custom exceptions.

Author: Islam Hassan

job-server/src/spark.jobserver/JobManagerActor.scala

@@ -287,8 +287,10 @@ class JobManagerActor(contextConfig: Config) extends InstrumentedActor {
         statusActor ! JobFinished(jobId, DateTime.now())
         resultActor ! JobResult(jobId, result)
       case Failure(error: Throwable) =>
+        // Wrapping the error inside a RuntimeException to handle the case of throwing custom exceptions.
+        val wrappedError = wrapInRuntimeException(error)
         // If and only if job validation fails, JobErroredOut message is dropped silently in JobStatusActor.
-        statusActor ! JobErroredOut(jobId, DateTime.now(), error)
+        statusActor ! JobErroredOut(jobId, DateTime.now(), wrappedError)
         logger.warn("Exception from job " + jobId + ": ", error)
     }(executionContext).andThen {
       case _ =>
@@ -301,6 +303,28 @@ class JobManagerActor(contextConfig: Config) extends InstrumentedActor {
     }(executionContext)
   }
 
+  // Wraps a Throwable object into a RuntimeException. This is useful in case
+  // a custom exception is thrown. Currently, throwing a custom exception doesn't
+  // work and this is a workaround to wrap it into a standard exception.
+  protected def wrapInRuntimeException(t: Throwable): RuntimeException = {
+    val cause : Throwable = getRootCause(t)
+    val e : RuntimeException = new RuntimeException("%s: %s"
+      .format(cause.getClass().getName() ,cause.getMessage))
+    e.setStackTrace(cause.getStackTrace())
+    return e
+  }
+
+  // Gets the very first exception that caused the current exception to be thrown.
+  protected def getRootCause(t: Throwable): Throwable = {
+    var result : Throwable = t
+    var cause : Throwable = result.getCause()
+    while(cause != null  && (result != cause) ) {
+      result = cause
+      cause = result.getCause()
+    }
+    return result
+  }
+
   // Use our classloader and a factory to create the SparkContext.  This ensures the SparkContext will use
   // our class loader when it spins off threads, and ensures SparkContext can find the job and dependent jars
   // when doing serialization, for example.