Use IP address on the executor side instead of hostname and random port for blockmanager

(cherry picked from commit e90b3e0)

Author: catalinii

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

@@ -18,7 +18,7 @@
 package org.apache.spark
 
 import java.io.File
-import java.net.Socket
+import java.net.{InetAddress, Socket}
 import java.util.Locale
 
 import scala.collection.JavaConverters._
@@ -207,11 +207,15 @@ object SparkEnv extends Logging {
       numCores: Int,
       ioEncryptionKey: Option[Array[Byte]],
       isLocal: Boolean): SparkEnv = {
+    var hostnameFinal = hostname
+    if (conf.getBoolean("spark.lyft.resolve", false)) {
+      hostnameFinal = InetAddress.getByName(hostname).getHostAddress
+    }
     val env = create(
       conf,
       executorId,
       bindAddress,
-      hostname,
+      hostnameFinal,
       None,
       isLocal,
       numCores,

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

@@ -2220,9 +2220,16 @@ private[spark] object Utils
    * privileged ports.
    */
   def userPort(base: Int, offset: Int): Int = {
-    (base + offset - 1024) % (65536 - 1024) + 1024
+      (base + offset - 1024) % (65536 - 1024) + 1024
   }
 
+  def randPort(base: Int, offset: Int, maxRand: Int): Int = {
+      val rand = new scala.util.Random
+      val r = rand.nextInt(maxRand)
+      (base + offset + r) % 65535
+  }
+
+
   /**
    * 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 (unless the port is 0).
@@ -2249,6 +2256,10 @@ private[spark] object Utils
       // Do not increment port if startPort is 0, which is treated as a special port
       val tryPort = if (startPort == 0) {
         startPort
+      } else if (offset > 0 && conf.getInt("spark.lyft.maxrand", 0) > 0)
+      {
+        logInfo(s"Using randPort")
+        randPort(startPort, offset, conf.getInt("spark.lyft.maxrand", 0))
       } else {
         userPort(startPort, offset)
       }