[SPARK-24560][CORE][MESOS] Fix some getTimeAsMs as getTimeAsSeconds

core/src/main/scala/org/apache/spark/api/python/PythonRunner.scala

@@ -354,7 +354,7 @@ private[spark] abstract class BasePythonRunner[IN, OUT](
     extends Thread(s"Worker Monitor for $pythonExec") {
 
     /** How long to wait before killing the python worker if a task cannot be interrupted. */
-    private val taskKillTimeout = env.conf.getTimeAsMs("spark.python.task.killTimeout", "2s")
+    private val taskKillTimeout = env.conf.getTimeAsSeconds("spark.python.task.killTimeout", "2s") * 1000L
 
     setDaemon(true)
 

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

@@ -58,7 +58,7 @@ private[deploy] class DriverRunner(
 
   // Timeout to wait for when trying to terminate a driver.
   private val DRIVER_TERMINATE_TIMEOUT_MS =
-    conf.getTimeAsMs("spark.worker.driverTerminateTimeout", "10s")
+    conf.getTimeAsSeconds("spark.worker.driverTerminateTimeout", "10s") * 1000L
 
   // Decoupled for testing
   def setClock(_clock: Clock): Unit = {

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

@@ -613,7 +613,7 @@ private[spark] class Executor(
     private[this] val taskId: Long = taskRunner.taskId
 
     private[this] val killPollingIntervalMs: Long =
-      conf.getTimeAsMs("spark.task.reaper.pollingInterval", "10s")
+      conf.getTimeAsSeconds("spark.task.reaper.pollingInterval", "10s") * 1000L
 
     private[this] val killTimeoutMs: Long = conf.getTimeAsMs("spark.task.reaper.killTimeout", "-1")
 
@@ -820,7 +820,7 @@ private[spark] class Executor(
    * Schedules a task to report heartbeat and partial metrics for active tasks to driver.
    */
   private def startDriverHeartbeater(): Unit = {
-    val intervalMs = conf.getTimeAsMs("spark.executor.heartbeatInterval", "10s")
+    val intervalMs = conf.getTimeAsSeconds("spark.executor.heartbeatInterval", "10s") * 1000L
 
     // Wait a random interval so the heartbeats don't end up in sync
     val initialDelay = intervalMs + (math.random * intervalMs).asInstanceOf[Int]

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

@@ -80,7 +80,7 @@ private[spark] class TaskSchedulerImpl(
     ThreadUtils.newDaemonSingleThreadScheduledExecutor("task-scheduler-speculation")
 
   // Threshold above which we warn user initial TaskSet may be starved
-  val STARVATION_TIMEOUT_MS = conf.getTimeAsMs("spark.starvation.timeout", "15s")
+  val STARVATION_TIMEOUT_MS = conf.getTimeAsSeconds("spark.starvation.timeout", "15s") * 1000L
 
   // CPUs to request per task
   val CPUS_PER_TASK = conf.getInt("spark.task.cpus", 1)

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

@@ -58,7 +58,7 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
   // Submit tasks after maxRegisteredWaitingTime milliseconds
   // if minRegisteredRatio has not yet been reached
   private val maxRegisteredWaitingTimeMs =
-    conf.getTimeAsMs("spark.scheduler.maxRegisteredResourcesWaitingTime", "30s")
+    conf.getTimeAsSeconds("spark.scheduler.maxRegisteredResourcesWaitingTime", "30s") * 1000L
   private val createTime = System.currentTimeMillis()
 
   // Accessing `executorDataMap` in `DriverEndpoint.receive/receiveAndReply` doesn't need any
@@ -108,7 +108,7 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
 
     override def onStart() {
       // Periodically revive offers to allow delay scheduling to work
-      val reviveIntervalMs = conf.getTimeAsMs("spark.scheduler.revive.interval", "1s")
+      val reviveIntervalMs = conf.getTimeAsSeconds("spark.scheduler.revive.interval", "1s") * 1000L
 
       reviveThread.scheduleAtFixedRate(new Runnable {
         override def run(): Unit = Utils.tryLogNonFatalError {

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

@@ -34,7 +34,7 @@ private[spark] class ConsoleProgressBar(sc: SparkContext) extends Logging {
   private val CR = '\r'
   // Update period of progress bar, in milliseconds
   private val updatePeriodMSec =
-    sc.getConf.getTimeAsMs("spark.ui.consoleProgress.update.interval", "200")
+    sc.getConf.getTimeAsMs("spark.ui.consoleProgress.update.interval", "200ms")
   // Delay to show up a progress bar, in milliseconds
   private val firstDelayMSec = 500L
 

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

@@ -39,7 +39,7 @@ private[spark] object RpcUtils {
 
   /** Returns the configured number of milliseconds to wait on each retry */
   def retryWaitMs(conf: SparkConf): Long = {
-    conf.getTimeAsMs("spark.rpc.retry.wait", "3s")
+    conf.getTimeAsSeconds("spark.rpc.retry.wait", "3s") * 1000L
   }
 
   /** Returns the default Spark timeout to use for RPC ask operations. */

resource-managers/mesos/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosCoarseGrainedSchedulerBackend.scala

@@ -85,7 +85,7 @@ private[spark] class MesosCoarseGrainedSchedulerBackend(
   private val taskLabels = conf.get("spark.mesos.task.labels", "")
 
   private[this] val shutdownTimeoutMS =
-    conf.getTimeAsMs("spark.mesos.coarse.shutdownTimeout", "10s")
+    (conf.getTimeAsSeconds("spark.mesos.coarse.shutdownTimeout", "10s") * 1000L)
       .ensuring(_ >= 0, "spark.mesos.coarse.shutdownTimeout must be >= 0")
 
   // Synchronization protected by stateLock
@@ -635,7 +635,7 @@ private[spark] class MesosCoarseGrainedSchedulerBackend(
             externalShufflePort,
             sc.conf.getTimeAsMs("spark.storage.blockManagerSlaveTimeoutMs",
               s"${sc.conf.getTimeAsSeconds("spark.network.timeout", "120s") * 1000L}ms"),
-            sc.conf.getTimeAsMs("spark.executor.heartbeatInterval", "10s"))
+            sc.conf.getTimeAsSeconds("spark.executor.heartbeatInterval", "10s") * 1000L)
         slave.shuffleRegistered = true
       }