Fixed the review comments

core/src/main/scala/org/apache/spark/deploy/master/Master.scala

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

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

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

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

@@ -22,27 +22,28 @@ import org.apache.spark.internal.Logging
 /**
  * The default uncaught exception handler for Spark daemons. It terminates the whole process for
  * any Errors, and also terminates the process for Exceptions when the exitOnException flag is true.
+ *
+ * @param exitOnUncaughtException Whether to exit the process on UncaughtException.
  */
-private[spark] class SparkUncaughtExceptionHandler(val exitOnException: Boolean = true)
+private[spark] class SparkUncaughtExceptionHandler(val exitOnUncaughtException: Boolean = true)
   extends Thread.UncaughtExceptionHandler with Logging {
 
   override def uncaughtException(thread: Thread, exception: Throwable) {
     try {
-      // Make it explicit that uncaught exceptions are thrown when process is shutting down.
+      // Make it explicit that uncaught exceptions are thrown when container 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)) {
-        logError(errMsg + ". Shutting down now..", exception)
+      val inShutdownMsg = if (ShutdownHookManager.inShutdown()) "[Container in shutdown] " else ""
+      val errMsg = "Uncaught exception in thread "
+      logError(inShutdownMsg + errMsg + thread, exception)
+
+      // 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()) {
         if (exception.isInstanceOf[OutOfMemoryError]) {
           System.exit(SparkExitCode.OOM)
-        } else {
+        } else if (exitOnUncaughtException) {
           System.exit(SparkExitCode.UNCAUGHT_EXCEPTION)
         }
-      } else {
-        logError(errMsg, exception)
       }
     } catch {
       case oom: OutOfMemoryError => Runtime.getRuntime.halt(SparkExitCode.OOM)