Made the most recent round of changes

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

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

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

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

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

@@ -174,30 +174,37 @@ 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(get(key))
+  /** 
+   * Get a time parameter as seconds; throws a NoSuchElementException if it's not set. If no 
+   * suffix is provided then seconds are assumed.
+   * @throws NoSuchElementException
+   */
+  def getTimeAsSeconds(key: String): Long = {
+    Utils.timeStringAsSeconds(get(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(get(key, defaultValue))
+  /** 
+   * Get a time parameter as seconds, falling back to a default if not set. If no 
+   * suffix is provided then seconds are assumed.
+   * 
+   */
+  def getTimeAsSeconds(key: String, defaultValue: String): Long = {
+    Utils.timeStringAsSeconds(get(key, defaultValue))
   }
 
-  /** Get a time parameter as milliseconds; throws a NoSuchElementException if it's not set. If no 
-    * suffix is provided then milliseconds are assumed. 
-    */
+  /** 
+   * Get a time parameter as milliseconds; throws a NoSuchElementException if it's not set. If no 
+   * suffix is provided then milliseconds are assumed. 
+   * @throws NoSuchElementException
+   */
   def getTimeAsMs(key: String): Long = {
     Utils.timeStringAsMs(get(key))
   }
 
-  /** Get a time parameter as milliseconds, falling back to a default if not set. If no 
-    * suffix is provided then milliseconds are assumed. 
-    */
+  /** 
+   * 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(get(key, defaultValue))
   }

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

@@ -82,7 +82,7 @@ private[nio] class ConnectionManager(
     new HashedWheelTimer(Utils.namedThreadFactory("AckTimeoutMonitor"))
 
   private val ackTimeout =
-    conf.getTimeAsSec("spark.core.connection.ack.wait.timeout",
+    conf.getTimeAsSeconds("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/TaskSetManager.scala

@@ -853,9 +853,14 @@ private[spark] class TaskSetManager(
       case TaskLocality.PROCESS_LOCAL => "spark.locality.wait.process"
       case TaskLocality.NODE_LOCAL => "spark.locality.wait.node"
       case TaskLocality.RACK_LOCAL => "spark.locality.wait.rack"
-      case _ => ""
+      case _ => null
+    }
+
+    if (localityWaitKey != null) {
+      conf.getTimeAsMs(localityWaitKey, defaultWait)  
+    } else {
+      0L
     }
-    conf.getTimeAsMs(localityWaitKey, defaultWait)
   }
 
   /**

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

@@ -65,7 +65,7 @@ private[spark] object AkkaUtils extends Logging {
 
     val akkaThreads   = conf.getInt("spark.akka.threads", 4)
     val akkaBatchSize = conf.getInt("spark.akka.batchSize", 15)
-    val akkaTimeoutS = conf.getTimeAsSec("spark.akka.timeout",
+    val akkaTimeoutS = conf.getTimeAsSeconds("spark.akka.timeout",
       conf.get("spark.network.timeout", "120s"))
     val akkaFrameSize = maxFrameSizeBytes(conf)
     val akkaLogLifecycleEvents = conf.getBoolean("spark.akka.logLifecycleEvents", false)
@@ -78,10 +78,8 @@ private[spark] object AkkaUtils extends Logging {
 
     val logAkkaConfig = if (conf.getBoolean("spark.akka.logAkkaConfig", false)) "on" else "off"
 
-    val akkaHeartBeatPausesS = conf.getTimeAsSec("spark.akka.heartbeat.pauses", 
-      "6000s")
-    val akkaHeartBeatIntervalS =
-      conf.getTimeAsSec("spark.akka.heartbeat.interval", "1000s")
+    val akkaHeartBeatPausesS = conf.getTimeAsSeconds("spark.akka.heartbeat.pauses", "6000s")
+    val akkaHeartBeatIntervalS = conf.getTimeAsSeconds("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 = {
-    conf.getTimeAsSec("spark.cleaner.ttl", "-1").toInt
+    conf.getTimeAsSeconds("spark.cleaner.ttl", "-1").toInt
   }
 
   def getDelaySeconds(

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

@@ -613,7 +613,7 @@ private[spark] object Utils extends Logging {
         Utils.setupSecureURLConnection(uc, securityMgr)
 
         val timeoutMs = 
-          conf.getTimeAsSec("spark.files.fetchTimeout", "60s").toInt * 1000
+          conf.getTimeAsSeconds("spark.files.fetchTimeout", "60s").toInt * 1000
         uc.setConnectTimeout(timeoutMs)
         uc.setReadTimeout(timeoutMs)
         uc.connect()
@@ -1016,15 +1016,15 @@ private[spark] object Utils extends Logging {
    * Convert a time parameter such as (50s, 100ms, or 250us) to microseconds for internal use. If
    * no suffix is provided, the passed number is assumed to be in ms.
    */
-  def timeStringAsMs(str: String) : Long = {
+  def timeStringAsMs(str: String): Long = {
     JavaUtils.timeStringAsMs(str)
   }
 
   /**
    * Convert a time parameter such as (50s, 100ms, or 250us) to microseconds for internal use. If
    * no suffix is provided, the passed number is assumed to be in seconds.
    */
-  def timeStringAsSec(str: String) : Long = {
+  def timeStringAsSeconds(str: String): Long = {
     JavaUtils.timeStringAsSec(str)
   }
 

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

@@ -39,58 +39,44 @@ class UtilsSuite extends FunSuite with ResetSystemProperties {
 
   test("timeConversion") {
     // Test -1
-    assert(Utils.timeStringAsSec("-1") === -1)
+    assert(Utils.timeStringAsSeconds("-1") === -1)
 
     // Test zero
-    assert(Utils.timeStringAsSec("0") === 0)
+    assert(Utils.timeStringAsSeconds("0") === 0)
 
-    assert(Utils.timeStringAsSec("1") === 1)
-    assert(Utils.timeStringAsSec("1s") === 1)
-    assert(Utils.timeStringAsSec("1000ms") === 1)
-    assert(Utils.timeStringAsSec("1000000us") === 1)
-    assert(Utils.timeStringAsSec("1min") === TimeUnit.MINUTES.toSeconds(1))
-    assert(Utils.timeStringAsSec("1h") === TimeUnit.HOURS.toSeconds(1))
-    assert(Utils.timeStringAsSec("1d") === TimeUnit.DAYS.toSeconds(1))
+    assert(Utils.timeStringAsSeconds("1") === 1)
+    assert(Utils.timeStringAsSeconds("1s") === 1)
+    assert(Utils.timeStringAsSeconds("1000ms") === 1)
+    assert(Utils.timeStringAsSeconds("1000000us") === 1)
+    assert(Utils.timeStringAsSeconds("1m") === TimeUnit.MINUTES.toSeconds(1))
+    assert(Utils.timeStringAsSeconds("1min") === TimeUnit.MINUTES.toSeconds(1))
+    assert(Utils.timeStringAsSeconds("1h") === TimeUnit.HOURS.toSeconds(1))
+    assert(Utils.timeStringAsSeconds("1d") === TimeUnit.DAYS.toSeconds(1))
 
     assert(Utils.timeStringAsMs("1") === 1)
     assert(Utils.timeStringAsMs("1ms") === 1)
     assert(Utils.timeStringAsMs("1000us") === 1)
     assert(Utils.timeStringAsMs("1s") === TimeUnit.SECONDS.toMillis(1))
+    assert(Utils.timeStringAsMs("1m") === TimeUnit.MINUTES.toMillis(1))
     assert(Utils.timeStringAsMs("1min") === TimeUnit.MINUTES.toMillis(1))
     assert(Utils.timeStringAsMs("1h") === TimeUnit.HOURS.toMillis(1))
     assert(Utils.timeStringAsMs("1d") === TimeUnit.DAYS.toMillis(1))
 
     // Test invalid strings
-    try {
+    intercept[NumberFormatException] {
       Utils.timeStringAsMs("This breaks 600s")
-      assert(false) // We should never reach this
-    } catch {
-      case e: NumberFormatException => assert(true)
     }
 
-    // Test invalid strings
-    try {
-      Utils.timeStringAsMs("600ds")
-      assert(false) // We should never reach this
-    } catch {
-      case e: NumberFormatException => assert(true)
+    intercept[NumberFormatException] {
+      Utils.timeStringAsMs("This breaks 600ds")
     }
 
-
-    // Test invalid strings
-    try {
+    intercept[NumberFormatException] {
       Utils.timeStringAsMs("600s This breaks")
-      assert(false) // We should never reach this
-    } catch {
-      case e: NumberFormatException => assert(true)
     }
 
-    // Test invalid strings
-    try {
+    intercept[NumberFormatException] {
       Utils.timeStringAsMs("This 123s breaks")
-      assert(false) // We should never reach this
-    } catch {
-      case e: NumberFormatException => assert(true)
     }
   }
 

docs/configuration.md

@@ -48,8 +48,6 @@ The following format is accepted:
     5d (days)
     1y (years)
 
-If no units are provided, the default unit specified in the documentation is assumed. 
-
 ## Dynamically Loading Spark Properties
 In some cases, you may want to avoid hard-coding certain configurations in a `SparkConf`. For
 instance, if you'd like to run the same application with different masters or different
@@ -444,7 +442,7 @@ Apart from these, the following properties are also available, and may be useful
   <td>5s</td>
   <td>
     (Netty only) How long to wait between retries of fetches. The maximum delay caused by retrying
-    is simply <code>maxRetries * retryWait</code>, by default 5 seconds.
+    is 15 seconds by default, calculated as <code>maxRetries * retryWait</code>.
   </td>
 </tr>
 <tr>

streaming/src/test/scala/org/apache/spark/streaming/StreamingContextSuite.scala

@@ -75,7 +75,7 @@ class StreamingContextSuite extends FunSuite with BeforeAndAfter with Timeouts w
     val myConf = SparkContext.updatedConf(new SparkConf(false), master, appName)
     myConf.set("spark.cleaner.ttl", "10s")
     ssc = new StreamingContext(myConf, batchDuration)
-    assert(ssc.conf.getTimeAsSec("spark.cleaner.ttl", "-1") === 10)
+    assert(ssc.conf.getTimeAsSeconds("spark.cleaner.ttl", "-1") === 10)
   }
 
   test("from existing SparkContext") {
@@ -87,7 +87,7 @@ class StreamingContextSuite extends FunSuite with BeforeAndAfter with Timeouts w
     val myConf = SparkContext.updatedConf(new SparkConf(false), master, appName)
     myConf.set("spark.cleaner.ttl", "10s")
     ssc = new StreamingContext(myConf, batchDuration)
-    assert(ssc.conf.getTimeAsSec("spark.cleaner.ttl", "-1") === 10)
+    assert(ssc.conf.getTimeAsSeconds("spark.cleaner.ttl", "-1") === 10)
   }
 
   test("from checkpoint") {
@@ -98,12 +98,12 @@ class StreamingContextSuite extends FunSuite with BeforeAndAfter with Timeouts w
     ssc1.start()
     val cp = new Checkpoint(ssc1, Time(1000))
     assert(
-      Utils.timeStringAsSec(cp.sparkConfPairs.toMap.getOrElse("spark.cleaner.ttl", "-1")) === 10)
+      Utils.timeStringAsSeconds(cp.sparkConfPairs.toMap.getOrElse("spark.cleaner.ttl", "-1")) === 10)
     ssc1.stop()
     val newCp = Utils.deserialize[Checkpoint](Utils.serialize(cp))
-    assert(newCp.createSparkConf().getTimeAsSec("spark.cleaner.ttl", "-1") === 10)
+    assert(newCp.createSparkConf().getTimeAsSeconds("spark.cleaner.ttl", "-1") === 10)
     ssc = new StreamingContext(null, newCp, null)
-    assert(ssc.conf.getTimeAsSec("spark.cleaner.ttl", "-1") === 10)
+    assert(ssc.conf.getTimeAsSeconds("spark.cleaner.ttl", "-1") === 10)
   }
 
   test("start and stop state check") {