Add a few miscellaneous ports

Author: andrewor14

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

@@ -146,6 +146,7 @@ object Client {
     }
 
     val conf = new SparkConf()
+    val port = conf.getInt("spark.standalone.client.port", 0) // TODO: document this
     val driverArgs = new ClientArguments(args)
 
     if (!driverArgs.logLevel.isGreaterOrEqual(Level.WARN)) {
@@ -158,7 +159,7 @@ object Client {
     // TODO: See if we can initialize akka so return messages are sent back using the same TCP
     //       flow. Else, this (sadly) requires the DriverClient be routable from the Master.
     val (actorSystem, _) = AkkaUtils.createActorSystem(
-      "driverClient", Utils.localHostName(), 0, conf, new SecurityManager(conf))
+      "driverClient", Utils.localHostName(), port, conf, new SecurityManager(conf))
 
     actorSystem.actorOf(Props(classOf[ClientActor], driverArgs, conf))
 

core/src/main/scala/org/apache/spark/deploy/worker/DriverWrapper.scala

@@ -30,8 +30,9 @@ object DriverWrapper {
     args.toList match {
       case workerUrl :: mainClass :: extraArgs =>
         val conf = new SparkConf()
+        val watcherPort = conf.getInt("spark.worker.watcher.port", 0) // TODO: document this
         val (actorSystem, _) = AkkaUtils.createActorSystem("Driver",
-          Utils.localHostName(), 0, conf, new SecurityManager(conf))
+          Utils.localHostName(), watcherPort, conf, new SecurityManager(conf))
         actorSystem.actorOf(Props(classOf[WorkerWatcher], workerUrl), name = "workerWatcher")
 
         // Delegate to supplied main class

core/src/main/scala/org/apache/spark/executor/Executor.scala

@@ -84,7 +84,8 @@ private[spark] class Executor(
   // Initialize Spark environment (using system properties read above)
   private val env = {
     if (!isLocal) {
-      val _env = SparkEnv.create(conf, executorId, slaveHostname, 0,
+      val port = conf.getInt("spark.executor.env.port", 0) // TODO: document this
+      val _env = SparkEnv.create(conf, executorId, slaveHostname, port,
         isDriver = false, isLocal = false)
       SparkEnv.set(_env)
       _env.metricsSystem.registerSource(executorSource)