Address the comments

Author: jerryshao

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

@@ -147,25 +147,8 @@ private[spark] class ExecutorAllocationManager(
   private val executor =
     ThreadUtils.newDaemonSingleThreadScheduledExecutor("spark-dynamic-executor-allocation")
 
-  // Metric source for ExecutorAllocationManager to expose the its internal executor allocation
-  // status to MetricsSystem.
-  private[spark] val executorAllocationManagerSource = new Source {
-    val sourceName = "ExecutorAllocationManager"
-    val metricRegistry = new MetricRegistry()
-
-    private def registerGauge[T](name: String, value: => T, defaultValue: T): Unit = {
-      metricRegistry.register(MetricRegistry.name("executors", name), new Gauge[T] {
-        override def getValue: T = synchronized { Option(value).getOrElse(defaultValue) }
-      })
-    }
-
-    registerGauge("numberExecutorsToAdd", numExecutorsToAdd, 0)
-    registerGauge("numberExecutorsPending", numExecutorsPending, 0)
-    registerGauge("numberExecutorsPendingToRemove", executorsPendingToRemove.size, 0)
-    registerGauge("numberAllExecutors", executorIds.size, 0)
-    registerGauge("numberTargetExecutors", targetNumExecutors(), 0)
-    registerGauge("numberMaxNeededExecutors", maxNumExecutorsNeeded(), 0)
-  }
+  // Metric source for ExecutorAllocationManager to expose internal status to MetricsSystem.
+  private[spark] val executorAllocationManagerSource = new ExecutorAllocationManagerSource
 
   /**
    * Verify that the settings specified through the config are valid.
@@ -602,6 +585,26 @@ private[spark] class ExecutorAllocationManager(
     }
   }
 
+  // Metric source for ExecutorAllocationManager to expose the its internal executor allocation
+  // status to MetricsSystem.
+  // Note: these metrics may not be stable across Spark version.
+  private[spark] class ExecutorAllocationManagerSource extends Source {
+    val sourceName = "ExecutorAllocationManager"
+    val metricRegistry = new MetricRegistry()
+
+    private def registerGauge[T](name: String, value: => T, defaultValue: T): Unit = {
+      metricRegistry.register(MetricRegistry.name("executors", name), new Gauge[T] {
+        override def getValue: T = synchronized { Option(value).getOrElse(defaultValue) }
+      })
+    }
+
+    registerGauge("numberExecutorsToAdd", numExecutorsToAdd, 0)
+    registerGauge("numberExecutorsPending", numExecutorsPending, 0)
+    registerGauge("numberExecutorsPendingToRemove", executorsPendingToRemove.size, 0)
+    registerGauge("numberAllExecutors", executorIds.size, 0)
+    registerGauge("numberTargetExecutors", targetNumExecutors(), 0)
+    registerGauge("numberMaxNeededExecutors", maxNumExecutorsNeeded(), 0)
+  }
 }
 
 private object ExecutorAllocationManager {