addressed review comment

Author: LucaCanali

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

@@ -629,11 +629,7 @@ class SparkContext(config: SparkConf) extends Logging {
     _env.metricsSystem.registerSource(_dagScheduler.metricsSource)
     _env.metricsSystem.registerSource(new BlockManagerSource(_env.blockManager))
     _env.metricsSystem.registerSource(new JVMCPUSource())
-    _executorMetricsSource match {
-      case Some(executorMetricsSource: ExecutorMetricsSource) =>
-        executorMetricsSource.register(_env.metricsSystem)
-      case None => None
-    }
+    _executorMetricsSource.foreach(_.register(_env.metricsSystem))
     _executorAllocationManager.foreach { e =>
       _env.metricsSystem.registerSource(e.executorAllocationManagerSource)
     }
@@ -2487,12 +2483,7 @@ class SparkContext(config: SparkConf) extends Logging {
   /** Reports heartbeat metrics for the driver. */
   private def reportHeartBeat(executorMetricsSource: Option[ExecutorMetricsSource]): Unit = {
     val currentMetrics = ExecutorMetrics.getCurrentMetrics(env.memoryManager)
-
-    executorMetricsSource match {
-      case Some(executorMetricsSource: ExecutorMetricsSource) =>
-        executorMetricsSource.updateMetricsSnapshot(currentMetrics)
-      case None => None
-    }
+    executorMetricsSource.foreach(_.updateMetricsSnapshot(currentMetrics))
 
     val driverUpdates = new HashMap[(Int, Int), ExecutorMetrics]
     // In the driver, we do not track per-stage metrics, so use a dummy stage for the key

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

@@ -124,11 +124,7 @@ private[spark] class Executor(
     env.blockManager.initialize(conf.getAppId)
     env.metricsSystem.registerSource(executorSource)
     env.metricsSystem.registerSource(new JVMCPUSource())
-    executorMetricsSource match {
-      case Some(executorMetricsSource: ExecutorMetricsSource) =>
-        executorMetricsSource.register(env.metricsSystem)
-      case None => None
-    }
+    executorMetricsSource.foreach(_.register(env.metricsSystem))
     env.metricsSystem.registerSource(env.blockManager.shuffleMetricsSource)
   }
 

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

@@ -80,12 +80,7 @@ private[spark] class ExecutorMetricsPoller(
 
     // get the latest values for the metrics
     val latestMetrics = ExecutorMetrics.getCurrentMetrics(memoryManager)
-
-    executorMetricsSource match {
-      case Some(executorMetricsSource: ExecutorMetricsSource) =>
-        executorMetricsSource.updateMetricsSnapshot(latestMetrics)
-      case None => None
-    }
+    executorMetricsSource.foreach(_.updateMetricsSnapshot(latestMetrics))
 
     def updatePeaks(metrics: AtomicLongArray): Unit = {
       (0 until metrics.length).foreach { i =>

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

@@ -22,15 +22,19 @@ import com.codahale.metrics.{Gauge, MetricRegistry}
 import org.apache.spark.metrics.{ExecutorMetricType, MetricsSystem}
 import org.apache.spark.metrics.source.Source
 
-// Expose executor metrics from [[ExecutorMetricsType]] using the Dropwizard metrics system
-// Metrics related the memory system can be expensive to gather, we implement some optimizations:
-// (1) Metrics values are cached, updated at each heartbeat (default period is 10 seconds)
-// An alternative faster polling mechanism is used only if activated, by setting
-// spark.executor.metrics.pollingInterval=<interval in ms>
-// (2) procfs metrics are gathered all in one-go and only conditionally:
-// if the /proc filesystem exists
-// and spark.eventLog.logStageExecutorProcessTreeMetrics.enabled=true
-// and spark.eventLog.logStageExecutorMetrics.enabled=true
+/**
+ * Expose executor metrics from [[ExecutorMetricsType]] using the Dropwizard metrics system.
+ *
+ * Metrics related to the memory system can be expensive to gather, therefore
+ * we implement some optimizations:
+ * (1) Metrics values are cached, updated at each heartbeat (default period is 10 seconds).
+ * An alternative faster polling mechanism is used, only if activated, by setting
+ * spark.executor.metrics.pollingInterval=<interval in ms>.
+ * (2) Procfs metrics are gathered all in one-go and only conditionally:
+ * if the /proc filesystem exists
+ * and spark.eventLog.logStageExecutorProcessTreeMetrics.enabled=true
+ * and spark.eventLog.logStageExecutorMetrics.enabled=true.
+ */
 private[spark] class ExecutorMetricsSource extends Source {
 
   override val metricRegistry = new MetricRegistry()