code review comments

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

@@ -305,11 +305,6 @@ class SparkContext(config: SparkConf) extends Logging {
     _dagScheduler = ds
   }
 
-  private[spark] def heartbeater: Heartbeater = _heartbeater
-  private[spark] def heartbeater_=(hb: Heartbeater): Unit = {
-    _heartbeater = hb
-  }
-
   /**
    * A unique identifier for the Spark application.
    * Its format depends on the scheduler implementation.

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

@@ -1753,7 +1753,7 @@ class DAGScheduler(
     messageScheduler.shutdownNow()
     eventProcessLoop.stop()
     taskScheduler.stop()
-   }
+  }
 
   eventProcessLoop.start()
 }

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

@@ -96,7 +96,7 @@ private[spark] class EventLoggingListener(
   // Visible for tests only.
   private[scheduler] val logPath = getLogPath(logBaseDir, appId, appAttemptId, compressionCodecName)
 
-  // map of live stages, to peak executor metrics for the stage
+  // map of (stageId, stageAttempt), to peak executor metrics for the stage
   private val liveStageExecutorMetrics = HashMap[(Int, Int), HashMap[String, PeakExecutorMetrics]]()
 
   /**
@@ -197,6 +197,9 @@ private[spark] class EventLoggingListener(
       executorMap.foreach {
        executorEntry => {
           for ((executorId, peakExecutorMetrics) <- executorEntry) {
+            // -1 timestamp indicates that the ExecutorMetricsUpdate event is being read from the
+            // history log, and contains the peak metrics for the stage whose StageCompleted event
+            // immediately follows
             val executorMetrics = new ExecutorMetrics(-1, peakExecutorMetrics.metrics)
             val executorUpdate = new SparkListenerExecutorMetricsUpdate(
               executorId, accumUpdates, Some(executorMetrics))

core/src/main/scala/org/apache/spark/status/api/v1/api.scala

@@ -23,6 +23,7 @@ import scala.xml.{NodeSeq, Text}
 
 import com.fasterxml.jackson.annotation.JsonIgnoreProperties
 import com.fasterxml.jackson.core.{JsonGenerator, JsonParser}
+import com.fasterxml.jackson.core.`type`.TypeReference
 import com.fasterxml.jackson.databind.{DeserializationContext, JsonDeserializer, JsonSerializer, SerializerProvider}
 import com.fasterxml.jackson.databind.annotation.{JsonDeserialize, JsonSerialize}
 
@@ -113,26 +114,21 @@ class MemoryMetrics private[spark](
     val totalOffHeapStorageMemory: Long)
 
 /** deserialzer for peakMemoryMetrics: convert to array ordered by metric name */
