Now using regexes to simplify code in parseTimeString. Introduces get…

…TimeAsSec and getTimeAsMs methods in SparkConf. Updated documentation

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

@@ -80,16 +80,16 @@ private[spark] class ExecutorAllocationManager(
     Integer.MAX_VALUE)
 
   // How long there must be backlogged tasks for before an addition is triggered (seconds)
-  private val schedulerBacklogTimeoutS = Utils.timeStringAsSec(conf.get(
-    "spark.dynamicAllocation.schedulerBacklogTimeout", "5s"))
+  private val schedulerBacklogTimeoutS = conf.getTimeAsSec(
+    "spark.dynamicAllocation.schedulerBacklogTimeout", "5s")
 
   // Same as above, but used only after `schedulerBacklogTimeoutS` is exceeded
-  private val sustainedSchedulerBacklogTimeoutS = Utils.timeStringAsSec(conf.get(
-    "spark.dynamicAllocation.sustainedSchedulerBacklogTimeout", s"${schedulerBacklogTimeoutS}s"))
+  private val sustainedSchedulerBacklogTimeoutS = conf.getTimeAsSec(
+    "spark.dynamicAllocation.sustainedSchedulerBacklogTimeout", s"${schedulerBacklogTimeoutS}s")
 
   // How long an executor must be idle for before it is removed (seconds)
-  private val executorIdleTimeoutS = Utils.timeStringAsSec(conf.get(
-    "spark.dynamicAllocation.executorIdleTimeout", "600s"))
+  private val executorIdleTimeoutS = conf.getTimeAsSec(
+    "spark.dynamicAllocation.executorIdleTimeout", "600s")
 
   // During testing, the methods to actually kill and add executors are mocked out
   private val testing = conf.getBoolean("spark.dynamicAllocation.testing", false)

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

@@ -62,17 +62,17 @@ private[spark] class HeartbeatReceiver(sc: SparkContext)
 
   // "spark.network.timeout" uses "seconds", while `spark.storage.blockManagerSlaveTimeoutMs` uses
   // "milliseconds"
-  private val slaveTimeoutMs = Utils.timeStringAsMs(
-    sc.conf.get("spark.storage.blockManagerSlaveTimeoutMs", "120s"))
-  private val executorTimeoutMs = Utils.timeStringAsSec(sc.conf.get("spark.network.timeout",
-    s"${slaveTimeoutMs}ms")) * 1000
+  private val slaveTimeoutMs = 
+    sc.conf.getTimeAsMs("spark.storage.blockManagerSlaveTimeoutMs", "120s")
+  private val executorTimeoutMs = 
+    sc.conf.getTimeAsSec("spark.network.timeout", s"${slaveTimeoutMs}ms") * 1000
 
   // "spark.network.timeoutInterval" uses "seconds", while
   // "spark.storage.blockManagerTimeoutIntervalMs" uses "milliseconds"
-  private val timeoutIntervalMs = Utils.timeStringAsMs(
-    sc.conf.get("spark.storage.blockManagerTimeoutIntervalMs", "60s"))
-  private val checkTimeoutIntervalMs =  Utils.timeStringAsSec(
-    sc.conf.get("spark.network.timeoutInterval", s"${timeoutIntervalMs}ms")) * 1000
+  private val timeoutIntervalMs = 
+    sc.conf.getTimeAsMs("spark.storage.blockManagerTimeoutIntervalMs", "60s")
+  private val checkTimeoutIntervalMs = 
+    sc.conf.getTimeAsSec("spark.network.timeoutInterval", s"${timeoutIntervalMs}ms") * 1000
 
   private var timeoutCheckingTask: ScheduledFuture[_] = null
 

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

@@ -174,6 +174,35 @@ class SparkConf(loadDefaults: Boolean) extends Cloneable with Logging {
     getOption(key).getOrElse(defaultValue)
   }
 
