specify range for port retry

Author: WangTaoTheTonic

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

@@ -26,7 +26,7 @@ import org.apache.spark.util.Utils
 private[spark] class HttpFileServer(
     conf: SparkConf,
     securityManager: SecurityManager,
-    requestedPort: Int = 0)
+    requestedPort: String = "0")
   extends Logging {
 
   var baseDir : File = null

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

@@ -46,12 +46,12 @@ private[spark] class HttpServer(
     conf: SparkConf,
     resourceBase: File,
     securityManager: SecurityManager,
-    requestedPort: Int = 0,
+    requestedPort: String = "0",
     serverName: String = "HTTP server")
   extends Logging {
 
   private var server: Server = null
-  private var port: Int = requestedPort
+  private var port: Int = 0
 
   def start() {
     if (server != null) {

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

@@ -182,7 +182,7 @@ object SparkEnv extends Logging {
     assert(conf.contains("spark.driver.host"), "spark.driver.host is not set on the driver!")
     assert(conf.contains("spark.driver.port"), "spark.driver.port is not set on the driver!")
     val hostname = conf.get("spark.driver.host")
-    val port = conf.get("spark.driver.port").toInt
+    val port = conf.get("spark.driver.port")
     create(
       conf,
       SparkContext.DRIVER_IDENTIFIER,
@@ -203,7 +203,7 @@ object SparkEnv extends Logging {
       conf: SparkConf,
       executorId: String,
       hostname: String,
-      port: Int,
+      port: String,
       numCores: Int,
       isLocal: Boolean): SparkEnv = {
     val env = create(
@@ -226,7 +226,7 @@ object SparkEnv extends Logging {
       conf: SparkConf,
       executorId: String,
       hostname: String,
-      port: Int,
+      port: String,
       isDriver: Boolean,
       isLocal: Boolean,
       listenerBus: LiveListenerBus = null,
@@ -342,7 +342,7 @@ object SparkEnv extends Logging {
 
     val httpFileServer =
       if (isDriver) {
-        val fileServerPort = conf.getInt("spark.fileserver.port", 0)
+        val fileServerPort = conf.get("spark.fileserver.port", "0")
         val server = new HttpFileServer(conf, securityManager, fileServerPort)
         server.initialize()
         conf.set("spark.fileserver.uri",  server.serverUri)

core/src/main/scala/org/apache/spark/broadcast/HttpBroadcast.scala

@@ -152,7 +152,7 @@ private[broadcast] object HttpBroadcast extends Logging {
 
   private def createServer(conf: SparkConf) {
     broadcastDir = Utils.createTempDir(Utils.getLocalDir(conf), "broadcast")
-    val broadcastPort = conf.getInt("spark.broadcast.port", 0)
+    val broadcastPort = conf.get("spark.broadcast.port", "0")
     server =
       new HttpServer(conf, broadcastDir, securityManager, broadcastPort, "HTTP broadcast server")
     server.start()

core/src/main/scala/org/apache/spark/deploy/Client.scala

@@ -160,7 +160,7 @@ object Client {
     Logger.getRootLogger.setLevel(driverArgs.logLevel)
 
     val (actorSystem, _) = AkkaUtils.createActorSystem(
-      "driverClient", Utils.localHostName(), 0, conf, new SecurityManager(conf))
+      "driverClient", Utils.localHostName(), "0", conf, new SecurityManager(conf))
 
     // Verify driverArgs.master is a valid url so that we can use it in ClientActor safely
     Master.toAkkaUrl(driverArgs.master, AkkaUtils.protocol(actorSystem))

core/src/main/scala/org/apache/spark/deploy/LocalSparkCluster.scala

@@ -51,14 +51,14 @@ class LocalSparkCluster(
     val _conf = conf.clone().setIfMissing("spark.master.rest.enabled", "false")
 
     /* Start the Master */
-    val (masterSystem, masterPort, _, _) = Master.startSystemAndActor(localHostname, 0, 0, _conf)
+    val (masterSystem, masterPort, _, _) = Master.startSystemAndActor(localHostname, "0", "0", _conf)
     masterActorSystems += masterSystem
     val masterUrl = "spark://" + Utils.localHostNameForURI() + ":" + masterPort
     val masters = Array(masterUrl)
 
     /* Start the Workers */
     for (workerNum <- 1 to numWorkers) {
-      val (workerSystem, _) = Worker.startSystemAndActor(localHostname, 0, 0, coresPerWorker,
+      val (workerSystem, _) = Worker.startSystemAndActor(localHostname, "0", "0", coresPerWorker,
         memoryPerWorker, masters, null, Some(workerNum), _conf)
       workerActorSystems += workerSystem
     }

core/src/main/scala/org/apache/spark/deploy/client/TestClient.scala

@@ -46,7 +46,7 @@ private[spark] object TestClient {
   def main(args: Array[String]) {
     val url = args(0)
     val conf = new SparkConf
-    val (actorSystem, _) = AkkaUtils.createActorSystem("spark", Utils.localHostName(), 0,
+    val (actorSystem, _) = AkkaUtils.createActorSystem("spark", Utils.localHostName(), "0",
       conf = conf, securityManager = new SecurityManager(conf))
     val desc = new ApplicationDescription("TestClient", Some(1), 512,
       Command("spark.deploy.client.TestExecutor", Seq(), Map(), Seq(), Seq(), Seq()), "ignored")

core/src/main/scala/org/apache/spark/deploy/history/HistoryServer.scala

@@ -44,7 +44,7 @@ class HistoryServer(
     conf: SparkConf,
     provider: ApplicationHistoryProvider,
     securityManager: SecurityManager,
-    port: Int)
+    port: String)
   extends WebUI(securityManager, port, conf) with Logging {
 
   // How many applications to retain
@@ -189,7 +189,7 @@ object HistoryServer extends Logging {
       .newInstance(conf)
       .asInstanceOf[ApplicationHistoryProvider]
 
-    val port = conf.getInt("spark.history.ui.port", 18080)
+    val port = conf.get("spark.history.ui.port", "18080")
 
     val server = new HistoryServer(conf, provider, securityManager, port)
     server.bind()

core/src/main/scala/org/apache/spark/deploy/master/Master.scala

@@ -52,7 +52,7 @@ import org.apache.spark.util.{ActorLogReceive, AkkaUtils, RpcUtils, SignalLogger
 private[master] class Master(
     host: String,
     port: Int,
-    webUiPort: Int,
+    webUiPort: String,
     val securityMgr: SecurityManager,
     val conf: SparkConf)
   extends Actor with ActorLogReceive with Logging with LeaderElectable {
@@ -129,7 +129,7 @@ private[master] class Master(
   private val restServerEnabled = conf.getBoolean("spark.master.rest.enabled", true)
   private val restServer =
     if (restServerEnabled) {
-      val port = conf.getInt("spark.master.rest.port", 6066)
+      val port = conf.get("spark.master.rest.port", "6066")
       Some(new StandaloneRestServer(host, port, self, masterUrl, conf))
     } else {
       None
@@ -923,8 +923,8 @@ private[deploy] object Master extends Logging {
    */
   def startSystemAndActor(
       host: String,
-      port: Int,
-      webUiPort: Int,
+      port: String,
+      webUiPort: String,
       conf: SparkConf): (ActorSystem, Int, Int, Option[Int]) = {
     val securityMgr = new SecurityManager(conf)
     val (actorSystem, boundPort) = AkkaUtils.createActorSystem(systemName, host, port, conf = conf,

core/src/main/scala/org/apache/spark/deploy/master/MasterArguments.scala

@@ -25,19 +25,19 @@ import org.apache.spark.util.{IntParam, Utils}
  */
 private[master] class MasterArguments(args: Array[String], conf: SparkConf) {
   var host = Utils.localHostName()
-  var port = 7077
-  var webUiPort = 8080
+  var port = "7077"
+  var webUiPort = "8080"
   var propertiesFile: String = null
 
   // Check for settings in environment variables
   if (System.getenv("SPARK_MASTER_HOST") != null) {
     host = System.getenv("SPARK_MASTER_HOST")
   }
   if (System.getenv("SPARK_MASTER_PORT") != null) {
-    port = System.getenv("SPARK_MASTER_PORT").toInt
+    port = System.getenv("SPARK_MASTER_PORT")
   }
   if (System.getenv("SPARK_MASTER_WEBUI_PORT") != null) {
-    webUiPort = System.getenv("SPARK_MASTER_WEBUI_PORT").toInt
+    webUiPort = System.getenv("SPARK_MASTER_WEBUI_PORT")
   }
 
   parse(args.toList)
@@ -46,7 +46,7 @@ private[master] class MasterArguments(args: Array[String], conf: SparkConf) {
   propertiesFile = Utils.loadDefaultSparkProperties(conf, propertiesFile)
 
   if (conf.contains("spark.master.ui.port")) {
-    webUiPort = conf.get("spark.master.ui.port").toInt
+    webUiPort = conf.get("spark.master.ui.port")
   }
 
   private def parse(args: List[String]): Unit = args match {
@@ -60,11 +60,11 @@ private[master] class MasterArguments(args: Array[String], conf: SparkConf) {
       host = value
       parse(tail)
 
-    case ("--port" | "-p") :: IntParam(value) :: tail =>
+    case ("--port" | "-p") :: value :: tail =>
       port = value
       parse(tail)
 
-    case "--webui-port" :: IntParam(value) :: tail =>
+    case "--webui-port" :: value :: tail =>
       webUiPort = value
       parse(tail)