[SPARK-14713][TESTS] Fix the flaky test NettyBlockTransferServiceSuite

## What changes were proposed in this pull request?

When there are multiple tests running, "NettyBlockTransferServiceSuite.can bind to a specific port twice and the second increments" may fail.

E.g., assume there are 2 tests running. Here are the execution order to reproduce the test failure.

| Execution Order | Test 1 | Test 2 |
| ------------- | ------------- | ------------- |
| 1 | service0 binds to 17634 |  |
| 2 |  | service0 binds to 17635 (17634 is occupied) |
| 3 | service1 binds to 17636 |  |
| 4 | pass test |  |
| 5 | service0.close (release 17634) |  |
| 6 |  | service1 binds to 17634 |
| 7 |  | `service1.port should be (service0.port + 1)` fails (17634 != 17635 + 1) |

Here is an example in Jenkins: https://amplab.cs.berkeley.edu/jenkins/view/Spark%20QA%20Test/job/spark-master-test-maven-hadoop-2.2/786/testReport/junit/org.apache.spark.network.netty/NettyBlockTransferServiceSuite/can_bind_to_a_specific_port_twice_and_the_second_increments/

This PR makes two changes:

- Use a random port between 17634 and 27634 to reduce the possibility of port conflicts.
- Make `service1` use `service0.port` to bind to avoid the above race condition.

## How was this patch tested?

Jenkins unit tests.

Author: Shixiong Zhu <shixiong@databricks.com>

Closes #12477 from zsxwing/SPARK-14713.

1 files changed, 17 insertions, 8 deletions

diff --git a/core/src/test/scala/org/apache/spark/network/netty/NettyBlockTransferServiceSuite.scala b/core/src/test/scala/org/apache/spark/network/netty/NettyBlockTransferServiceSuite.scala
index f3c156e4f7..e7df7cb419 100644
--- a/core/src/test/scala/org/apache/spark/network/netty/NettyBlockTransferServiceSuite.scala
+++ b/core/src/test/scala/org/apache/spark/network/netty/NettyBlockTransferServiceSuite.scala
@@ -17,6 +17,8 @@
 
 package org.apache.spark.network.netty
 
+import scala.util.Random
+
 import org.mockito.Mockito.mock
 import org.scalatest._
 
@@ -59,19 +61,26 @@ class NettyBlockTransferServiceSuite
   }
 
   test("can bind to a specific port") {
-    val port = 17634
+    val port = 17634 + Random.nextInt(10000)
+    logInfo("random port for test: " + port)
     service0 = createService(port)
-    service0.port should be >= port
-    service0.port should be <= (port + 10) // avoid testing equality in case of simultaneous tests
+    verifyServicePort(expectedPort = port, actualPort = service0.port)
   }
 
   test("can bind to a specific port twice and the second increments") {
-    val port = 17634
+    val port = 17634 + Random.nextInt(10000)
+    logInfo("random port for test: " + port)
     service0 = createService(port)
-    service1 = createService(port)
-    service0.port should be >= port
-    service0.port should be <= (port + 10)
-    service1.port should be (service0.port + 1)
+    verifyServicePort(expectedPort = port, actualPort = service0.port)
+    service1 = createService(service0.port)
+    // `service0.port` is occupied, so `service1.port` should not be `service0.port`
+    verifyServicePort(expectedPort = service0.port + 1, actualPort = service1.port)
+  }
+
+  private def verifyServicePort(expectedPort: Int, actualPort: Int): Unit = {
+    actualPort should be >= expectedPort
+    // avoid testing equality in case of simultaneous tests
+    actualPort should be <= (expectedPort + 10)
   }
 
   private def createService(port: Int): NettyBlockTransferService = {