+  /** Get a time parameter as seconds; throws a NoSuchElementException if it's not set. If no 
+    * suffix is provided then seconds are assumed. 
+    */
+  def getTimeAsSec(key: String): Long = {
+    Utils.timeStringAsSec(getOption(key).getOrElse(throw new NoSuchElementException(key)))
+  }
+
+  /** Get a time parameter as seconds, falling back to a default if not set. If no 
+    * suffix is provided then seconds are assumed. 
+    */
+  def getTimeAsSec(key: String, defaultValue: String): Long = {
+    Utils.timeStringAsSec(getOption(key).getOrElse(defaultValue))
+  }
+
+  /** Get a time parameter as milliseconds; throws a NoSuchElementException if it's not set. If no 
+    * suffix is provided then milliseconds are assumed. 
+    */
+  def getTimeAsMs(key: String): Long = {
+    Utils.timeStringAsMs(getOption(key).getOrElse(throw new NoSuchElementException(key)))
+  }
+
+  /** Get a time parameter as milliseconds, falling back to a default if not set. If no 
+    * suffix is provided then milliseconds are assumed. 
+    */
+  def getTimeAsMs(key: String, defaultValue: String): Long = {
+    Utils.timeStringAsMs(getOption(key).getOrElse(defaultValue))
+  }
+
+
   /** Get a parameter as an Option */
   def getOption(key: String): Option[String] = {
     Option(settings.get(key))

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

@@ -436,7 +436,7 @@ private[spark] class Executor(
    * This thread stops running when the executor is stopped.
    */
   private def startDriverHeartbeater(): Unit = {
-    val intervalMs = Utils.timeStringAsMs(conf.get("spark.executor.heartbeatInterval", "10s"))
+    val intervalMs = conf.getTimeAsMs("spark.executor.heartbeatInterval", "10s")
     val thread = new Thread() {
       override def run() {
         // Sleep a random interval so the heartbeats don't end up in sync

core/src/main/scala/org/apache/spark/network/nio/ConnectionManager.scala

@@ -82,8 +82,8 @@ private[nio] class ConnectionManager(
     new HashedWheelTimer(Utils.namedThreadFactory("AckTimeoutMonitor"))
 
   private val ackTimeout =
-    Utils.timeStringAsSec(conf.get("spark.core.connection.ack.wait.timeout",
-      conf.get("spark.network.timeout", "120s")))
+    conf.getTimeAsSec("spark.core.connection.ack.wait.timeout",
+      conf.get("spark.network.timeout", "120s"))
 
   // Get the thread counts from the Spark Configuration.
   // 

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

@@ -62,11 +62,10 @@ private[spark] class TaskSchedulerImpl(
   val conf = sc.conf
 
   // How often to check for speculative tasks
-  val SPECULATION_INTERVAL_MS =
-    Utils.timeStringAsMs(conf.get("spark.speculation.interval", "100ms"))
+  val SPECULATION_INTERVAL_MS = conf.getTimeAsMs("spark.speculation.interval", "100ms")
 
   // Threshold above which we warn user initial TaskSet may be starved
-  val STARVATION_TIMEOUT_MS = Utils.timeStringAsMs(conf.get("spark.starvation.timeout", "15s"))
+  val STARVATION_TIMEOUT_MS = conf.getTimeAsMs("spark.starvation.timeout", "15s")
 
   // CPUs to request per task
   val CPUS_PER_TASK = conf.getInt("spark.task.cpus", 1)

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

@@ -855,7 +855,7 @@ private[spark] class TaskSetManager(
       case TaskLocality.RACK_LOCAL => "spark.locality.wait.rack"
       case _ => ""
     }
-    Utils.timeStringAsMs(conf.get(localityWaitKey, defaultWait))
+    conf.getTimeAsMs(localityWaitKey, defaultWait)
   }
 
   /**

core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala

@@ -53,7 +53,7 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
   // Submit tasks after maxRegisteredWaitingTime milliseconds
   // if minRegisteredRatio has not yet been reached
   val maxRegisteredWaitingTimeMs =
-    Utils.timeStringAsMs(conf.get("spark.scheduler.maxRegisteredResourcesWaitingTime", "30s"))
+    conf.getTimeAsMs("spark.scheduler.maxRegisteredResourcesWaitingTime", "30s")
   val createTime = System.currentTimeMillis()
 
   private val executorDataMap = new HashMap[String, ExecutorData]
@@ -77,8 +77,7 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
 
     override def onStart() {
       // Periodically revive offers to allow delay scheduling to work
-      val reviveIntervalMs = Utils.timeStringAsMs(
-        conf.get("spark.scheduler.revive.interval", "1s"))
+      val reviveIntervalMs = conf.getTimeAsMs("spark.scheduler.revive.interval", "1s")
 
       reviveThread.scheduleAtFixedRate(new Runnable {
         override def run(): Unit = Utils.tryLogNonFatalError {

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

@@ -20,7 +20,6 @@ package org.apache.spark.util
 import scala.collection.JavaConversions.mapAsJavaMap
 import scala.concurrent.Await
 import scala.concurrent.duration.{Duration, FiniteDuration}
-import scala.util.Try
 
 import akka.actor.{ActorRef, ActorSystem, ExtendedActorSystem}
 import akka.pattern.ask
@@ -66,8 +65,8 @@ private[spark] object AkkaUtils extends Logging {
 
     val akkaThreads   = conf.getInt("spark.akka.threads", 4)
     val akkaBatchSize = conf.getInt("spark.akka.batchSize", 15)
-    val akkaTimeoutS = Utils.timeStringAsSec(conf.get("spark.akka.timeout",
-      conf.get("spark.network.timeout", "120s")))
+    val akkaTimeoutS = conf.getTimeAsSec("spark.akka.timeout",
+      conf.get("spark.network.timeout", "120s"))
     val akkaFrameSize = maxFrameSizeBytes(conf)
     val akkaLogLifecycleEvents = conf.getBoolean("spark.akka.logLifecycleEvents", false)
     val lifecycleEvents = if (akkaLogLifecycleEvents) "on" else "off"
@@ -79,10 +78,10 @@ private[spark] object AkkaUtils extends Logging {
 
     val logAkkaConfig = if (conf.getBoolean("spark.akka.logAkkaConfig", false)) "on" else "off"
 
-    val akkaHeartBeatPausesS = Utils.timeStringAsSec(conf.get("spark.akka.heartbeat.pauses", 
-      "6000s"))
+    val akkaHeartBeatPausesS = conf.getTimeAsSec("spark.akka.heartbeat.pauses", 
+      "6000s")
     val akkaHeartBeatIntervalS =
-      Utils.timeStringAsSec(conf.get("spark.akka.heartbeat.interval", "1000s"))
+      conf.getTimeAsSec("spark.akka.heartbeat.interval", "1000s")
 
     val secretKey = securityManager.getSecretKey()
     val isAuthOn = securityManager.isAuthenticationEnabled()

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

@@ -76,7 +76,7 @@ private[spark] object MetadataCleanerType extends Enumeration {
 // initialization of StreamingContext. It's okay for users trying to configure stuff themselves.
 private[spark] object MetadataCleaner {
   def getDelaySeconds(conf: SparkConf): Int = {
-    Utils.timeStringAsSec(conf.get("spark.cleaner.ttl", "-1")).toInt
+    conf.getTimeAsSec("spark.cleaner.ttl", "-1").toInt
   }
 
   def getDelaySeconds(

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

@@ -613,8 +613,8 @@ private[spark] object Utils extends Logging {
         }
         Utils.setupSecureURLConnection(uc, securityMgr)
 
-        val timeoutMs = Utils.timeStringAsSec(
-          conf.get("spark.files.fetchTimeout", "60s")).toInt * 1000
+        val timeoutMs = 
+          conf.getTimeAsSec("spark.files.fetchTimeout", "60s").toInt * 1000
         uc.setConnectTimeout(timeoutMs)
         uc.setReadTimeout(timeoutMs)
         uc.connect()

core/src/test/scala/org/apache/spark/scheduler/TaskSetManagerSuite.scala

@@ -151,7 +151,7 @@ class TaskSetManagerSuite extends FunSuite with LocalSparkContext with Logging {
 
   private val conf = new SparkConf
 
-  val LOCALITY_WAIT_MS = Utils.timeStringAsMs(conf.get("spark.locality.wait", "3s"))
+  val LOCALITY_WAIT_MS = conf.getTimeAsMs("spark.locality.wait", "3s")
   val MAX_TASK_FAILURES = 4
 
   override def beforeEach() {

core/src/test/scala/org/apache/spark/util/UtilsSuite.scala

@@ -17,6 +17,7 @@
 
 package org.apache.spark.util
 
+import java.lang.NumberFormatException
 import java.util.concurrent.TimeUnit
 
 import scala.util.Random
@@ -62,6 +63,30 @@ class UtilsSuite extends FunSuite with ResetSystemProperties {
     assert(Utils.timeStringAsUs("1min") === TimeUnit.MINUTES.toMicros(1))
     assert(Utils.timeStringAsUs("1h") === TimeUnit.HOURS.toMicros(1))
     assert(Utils.timeStringAsUs("1d") === TimeUnit.DAYS.toMicros(1))
+
+    // Test invalid strings
+    try {
+      Utils.timeStringAsMs("This breaks 600s")
+      assert(false) // We should never reach this
+    } catch {
+      case e: NumberFormatException => assert(true)
+    }
+
+    // Test invalid strings
+    try {
+      Utils.timeStringAsMs("600s This breaks")
+      assert(false) // We should never reach this
+    } catch {
+      case e: NumberFormatException => assert(true)
+    }
+
+    // Test invalid strings
+    try {
+      Utils.timeStringAsMs("This 123s breaks")
+      assert(false) // We should never reach this
+    } catch {
+      case e: NumberFormatException => assert(true)
+    }
   }
 
   test("bytesToString") {