Merge branch 'master' of https://github.com/apache/spark into SPARK-1712_new

Author: witgo

.rat-excludes

@@ -43,3 +43,8 @@ test.out/*
 .*iml
 service.properties
 db.lck
+build/*
+dist/*
+.*out
+.*ipr
+.*iws

core/src/main/scala/org/apache/spark/SparkContext.scala

@@ -66,6 +66,12 @@ class SparkContext(config: SparkConf) extends Logging {
   // contains a map from hostname to a list of input format splits on the host.
   private[spark] var preferredNodeLocationData: Map[String, Set[SplitInfo]] = Map()
 
+  /**
+   * Create a SparkContext that loads settings from system properties (for instance, when
+   * launching with ./bin/spark-submit).
+   */
+  def this() = this(new SparkConf())
+
   /**
    * :: DeveloperApi ::
    * Alternative constructor for setting preferred locations where Spark will create executors.

core/src/main/scala/org/apache/spark/SparkEnv.scala

@@ -278,10 +278,11 @@ object SparkEnv extends Logging {
       addedJars: Seq[String],
       addedFiles: Seq[String]): Map[String, Seq[(String, String)]] = {
 
+    import Properties._
     val jvmInformation = Seq(
-      ("Java Version", "%s (%s)".format(Properties.javaVersion, Properties.javaVendor)),
-      ("Java Home", Properties.javaHome),
-      ("Scala Version", Properties.versionString)
+      ("Java Version", s"$javaVersion ($javaVendor)"),
+      ("Java Home", javaHome),
+      ("Scala Version", versionString)
     ).sorted
 
     // Spark properties
@@ -296,18 +297,15 @@ object SparkEnv extends Logging {
 
     // System properties that are not java classpaths
     val systemProperties = System.getProperties.iterator.toSeq
-    val otherProperties = systemProperties.filter { case (k, v) =>
+    val otherProperties = systemProperties.filter { case (k, _) =>
       k != "java.class.path" && !k.startsWith("spark.")
     }.sorted
 
     // Class paths including all added jars and files
-    val classPathProperty = systemProperties.find { case (k, v) =>
-      k == "java.class.path"
-    }.getOrElse(("", ""))
-    val classPathEntries = classPathProperty._2
+    val classPathEntries = javaClassPath
       .split(File.pathSeparator)
-      .filterNot(e => e.isEmpty)
-      .map(e => (e, "System Classpath"))
+      .filterNot(_.isEmpty)
+      .map((_, "System Classpath"))
     val addedJarsAndFiles = (addedJars ++ addedFiles).map((_, "Added By User"))
     val classPaths = (addedJarsAndFiles ++ classPathEntries).sorted
 

core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala

@@ -68,7 +68,7 @@ object SparkSubmit {
 
   /**
    * @return a tuple containing the arguments for the child, a list of classpath
-   *         entries for the child, a list of system propertes, a list of env vars
+   *         entries for the child, a list of system properties, a list of env vars
    *         and the main class for the child
    */
   private[spark] def createLaunchEnv(args: SparkSubmitArguments): (ArrayBuffer[String],

core/src/main/scala/org/apache/spark/deploy/client/AppClient.scala

@@ -30,7 +30,7 @@ import org.apache.spark.{Logging, SparkConf, SparkException}
 import org.apache.spark.deploy.{ApplicationDescription, ExecutorState}
 import org.apache.spark.deploy.DeployMessages._
 import org.apache.spark.deploy.master.Master
-import org.apache.spark.util.AkkaUtils
+import org.apache.spark.util.{Utils, AkkaUtils}
 
 /**
  * Interface allowing applications to speak with a Spark deploy cluster. Takes a master URL,
@@ -88,13 +88,15 @@ private[spark] class AppClient(
       var retries = 0
       registrationRetryTimer = Some {
         context.system.scheduler.schedule(REGISTRATION_TIMEOUT, REGISTRATION_TIMEOUT) {
-          retries += 1
-          if (registered) {
-            registrationRetryTimer.foreach(_.cancel())
-          } else if (retries >= REGISTRATION_RETRIES) {
-            markDead("All masters are unresponsive! Giving up.")
-          } else {
-            tryRegisterAllMasters()
+          Utils.tryOrExit {
+            retries += 1
+            if (registered) {
+              registrationRetryTimer.foreach(_.cancel())
+            } else if (retries >= REGISTRATION_RETRIES) {
+              markDead("All masters are unresponsive! Giving up.")
+            } else {
+              tryRegisterAllMasters()
+            }
           }
         }
       }

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

@@ -166,14 +166,16 @@ private[spark] class Worker(
     var retries = 0
     registrationRetryTimer = Some {
       context.system.scheduler.schedule(REGISTRATION_TIMEOUT, REGISTRATION_TIMEOUT) {
-        retries += 1
-        if (registered) {
-          registrationRetryTimer.foreach(_.cancel())
-        } else if (retries >= REGISTRATION_RETRIES) {
-          logError("All masters are unresponsive! Giving up.")
-          System.exit(1)
-        } else {
-          tryRegisterAllMasters()
+        Utils.tryOrExit {
+          retries += 1
+          if (registered) {
+            registrationRetryTimer.foreach(_.cancel())
+          } else if (retries >= REGISTRATION_RETRIES) {
+            logError("All masters are unresponsive! Giving up.")
+            System.exit(1)
+          } else {
+            tryRegisterAllMasters()
+          }
         }
       }
     }

core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala

@@ -31,6 +31,7 @@ import scala.util.Random
 import org.apache.spark._
 import org.apache.spark.TaskState.TaskState
 import org.apache.spark.scheduler.SchedulingMode.SchedulingMode
+import org.apache.spark.util.Utils
 
 /**
  * Schedules tasks for multiple types of clusters by acting through a SchedulerBackend.
@@ -139,7 +140,7 @@ private[spark] class TaskSchedulerImpl(
       import sc.env.actorSystem.dispatcher
       sc.env.actorSystem.scheduler.schedule(SPECULATION_INTERVAL milliseconds,
             SPECULATION_INTERVAL milliseconds) {
-        checkSpeculatableTasks()
+        Utils.tryOrExit { checkSpeculatableTasks() }
       }
     }
   }

core/src/main/scala/org/apache/spark/storage/BlockManager.scala

@@ -155,7 +155,7 @@ private[spark] class BlockManager(
     BlockManagerWorker.startBlockManagerWorker(this)
     if (!BlockManager.getDisableHeartBeatsForTesting(conf)) {
       heartBeatTask = actorSystem.scheduler.schedule(0.seconds, heartBeatFrequency.milliseconds) {
-        heartBeat()
+        Utils.tryOrExit { heartBeat() }
       }
     }
   }

core/src/main/scala/org/apache/spark/ui/UIUtils.scala

@@ -36,7 +36,13 @@ private[spark] object UIUtils extends Logging {
   def formatDate(timestamp: Long): String = dateFormat.get.format(new Date(timestamp))
 
   def formatDuration(milliseconds: Long): String = {
+    if (milliseconds < 100) {
+      return "%d ms".format(milliseconds)
+    }
     val seconds = milliseconds.toDouble / 1000
+    if (seconds < 1) {
+      return "%.1f s".format(seconds)
+    }
     if (seconds < 60) {
       return "%.0f s".format(seconds)
     }

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

@@ -40,6 +40,7 @@ import tachyon.client.{TachyonFile,TachyonFS}
 
 import org.apache.spark.{Logging, SecurityManager, SparkConf, SparkException}
 import org.apache.spark.deploy.SparkHadoopUtil
+import org.apache.spark.executor.ExecutorUncaughtExceptionHandler
 import org.apache.spark.serializer.{DeserializationStream, SerializationStream, SerializerInstance}
 
 /**
@@ -780,6 +781,18 @@ private[spark] object Utils extends Logging {
     output.toString
   }
 
+  /**
+   * Execute a block of code that evaluates to Unit, forwarding any uncaught exceptions to the
+   * default UncaughtExceptionHandler
+   */
+  def tryOrExit(block: => Unit) {
+    try {
+      block
+    } catch {
+      case t: Throwable => ExecutorUncaughtExceptionHandler.uncaughtException(t)
+    }
+  }
+
   /**
    * A regular expression to match classes of the "core" Spark API that we want to skip when
    * finding the call site of a method.