Address the comments

Author: jerryshao

yarn/src/main/scala/org/apache/spark/deploy/yarn/ClientArguments.scala

@@ -81,7 +81,7 @@ private[spark] class ClientArguments(args: Array[String], sparkConf: SparkConf)
       .orNull
     // If dynamic allocation is enabled, start at the configured initial number of executors.
     // Default to minExecutors if no initialExecutors is set.
-    numExecutors = YarnSparkHadoopUtil.getTargetExecutorNumber(sparkConf)
+    numExecutors = YarnSparkHadoopUtil.getInitialTargetExecutorNumber(sparkConf)
     principal = Option(principal)
       .orElse(sparkConf.getOption("spark.yarn.principal"))
       .orNull

yarn/src/main/scala/org/apache/spark/deploy/yarn/YarnAllocator.scala

@@ -88,7 +88,8 @@ private[yarn] class YarnAllocator(
   private var executorIdCounter = 0
   @volatile private var numExecutorsFailed = 0
 
-  @volatile private var targetNumExecutors = YarnSparkHadoopUtil.getTargetExecutorNumber(sparkConf)
+  @volatile private var targetNumExecutors =
+    YarnSparkHadoopUtil.getInitialTargetExecutorNumber(sparkConf)
 
   // Executor loss reason requests that are pending - maps from executor ID for inquiry to a
   // list of requesters that should be responded to once we find out why the given executor

yarn/src/main/scala/org/apache/spark/deploy/yarn/YarnSparkHadoopUtil.scala

@@ -316,7 +316,7 @@ object YarnSparkHadoopUtil {
   }
 
   /** Get the initial target number of executors depends on dynamic allocation is enabled or not */
-  def getTargetExecutorNumber(conf: SparkConf): Int = {
+  def getInitialTargetExecutorNumber(conf: SparkConf): Int = {
     if (Utils.isDynamicAllocationEnabled(conf)) {
       val minNumExecutors = conf.getInt("spark.dynamicAllocation.minExecutors", 0)
       val initialNumExecutors =
@@ -328,10 +328,10 @@ object YarnSparkHadoopUtil {
 
       initialNumExecutors
     } else {
-      var targetNumExecutors = DEFAULT_NUMBER_EXECUTORS
-      if (System.getenv("SPARK_EXECUTOR_INSTANCES") != null) {
-        targetNumExecutors = IntParam.unapply(System.getenv("SPARK_EXECUTOR_INSTANCES"))
-          .getOrElse(targetNumExecutors)
+      val targetNumExecutors = if (System.getenv("SPARK_EXECUTOR_INSTANCES") != null) {
+        IntParam.unapply(System.getenv("SPARK_EXECUTOR_INSTANCES")).get
+      } else {
+        DEFAULT_NUMBER_EXECUTORS
       }
       // System property can override environment variable.
       conf.getInt("spark.executor.instances", targetNumExecutors)

yarn/src/main/scala/org/apache/spark/scheduler/cluster/YarnClusterSchedulerBackend.scala

@@ -32,7 +32,7 @@ private[spark] class YarnClusterSchedulerBackend(
 
   override def start() {
     super.start()
-    totalExpectedExecutors = YarnSparkHadoopUtil.getTargetExecutorNumber(sc.conf)
+    totalExpectedExecutors = YarnSparkHadoopUtil.getInitialTargetExecutorNumber(sc.conf)
   }
 
   override def applicationId(): String =