code review comments

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

@@ -68,10 +68,10 @@ class ExecutorMetrics private[spark] extends Serializable {
    * @return if there is a new peak value for any metric
    */
   private[spark] def compareAndUpdatePeakValues(executorMetrics: ExecutorMetrics): Boolean = {
-    var updated: Boolean = false
+    var updated = false
 
     (0 until ExecutorMetricType.values.length).foreach { idx =>
-       if ( executorMetrics.metrics(idx) > metrics(idx)) {
+       if (executorMetrics.metrics(idx) > metrics(idx)) {
         updated = true
         metrics(idx) = executorMetrics.metrics(idx)
       }

core/src/main/scala/org/apache/spark/memory/MemoryManager.scala

@@ -183,22 +183,30 @@ private[spark] abstract class MemoryManager(
   /**
    *  On heap execution memory currently in use, in bytes.
    */
-  final def onHeapExecutionMemoryUsed: Long = onHeapExecutionMemoryPool.memoryUsed
+  final def onHeapExecutionMemoryUsed: Long = synchronized {
+    onHeapExecutionMemoryPool.memoryUsed
+  }
 
   /**
    *  Off heap execution memory currently in use, in bytes.
    */
-  final def offHeapExecutionMemoryUsed: Long = offHeapExecutionMemoryPool.memoryUsed
+  final def offHeapExecutionMemoryUsed: Long = synchronized {
+    offHeapExecutionMemoryPool.memoryUsed
+  }
 
   /**
    *  On heap storage memory currently in use, in bytes.
    */
-  final def onHeapStorageMemoryUsed: Long = onHeapStorageMemoryPool.memoryUsed
+  final def onHeapStorageMemoryUsed: Long = synchronized {
+    onHeapStorageMemoryPool.memoryUsed
+  }
 
   /**
    *  Off heap storage memory currently in use, in bytes.
    */
-  final def offHeapStorageMemoryUsed: Long = offHeapStorageMemoryPool.memoryUsed
+  final def offHeapStorageMemoryUsed: Long = synchronized {
+    offHeapStorageMemoryPool.memoryUsed
+  }
 
   /**
    * Returns the execution memory consumption, in bytes, for the given task.

core/src/main/scala/org/apache/spark/metrics/ExecutorMetricType.scala

@@ -36,7 +36,7 @@ private[spark] abstract class MemoryManagerExecutorMetricType(
   }
 }
 
-private[spark]abstract class MBeanExecutorMetricType(mBeanName: String)
+private[spark] abstract class MBeanExecutorMetricType(mBeanName: String)
   extends ExecutorMetricType {
   private val bean = ManagementFactory.newPlatformMXBeanProxy(
     ManagementFactory.getPlatformMBeanServer,

core/src/main/scala/org/apache/spark/status/AppStatusListener.scala

@@ -690,8 +690,8 @@ private[spark] class AppStatusListener(
     // check if there is a new peak value for any of the executor level memory metrics
     // for the live UI. SparkListenerExecutorMetricsUpdate events are only processed
     // for the live UI.
-    event.executorUpdates.foreach { updates: ExecutorMetrics =>
-      liveExecutors.get(event.execId).foreach { exec: LiveExecutor =>
+    event.executorUpdates.foreach { updates =>
+      liveExecutors.get(event.execId).foreach { exec =>
         if (exec.peakExecutorMetrics.compareAndUpdatePeakValues(updates)) {
           maybeUpdate(exec, now)
         }