[SPARK-27198][core] Heartbeat interval mismatch in driver and executor

common/network-common/src/main/java/org/apache/spark/network/util/JavaUtils.java

@@ -226,6 +226,14 @@ private static boolean isSymlink(File file) throws IOException {
    * The unit is also considered the default if the given string does not specify a unit.
    */
   public static long timeStringAs(String str, TimeUnit unit) {
+    return timeStringAs(str, unit, unit);
+  }
+
+  /**
+   * Convert a passed time string (e.g. 50s, 100ms, or 250us) to a time count in the given unit.
+   * defaultUnit is used for string which have just number and no units mentioned
+   */
+  public static long timeStringAs(String str, TimeUnit unit, TimeUnit defaultUnit) {
     String lower = str.toLowerCase(Locale.ROOT).trim();
 
     try {
@@ -243,7 +251,7 @@ public static long timeStringAs(String str, TimeUnit unit) {
       }
 
       // If suffix is valid use that, otherwise none was provided and use the default passed
-      return unit.convert(val, suffix != null ? timeSuffixes.get(suffix) : unit);
+      return unit.convert(val, suffix != null ? timeSuffixes.get(suffix) : defaultUnit);
     } catch (NumberFormatException e) {
       String timeError = "Time must be specified as seconds (s), " +
               "milliseconds (ms), microseconds (us), minutes (m or min), hour (h), or day (d). " +

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

@@ -18,7 +18,7 @@
 package org.apache.spark
 
 import java.util.{Map => JMap}
-import java.util.concurrent.ConcurrentHashMap
+import java.util.concurrent.{ConcurrentHashMap, TimeUnit}
 
 import scala.collection.JavaConverters._
 import scala.collection.mutable.LinkedHashSet
@@ -28,6 +28,7 @@ import org.apache.avro.{Schema, SchemaNormalization}
 import org.apache.spark.deploy.history.config._
 import org.apache.spark.internal.Logging
 import org.apache.spark.internal.config._
+import org.apache.spark.network.util.JavaUtils
 import org.apache.spark.serializer.KryoSerializer
 import org.apache.spark.util.Utils
 
@@ -280,6 +281,16 @@ class SparkConf(loadDefaults: Boolean) extends Cloneable with Logging with Seria
     Utils.timeStringAsSeconds(get(key, defaultValue))
   }
 
+  /**
+   * Get a time parameter as seconds, falling back to a default if not set. If no
+   * suffix is provided then defaultUnit is assumed.
+   * @throws NumberFormatException If the value cannot be interpreted as seconds
+   */
+  def getTimeAsSeconds(key: String, defaultValue: String, defaultUnit: TimeUnit): Long =
+    catchIllegalValue(key) {
+      JavaUtils.timeStringAs(get(key, defaultValue), TimeUnit.SECONDS, defaultUnit)
+  }
+
   /**
    * Get a time parameter as milliseconds; throws a NoSuchElementException if it's not set. If no
    * suffix is provided then milliseconds are assumed.
@@ -610,7 +621,8 @@ class SparkConf(loadDefaults: Boolean) extends Cloneable with Logging with Seria
       s"${NETWORK_AUTH_ENABLED.key} must be enabled when enabling encryption.")
 
     val executorTimeoutThreshold = getTimeAsSeconds("spark.network.timeout", "120s")
-    val executorHeartbeatInterval = getTimeAsSeconds("spark.executor.heartbeatInterval", "10s")
+    val executorHeartbeatInterval =
+      getTimeAsSeconds("spark.executor.heartbeatInterval", "10s", TimeUnit.MILLISECONDS)
     // If spark.executor.heartbeatInterval bigger than spark.network.timeout,
     // it will almost always cause ExecutorLostFailure. See SPARK-22754.
     require(executorTimeoutThreshold > executorHeartbeatInterval, "The value of " +

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

@@ -28,6 +28,7 @@ import javax.annotation.concurrent.GuardedBy
 
 import scala.collection.JavaConverters._
 import scala.collection.mutable.{ArrayBuffer, HashMap, Map}
+import scala.concurrent.duration._
 import scala.util.control.NonFatal
 
 import com.google.common.util.concurrent.ThreadFactoryBuilder
@@ -71,6 +72,11 @@ private[spark] class Executor(
 
   private val conf = env.conf
 
+  private val HEARTBEAT_INTERVAL_KEY = "spark.executor.heartbeatInterval"
+
+  private val heartbeatIntervalInSec =
+    conf.getTimeAsSeconds(HEARTBEAT_INTERVAL_KEY, "10s", TimeUnit.MILLISECONDS).seconds
+
   // No ip or host:port - just hostname
   Utils.checkHost(executorHostname)
   // must not have port specified.
@@ -832,8 +838,9 @@ private[spark] class Executor(
 
     val message = Heartbeat(executorId, accumUpdates.toArray, env.blockManager.blockManagerId)
     try {
+
       val response = heartbeatReceiverRef.askSync[HeartbeatResponse](
-          message, RpcTimeout(conf, "spark.executor.heartbeatInterval", "10s"))
+          message, new RpcTimeout(heartbeatIntervalInSec, HEARTBEAT_INTERVAL_KEY))
       if (response.reregisterBlockManager) {
         logInfo("Told to re-register on heartbeat")
         env.blockManager.reregister()
@@ -855,7 +862,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 = heartbeatIntervalInSec.toMillis
 
     // Wait a random interval so the heartbeats don't end up in sync
     val initialDelay = intervalMs + (math.random * intervalMs).asInstanceOf[Int]