[SPARK-6955][NETWORK]Do not let Yarn Shuffle Server retry its server port.

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

@@ -39,7 +39,8 @@ class StandaloneWorkerShuffleService(sparkConf: SparkConf, securityManager: Secu
   private val port = sparkConf.getInt("spark.shuffle.service.port", 7337)
   private val useSasl: Boolean = securityManager.isAuthenticationEnabled()
 
-  private val transportConf = SparkTransportConf.fromSparkConf(sparkConf, numUsableCores = 0)
+  private val transportConf = SparkTransportConf.fromSparkConf(sparkConf, numUsableCores = 0,
+    disablePortRetry = true)
   private val blockHandler = new ExternalShuffleBlockHandler(transportConf)
   private val transportContext: TransportContext = {
     val handler = if (useSasl) new SaslRpcHandler(blockHandler, securityManager) else blockHandler

core/src/main/scala/org/apache/spark/network/netty/SparkTransportConf.scala

@@ -43,8 +43,12 @@ object SparkTransportConf {
    * @param numUsableCores if nonzero, this will restrict the server and client threads to only
    *                       use the given number of cores, rather than all of the machine's cores.
    *                       This restriction will only occur if these properties are not already set.
+   * @param disablePortRetry if true, server will not retry its port. It's better for the long-run
+   *                        server to disable it since the server and client had the agreement of
+   *                        the specific port.
    */
-  def fromSparkConf(_conf: SparkConf, numUsableCores: Int = 0): TransportConf = {
+  def fromSparkConf(_conf: SparkConf, numUsableCores: Int = 0,
+                    disablePortRetry: Boolean = false): TransportConf = {
     val conf = _conf.clone
 
     // Specify thread configuration based on our JVM's allocation of cores (rather than necessarily
@@ -55,6 +59,10 @@ object SparkTransportConf {
       conf.get("spark.shuffle.io.serverThreads", numThreads.toString))
     conf.set("spark.shuffle.io.clientThreads",
       conf.get("spark.shuffle.io.clientThreads", numThreads.toString))
+
+    if (disablePortRetry) {
+      conf.set("spark.port.maxRetries", "0")
+    }
 
     new TransportConf(new ConfigProvider {
       override def get(name: String): String = conf.get(name)

network/common/src/main/java/org/apache/spark/network/server/TransportServer.java

@@ -122,7 +122,8 @@ public void close() {
   }
 
   /**
-   * Attempt to bind to the specified port up to a fixed number of retries.
+   * Attempt to bind 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).
    * If all attempts fail after the max number of retries, exit.
    */
   private void bindRightPort(int portToBind) {

network/yarn/src/main/java/org/apache/spark/network/yarn/YarnShuffleService.java

@@ -98,18 +98,26 @@ private boolean isAuthenticationEnabled() {
    */
   @Override
   protected void serviceInit(Configuration conf) {
-    TransportConf transportConf = new TransportConf(new HadoopConfigProvider(conf));
+    Configuration newConf = new Configuration(conf);
+
+    // It's better to let the NodeManager get down rather than take a port retry
+    // when `spark.shuffle.service.port` has been conflicted during starting
+    // the Spark Yarn Shuffle Server, because the retry mechanism will make the
+    // inconsistency of shuffle port and also make client fail to find the port.
+    newConf.setInt("spark.port.maxRetries", 0);
+
+    TransportConf transportConf = new TransportConf(new HadoopConfigProvider(newConf));
     // If authentication is enabled, set up the shuffle server to use a
     // special RPC handler that filters out unauthenticated fetch requests
-    boolean authEnabled = conf.getBoolean(SPARK_AUTHENTICATE_KEY, DEFAULT_SPARK_AUTHENTICATE);
+    boolean authEnabled = newConf.getBoolean(SPARK_AUTHENTICATE_KEY, DEFAULT_SPARK_AUTHENTICATE);
     blockHandler = new ExternalShuffleBlockHandler(transportConf);
     RpcHandler rpcHandler = blockHandler;
     if (authEnabled) {
       secretManager = new ShuffleSecretManager();
       rpcHandler = new SaslRpcHandler(rpcHandler, secretManager);
     }
 
-    int port = conf.getInt(
+    int port = newConf.getInt(
       SPARK_SHUFFLE_SERVICE_PORT_KEY, DEFAULT_SPARK_SHUFFLE_SERVICE_PORT);
     TransportContext transportContext = new TransportContext(transportConf, rpcHandler);
     shuffleServer = transportContext.createServer(port);