-class PeakMemoryMetricsDeserializer extends JsonDeserializer[Option[Array[Long]]] {
+class PeakMemoryMetricsDeserializer private[spark] extends JsonDeserializer[Option[Array[Long]]] {
   override def deserialize(
       jsonParser: JsonParser,
       deserializationContext: DeserializationContext): Option[Array[Long]] = {
-    val metricsMap = jsonParser.readValueAs(classOf[Option[Map[String, Object]]])
+    val metricsMap = jsonParser.readValueAs[Option[Map[String, Long]]](
+      new TypeReference[Option[Map[String, java.lang.Long]]] {})
     metricsMap match {
       case Some(metrics) =>
-        Some(MetricGetter.values.map { m =>
-          metrics.getOrElse (m.name, 0L) match {
-            case intVal: Int => intVal.toLong
-            case longVal: Long => longVal
-          }
-        }.toArray)
+        Some(MetricGetter.values.map(m => metrics.getOrElse(m.name, 0L)).toArray)
       case None => None
     }
   }
 }
-
 /** serializer for peakMemoryMetrics: convert array to map with metric name as key */
-class PeakMemoryMetricsSerializer extends JsonSerializer[Option[Array[Long]]] {
+class PeakMemoryMetricsSerializer private[spark] extends JsonSerializer[Option[Array[Long]]] {
   override def serialize(
       metrics: Option[Array[Long]],
       jsonGenerator: JsonGenerator,

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

@@ -21,7 +21,7 @@ import java.io.{File, FileOutputStream, InputStream, IOException}
 
 import scala.collection.immutable.Map
 import scala.collection.mutable
-import scala.collection.mutable.ArrayBuffer
+import scala.collection.mutable.Set
 import scala.io.Source
 
 import org.apache.hadoop.fs.Path
@@ -301,48 +301,66 @@ class EventLoggingListenerSuite extends SparkFunSuite with LocalSparkContext wit
       createExecutorAddedEvent(1),
       createExecutorAddedEvent(2),
       createStageSubmittedEvent(0),
+      // receive 3 metric updates from each executor with just stage 0 running,
+      // with different peak updates for each executor
       createExecutorMetricsUpdateEvent(1,
         new ExecutorMetrics(10L,
           Array(4000L, 50L, 20L, 0L, 40L, 0L, 60L, 0L, 70L, 20L))),
       createExecutorMetricsUpdateEvent(2,
         new ExecutorMetrics(10L,
           Array(1500L, 50L, 20L, 0L, 0L, 0L, 20L, 0L, 70L, 0L))),
+      // exec 1: new stage 0 peaks for metrics at indexes: 2, 4, 6
       createExecutorMetricsUpdateEvent(1,
         new ExecutorMetrics(15L,
           Array(4000L, 50L, 50L, 0L, 50L, 0L, 100L, 0L, 70L, 20L))),
+      // exec 2: new stage 0 peaks for metrics at indexes: 0, 4, 6
       createExecutorMetricsUpdateEvent(2,
         new ExecutorMetrics(15L,
           Array(2000L, 50L, 10L, 0L, 10L, 0L, 30L, 0L, 70L, 0L))),
+      // exec 1: new stage 0 peaks for metrics at indexes: 5, 7
       createExecutorMetricsUpdateEvent(1,
         new ExecutorMetrics(20L,
           Array(2000L, 40L, 50L, 0L, 40L, 10L, 90L, 10L, 50L, 0L))),
+      // exec 2: new stage 0 peaks for metrics at indexes: 0, 5, 6, 7, 8
       createExecutorMetricsUpdateEvent(2,
         new ExecutorMetrics(20L,
           Array(3500L, 50L, 15L, 0L, 10L, 10L, 35L, 10L, 80L, 0L))),
+      // now start stage 1, one more metric update for each executor, and new
+      // peaks for some stage 1 metrics (as listed), initialize stage 1 peaks
       createStageSubmittedEvent(1),
+      // exec 1: new stage 0 peaks for metrics at indexes: 0, 3, 7
       createExecutorMetricsUpdateEvent(1,
         new ExecutorMetrics(25L,
           Array(5000L, 30L, 50L, 20L, 30L, 10L, 80L, 30L, 50L, 0L))),
+      // exec 2: new stage 0 peaks for metrics at indexes: 0, 1, 2, 3, 6, 7, 9
       createExecutorMetricsUpdateEvent(2,
         new ExecutorMetrics(25L,
           Array(7000L, 70L, 50L, 20L, 0L, 10L, 50L, 30L, 10L, 40L))),
+      // complete stage 0, and 3 more updates for each executor with just
+      // stage 1 running
       createStageCompletedEvent(0),
+      // exec 1: new stage 1 peaks for metrics at indexes: 0, 1, 3
       createExecutorMetricsUpdateEvent(1,
         new ExecutorMetrics(30L,
           Array(6000L, 70L, 20L, 30L, 10L, 0L, 30L, 30L, 30L, 0L))),
+      // exec 2: new stage 1 peaks for metrics at indexes: 3, 4, 7, 8
       createExecutorMetricsUpdateEvent(2,
         new ExecutorMetrics(30L,
           Array(5500L, 30L, 20L, 40L, 10L, 0L, 30L, 40L, 40L, 20L))),
+      // exec 1: new stage 1 peaks for metrics at indexes: 0, 4, 5, 7
       createExecutorMetricsUpdateEvent(1,
         new ExecutorMetrics(35L,
           Array(7000L, 70L, 5L, 25L, 60L, 30L, 65L, 55L, 30L, 0L))),
+      // exec 2: new stage 1 peak for metrics at index: 7
       createExecutorMetricsUpdateEvent(2,
         new ExecutorMetrics(35L,
           Array(5500L, 40L, 25L, 30L, 10L, 30L, 35L, 60L, 0L, 20L))),
+      // exec 1: no new stage 1 peaks
       createExecutorMetricsUpdateEvent(1,
         new ExecutorMetrics(40L,
           Array(5500L, 70L, 15L, 20L, 55L, 20L, 70L, 40L, 20L, 0L))),
       createExecutorRemovedEvent(1),
+      // exec 2: new stage 1 peak for metrics at index: 6
       createExecutorMetricsUpdateEvent(2,
         new ExecutorMetrics(40L,
           Array(4000L, 20L, 25L, 30L, 10L, 30L, 35L, 60L, 0L, 0L))),
@@ -373,11 +391,14 @@ class EventLoggingListenerSuite extends SparkFunSuite with LocalSparkContext wit
         event match {
           case metricsUpdate: SparkListenerExecutorMetricsUpdate =>
           case stageCompleted: SparkListenerStageCompleted =>
+            val execIds = Set[String]()
             (1 to 2).foreach { _ =>
-              checkExecutorMetricsUpdate(lines(logIdx), stageCompleted.stageInfo.stageId,
-                expectedMetricsEvents)
+              val execId = checkExecutorMetricsUpdate(lines(logIdx),
+                stageCompleted.stageInfo.stageId, expectedMetricsEvents)
+              execIds += execId
               logIdx += 1
             }
+            assert(execIds.size == 2) // check that each executor was logged
             checkEvent(lines(logIdx), event)
             logIdx += 1
         case _ =>
@@ -462,16 +483,17 @@ class EventLoggingListenerSuite extends SparkFunSuite with LocalSparkContext wit
   private def checkExecutorMetricsUpdate(
       line: String,
       stageId: Int,
-      expectedEvents: Map[(Int, String), SparkListenerExecutorMetricsUpdate]): Unit = {
+      expectedEvents: Map[(Int, String), SparkListenerExecutorMetricsUpdate]): String = {
     JsonProtocol.sparkEventFromJson(parse(line)) match {
       case executorMetrics: SparkListenerExecutorMetricsUpdate =>
           expectedEvents.get((stageId, executorMetrics.execId)) match {
             case Some(expectedMetrics) =>
               assert(executorMetrics.accumUpdates.isEmpty)
               checkExecutorMetrics(executorMetrics.executorUpdates, expectedMetrics.executorUpdates)
-            case None =>
+             case None =>
               assert(false)
         }
+        executorMetrics.execId
       case _ =>
         fail("expecting SparkListenerExecutorMetricsUpdate")
     }