Updated SparkUncaughtExceptionHandler to take an argument whether to

kill/not-to-kill the process for the unhandled exceptions with the default
value as true, behaviour remains same for unhandled Error instances.

Author: Devaraj K

core/src/main/scala/org/apache/spark/deploy/worker/Worker.scala

@@ -737,7 +737,7 @@ private[deploy] object Worker extends Logging {
   val ENDPOINT_NAME = "Worker"
 
   def main(argStrings: Array[String]) {
-    Thread.setDefaultUncaughtExceptionHandler(SparkUncaughtExceptionHandler)
+    Thread.setDefaultUncaughtExceptionHandler(SparkUncaughtExceptionHandler(false))
     Utils.initDaemon(log)
     val conf = new SparkConf
     val args = new WorkerArguments(argStrings, conf)

core/src/main/scala/org/apache/spark/util/SparkUncaughtExceptionHandler.scala

@@ -26,27 +26,34 @@ import org.apache.spark.internal.Logging
  */
 private[spark] object SparkUncaughtExceptionHandler
   extends Thread.UncaughtExceptionHandler with Logging {
+  private[this] var exitOnException = true
 
-  override def uncaughtException(thread: Thread, exception: Throwable) {
-    try {
-      // Make it explicit that uncaught exceptions are thrown when container is shutting down.
-      // It will help users when they analyze the executor logs
-      val inShutdownMsg = if (ShutdownHookManager.inShutdown()) "[Container in shutdown] " else ""
-      val errMsg = "Uncaught exception in thread "
-      logError(inShutdownMsg + errMsg + thread, exception)
+  def apply(exitOnException: Boolean): Thread.UncaughtExceptionHandler = {
+    this.exitOnException = exitOnException
+    this
+  }
 
-      // We may have been called from a shutdown hook. If so, we must not call System.exit().
-      // (If we do, we will deadlock.)
-      if (!ShutdownHookManager.inShutdown()) {
+  override def uncaughtException(thread: Thread, exception: Throwable) {
+    // Make it explicit that uncaught exceptions are thrown when process is shutting down.
+    // It will help users when they analyze the executor logs
+    val errMsg = "Uncaught exception in thread " + thread
+    if (ShutdownHookManager.inShutdown()) {
+      logError("[Process in shutdown] " + errMsg, exception)
+    } else if (exception.isInstanceOf[Error] ||
+      (!exception.isInstanceOf[Error] && exitOnException)) {
+      try {
+        logError(errMsg + ". Shutting down now..", exception)
         if (exception.isInstanceOf[OutOfMemoryError]) {
           System.exit(SparkExitCode.OOM)
         } else {
           System.exit(SparkExitCode.UNCAUGHT_EXCEPTION)
         }
+      } catch {
+        case oom: OutOfMemoryError => Runtime.getRuntime.halt(SparkExitCode.OOM)
+        case t: Throwable => Runtime.getRuntime.halt(SparkExitCode.UNCAUGHT_EXCEPTION_TWICE)
       }
-    } catch {
-      case oom: OutOfMemoryError => Runtime.getRuntime.halt(SparkExitCode.OOM)
-      case t: Throwable => Runtime.getRuntime.halt(SparkExitCode.UNCAUGHT_EXCEPTION_TWICE)
+    } else {
+      logError(errMsg, exception)
     }
   }