Address Prashant and Matei's comments

Author: aarondav

core/src/main/scala/org/apache/spark/api/python/PythonRDD.scala

@@ -281,7 +281,6 @@ private[spark] object PythonRDD {
       }
     } catch {
       case eof: EOFException => {}
-      case e: Exception => throw e
     }
     JavaRDD.fromRDD(sc.sc.parallelize(objs, parallelism))
   }

core/src/main/scala/org/apache/spark/executor/Executor.scala

@@ -241,26 +241,21 @@ private[spark] class Executor(
           // Attempt to exit cleanly by informing the driver of our failure.
           // If anything goes wrong (or this was a fatal exception), we will delegate to
           // the default uncaught exception handler, which will terminate the Executor.
-          try {
-            logError("Exception in task ID " + taskId, t)
-
-            val serviceTime = System.currentTimeMillis() - taskStart
-            val metrics = attemptedTask.flatMap(t => t.metrics)
-            for (m <- metrics) {
-              m.executorRunTime = serviceTime
-              m.jvmGCTime = gcTime - startGCTime
-            }
-            val reason = ExceptionFailure(t.getClass.getName, t.toString, t.getStackTrace, metrics)
-            execBackend.statusUpdate(taskId, TaskState.FAILED, ser.serialize(reason))
-
-            // Don't forcibly exit unless the exception was inherently fatal, to avoid
-            // stopping other tasks unnecessarily.
-            if (Utils.isFatalError(t)) {
-              ExecutorUncaughtExceptionHandler.uncaughtException(t)
-            }
-          } catch {
-            case t2: Throwable =>
-              ExecutorUncaughtExceptionHandler.uncaughtException(t2)
+          logError("Exception in task ID " + taskId, t)
+
+          val serviceTime = System.currentTimeMillis() - taskStart
+          val metrics = attemptedTask.flatMap(t => t.metrics)
+          for (m <- metrics) {
+            m.executorRunTime = serviceTime
+            m.jvmGCTime = gcTime - startGCTime
+          }
+          val reason = ExceptionFailure(t.getClass.getName, t.toString, t.getStackTrace, metrics)
+          execBackend.statusUpdate(taskId, TaskState.FAILED, ser.serialize(reason))
+
+          // Don't forcibly exit unless the exception was inherently fatal, to avoid
+          // stopping other tasks unnecessarily.
+          if (Utils.isFatalError(t)) {
+            ExecutorUncaughtExceptionHandler.uncaughtException(t)
           }
         }
       } finally {