Add spark.ports.maxRetries

Author: andrewor14

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

@@ -22,7 +22,6 @@ import java.net.Socket
 
 import scala.collection.JavaConversions._
 import scala.collection.mutable
-import scala.concurrent.Await
 import scala.util.Properties
 
 import akka.actor._

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

@@ -1331,6 +1331,17 @@ private[spark] object Utils extends Logging {
       .map { case (k, v) => s"-D$k=$v" }
   }
 
+  /**
+   * Default number of retries in binding to a port.
+   */
+  val portMaxRetries: Int = {
+    // SparkEnv may be null during tests
+    Option(SparkEnv.get)
+      .flatMap(_.conf.getOption("spark.ports.maxRetries"))
+      .map(_.toInt)
+      .getOrElse(16)
+  }
+
   /**
    * Attempt to start a service on the given port, or fail after a number of attempts.
    * Each subsequent attempt uses 1 + the port used in the previous attempt.
@@ -1346,7 +1357,7 @@ private[spark] object Utils extends Logging {
       startPort: Int,
       startService: Int => (T, Int),
       serviceName: String = "",
-      maxRetries: Int = 3): (T, Int) = {
+      maxRetries: Int = portMaxRetries): (T, Int) = {
     val serviceString = if (serviceName.isEmpty) "" else s" '$serviceName'"
     for (offset <- 0 to maxRetries) {
       // Do not increment port if startPort is 0, which is treated as a special port

docs/configuration.md

@@ -567,6 +567,13 @@ Apart from these, the following properties are also available, and may be useful
     Port for the driver to listen on.
   </td>
 </tr>
+<tr>
+  <td><code>spark.port.maxRetries</code></td>
+  <td>16</td>
+  <td>
+    Maximum number of retries when binding to a port before giving up.
+  </td>
+</tr>
 <tr>
   <td><code>spark.akka.frameSize</code></td>
   <td>10</td>