Address various feedback

This includes moving the Yarn logic out of BM into Utils,
updating a few test comments, adding a HadoopConfigProvider
and using it in the Yarn shuffle service.

Author: Andrew Or

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

@@ -79,7 +79,7 @@ private[spark] class ExecutorAllocationManager(sc: SparkContext) extends Logging
   private val executorIdleTimeout = conf.getLong(
     "spark.dynamicAllocation.executorIdleTimeout", 600)
 
-  // Whether we are testing this class. This should only be used internally.
+  // During testing, the methods to actually kill and add executors are mocked out
   private val testing = conf.getBoolean("spark.dynamicAllocation.testing", false)
 
   validateSettings()

core/src/main/scala/org/apache/spark/storage/BlockManager.scala

@@ -30,7 +30,6 @@ import akka.actor.{ActorSystem, Props}
 import sun.nio.ch.DirectBuffer
 
 import org.apache.spark._
-import org.apache.spark.deploy.SparkHadoopUtil
 import org.apache.spark.executor._
 import org.apache.spark.io.CompressionCodec
 import org.apache.spark.network._
@@ -41,7 +40,6 @@ import org.apache.spark.network.util.{ConfigProvider, TransportConf}
 import org.apache.spark.serializer.Serializer
 import org.apache.spark.shuffle.ShuffleManager
 import org.apache.spark.shuffle.hash.HashShuffleManager
-import org.apache.spark.shuffle.sort.SortShuffleManager
 import org.apache.spark.util._
 
 private[spark] sealed trait BlockValues
@@ -93,18 +91,7 @@ private[spark] class BlockManager(
 
   private[spark]
   val externalShuffleServiceEnabled = conf.getBoolean("spark.shuffle.service.enabled", false)
-
-  // In Yarn, the shuffle service port maybe set through the Hadoop config
-  private val shuffleServicePortKey = "spark.shuffle.service.port"
-  private val externalShuffleServicePort = {
-    val sparkPort = conf.getInt(shuffleServicePortKey, 7337)
-    if (SparkHadoopUtil.get.isYarnMode) {
-      val hadoopConf = SparkHadoopUtil.get.newConfiguration(conf)
-      hadoopConf.getInt(shuffleServicePortKey, sparkPort)
-    } else {
-      sparkPort
-    }
-  }
+  private val externalShuffleServicePort = Utils.getExternalShuffleServicePort(conf)
 
   // Check that we're not using external shuffle service with consolidated shuffle files.
   if (externalShuffleServiceEnabled

core/src/main/scala/org/apache/spark/util/Utils.scala

@@ -44,6 +44,7 @@ import org.json4s._
 import tachyon.client.{TachyonFile,TachyonFS}
 
 import org.apache.spark._
+import org.apache.spark.deploy.SparkHadoopUtil
 import org.apache.spark.serializer.{DeserializationStream, SerializationStream, SerializerInstance}
 
 /** CallSite represents a place in user code. It can have a short and a long form. */
@@ -1767,6 +1768,21 @@ private[spark] object Utils extends Logging {
       val manifest = new JarManifest(manifestUrl.openStream())
       manifest.getMainAttributes.getValue(Name.IMPLEMENTATION_VERSION)
     }.getOrElse("Unknown")
+
+  /**
+   * Return the port used in the external shuffle service as specified through
+   * `spark.shuffle.service.port`. In Yarn, this is set in the Hadoop configuration.
+   */
+  def getExternalShuffleServicePort(conf: SparkConf): Int = {
+    val shuffleServicePortKey = "spark.shuffle.service.port"
+    val sparkPort = conf.getInt(shuffleServicePortKey, 7337)
+    if (SparkHadoopUtil.get.isYarnMode) {
+      val hadoopConf = SparkHadoopUtil.get.newConfiguration(conf)
+      hadoopConf.getInt(shuffleServicePortKey, sparkPort)
+    } else {
+      sparkPort
+    }
+  }
 }
 
 /**

network/yarn/src/main/java/org/apache/spark/network/yarn/YarnShuffleService.java

@@ -20,13 +20,6 @@
 import java.lang.Override;
 import java.nio.ByteBuffer;
 
-import org.apache.spark.network.TransportContext;
-import org.apache.spark.network.server.RpcHandler;
-import org.apache.spark.network.server.TransportServer;
-import org.apache.spark.network.shuffle.ExternalShuffleBlockHandler;
-import org.apache.spark.network.util.TransportConf;
-import org.apache.spark.network.util.SystemPropertyConfigProvider;
-
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.yarn.api.records.ApplicationId;
 import org.apache.hadoop.yarn.api.records.ContainerId;
@@ -38,6 +31,13 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.apache.spark.network.TransportContext;
+import org.apache.spark.network.server.RpcHandler;
+import org.apache.spark.network.server.TransportServer;
+import org.apache.spark.network.shuffle.ExternalShuffleBlockHandler;
+import org.apache.spark.network.util.TransportConf;
+import org.apache.spark.network.yarn.util.HadoopConfigProvider;
+
 /**
  * External shuffle service used by Spark on Yarn.
  */
@@ -63,7 +63,7 @@ protected void serviceInit(Configuration conf) {
     try {
       int port = conf.getInt(
         SPARK_SHUFFLE_SERVICE_PORT_KEY, DEFAULT_SPARK_SHUFFLE_SERVICE_PORT);
-      TransportConf transportConf = new TransportConf(new SystemPropertyConfigProvider());
+      TransportConf transportConf = new TransportConf(new HadoopConfigProvider(conf));
       RpcHandler rpcHandler = new ExternalShuffleBlockHandler();
       TransportContext transportContext = new TransportContext(transportConf, rpcHandler);
       shuffleServer = transportContext.createServer(port);

network/yarn/src/main/java/org/apache/spark/network/yarn/util/HadoopConfigProvider.java

@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.network.yarn.util;
+
+import java.util.NoSuchElementException;
+
+import org.apache.hadoop.conf.Configuration;
+
+import org.apache.spark.network.util.ConfigProvider;
+
+/** Use the Hadoop configuration to obtain config values. */
+public class HadoopConfigProvider extends ConfigProvider {
+  private final Configuration conf;
+
+  public HadoopConfigProvider(Configuration conf) {
+    this.conf = conf;
+  }
+
+  @Override
+  public String get(String name) {
+    String value = conf.get(name);
+    if (value == null) {
+      throw new NoSuchElementException(name);
+    }
+    return value;
+  }
+}