diff --git a/core/src/main/scala/org/apache/spark/HttpServer.scala b/core/src/main/scala/org/apache/spark/HttpServer.scala index 912558d0cab7d..d8d6deef1c33a 100644 --- a/core/src/main/scala/org/apache/spark/HttpServer.scala +++ b/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 } diff --git a/core/src/main/scala/org/apache/spark/SparkContext.scala b/core/src/main/scala/org/apache/spark/SparkContext.scala index 9807e392423c1..3bf3acd245d8f 100644 --- a/core/src/main/scala/org/apache/spark/SparkContext.scala +++ b/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() /** diff --git a/core/src/main/scala/org/apache/spark/network/nio/ConnectionManager.scala b/core/src/main/scala/org/apache/spark/network/nio/ConnectionManager.scala index 3340fca08014e..76c313b27f0d6 100644 --- a/core/src/main/scala/org/apache/spark/network/nio/ConnectionManager.scala +++ b/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) diff --git a/core/src/main/scala/org/apache/spark/ui/JettyUtils.scala b/core/src/main/scala/org/apache/spark/ui/JettyUtils.scala index 2a27d49d2de05..3bc1b8a84c551 100644 --- a/core/src/main/scala/org/apache/spark/ui/JettyUtils.scala +++ b/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) } diff --git a/core/src/main/scala/org/apache/spark/util/AkkaUtils.scala b/core/src/main/scala/org/apache/spark/util/AkkaUtils.scala index db2531dc171f8..06828e9eea4b1 100644 --- a/core/src/main/scala/org/apache/spark/util/AkkaUtils.scala +++ b/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( diff --git a/core/src/main/scala/org/apache/spark/util/Utils.scala b/core/src/main/scala/org/apache/spark/util/Utils.scala index f6997fa4d0f4f..310eaa4c82fd4 100644 --- a/core/src/main/scala/org/apache/spark/util/Utils.scala +++ b/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