startServiceOnPort will use a SparkConf arg

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

@@ -57,7 +57,7 @@ private[spark] class HttpServer(
     } else {
       logInfo("Starting HTTP Server")
       val (actualServer, actualPort) =
-        Utils.startServiceOnPort[Server](requestedPort, doStart, serverName)
+        Utils.startServiceOnPort[Server](requestedPort, doStart, serverName, new SparkConf())
       server = actualServer
       port = actualPort
     }

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

@@ -176,8 +176,6 @@ class SparkContext(config: SparkConf) extends Logging with ExecutorAllocationCli
   logInfo(s"Running Spark version $SPARK_VERSION")
 
   private[spark] val conf = config.clone()
-  conf.getOption("spark.port.maxRetries")
-      .foreach(portRetriesConf => System.setProperty("spark.port.maxRetries", portRetriesConf))
   conf.validateSettings()
 
   /**

core/src/main/scala/org/apache/spark/network/nio/ConnectionManager.scala

@@ -174,7 +174,7 @@ private[nio] class ConnectionManager(
     serverChannel.socket.bind(new InetSocketAddress(port))
     (serverChannel, serverChannel.socket.getLocalPort)
   }
-  Utils.startServiceOnPort[ServerSocketChannel](port, startService, name)
+  Utils.startServiceOnPort[ServerSocketChannel](port, startService, name, conf)
   serverChannel.register(selector, SelectionKey.OP_ACCEPT)
 
   val id = new ConnectionManagerId(Utils.localHostName, serverChannel.socket.getLocalPort)

core/src/main/scala/org/apache/spark/ui/JettyUtils.scala

@@ -201,7 +201,7 @@ private[spark] object JettyUtils extends Logging {
       }
     }
 
-    val (server, boundPort) = Utils.startServiceOnPort[Server](port, connect, serverName)
+    val (server, boundPort) = Utils.startServiceOnPort[Server](port, connect, serverName, conf)
     ServerInfo(server, boundPort, collection)
   }
 

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

@@ -53,7 +53,7 @@ private[spark] object AkkaUtils extends Logging {
     val startService: Int => (ActorSystem, Int) = { actualPort =>
       doCreateActorSystem(name, host, actualPort, conf, securityManager)
     }
-    Utils.startServiceOnPort(port, startService, name)
+    Utils.startServiceOnPort(port, startService, name, conf)
   }
 
   private def doCreateActorSystem(

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

@@ -1692,12 +1692,12 @@ private[spark] object Utils extends Logging {
   /**
    * Default maximum number of retries when binding to a port before giving up.
    */
-  lazy val portMaxRetries: Int = {
+  def portMaxRetries(conf: SparkConf): Int = {
     if (sys.props.contains("spark.testing")) {
       // Set a higher number of retries for tests...
       sys.props.get("spark.port.maxRetries").map(_.toInt).getOrElse(100)
     } else {
-      sys.props.get("spark.port.maxRetries").map(_.toInt).getOrElse(16)
+      conf.getOption("spark.port.maxRetries").map(_.toInt).getOrElse(16)
     }
   }
 
@@ -1715,8 +1715,9 @@ private[spark] object Utils extends Logging {
       startPort: Int,
       startService: Int => (T, Int),
       serviceName: String = "",
-      maxRetries: Int = portMaxRetries): (T, Int) = {
+      conf: SparkConf): (T, Int) = {
     val serviceString = if (serviceName.isEmpty) "" else s" '$serviceName'"
+    val maxRetries = portMaxRetries(conf)
     logInfo(s"Starting service$serviceString on port $startPort with maximum $maxRetries retries. ")
     for (offset <- 0 to maxRetries) {
       // Do not increment port if startPort is 0, which is treated as a special port