[SPARK-19750][UI][BRANCH-2.1] Fix redirect issue from http to https

## What changes were proposed in this pull request?

If spark ui port (4040) is not set, it will choose port number 0, this will make https port to also choose 0. And in Spark 2.1 code, it will use this https port (0) to do redirect, so when redirect triggered, it will point to a wrong url:

like:

```
/tmp/temp$ wget http://172.27.25.134:55015
--2017-02-23 12:13:54--  http://172.27.25.134:55015/
Connecting to 172.27.25.134:55015... connected.
HTTP request sent, awaiting response... 302 Found
Location: https://172.27.25.134:0/ [following]
--2017-02-23 12:13:54--  https://172.27.25.134:0/
Connecting to 172.27.25.134:0... failed: Can't assign requested address.
Retrying.

--2017-02-23 12:13:55--  (try: 2)  https://172.27.25.134:0/
Connecting to 172.27.25.134:0... failed: Can't assign requested address.
Retrying.

--2017-02-23 12:13:57--  (try: 3)  https://172.27.25.134:0/
Connecting to 172.27.25.134:0... failed: Can't assign requested address.
Retrying.

--2017-02-23 12:14:00--  (try: 4)  https://172.27.25.134:0/
Connecting to 172.27.25.134:0... failed: Can't assign requested address.
Retrying.

```

So instead of using 0 to do redirect, we should pick a bound port instead.

This issue only exists in Spark 2.1-, and can be reproduced in yarn cluster mode.

## How was this patch tested?

Current redirect UT doesn't verify this issue, so extend current UT to do correct verification.

Author: jerryshao <sshao@hortonworks.com>

Closes #17083 from jerryshao/SPARK-19750.

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

@@ -27,6 +27,7 @@ import java.util.Arrays
 import java.util.concurrent.{CountDownLatch, TimeUnit}
 import java.util.jar.{JarEntry, JarOutputStream}
 import javax.net.ssl._
+import javax.servlet.http.HttpServletResponse
 import javax.tools.{JavaFileObject, SimpleJavaFileObject, ToolProvider}
 
 import scala.collection.JavaConverters._
@@ -186,12 +187,12 @@ private[spark] object TestUtils {
   }
 
   /**
-   * Returns the response code from an HTTP(S) URL.
+   * Returns the response code and url (if redirected) from an HTTP(S) URL.
    */
-  def httpResponseCode(
+  def httpResponseCodeAndURL(
       url: URL,
       method: String = "GET",
-      headers: Seq[(String, String)] = Nil): Int = {
+      headers: Seq[(String, String)] = Nil): (Int, Option[String]) = {
     val connection = url.openConnection().asInstanceOf[HttpURLConnection]
     connection.setRequestMethod(method)
     headers.foreach { case (k, v) => connection.setRequestProperty(k, v) }
@@ -210,16 +211,30 @@ private[spark] object TestUtils {
       sslCtx.init(null, Array(trustManager), new SecureRandom())
       connection.asInstanceOf[HttpsURLConnection].setSSLSocketFactory(sslCtx.getSocketFactory())
       connection.asInstanceOf[HttpsURLConnection].setHostnameVerifier(verifier)
+      connection.setInstanceFollowRedirects(false)
     }
 
     try {
       connection.connect()
-      connection.getResponseCode()
+      if (connection.getResponseCode == HttpServletResponse.SC_FOUND) {
+        (connection.getResponseCode, Option(connection.getHeaderField("Location")))
+      } else {
+        (connection.getResponseCode(), None)
+      }
     } finally {
       connection.disconnect()
     }
   }
 
+  /**
+   * Returns the response code from an HTTP(S) URL.
+   */
+  def httpResponseCode(
+      url: URL,
+      method: String = "GET",
+      headers: Seq[(String, String)] = Nil): Int = {
+    httpResponseCodeAndURL(url, method, headers)._1
+  }
 }
 
 

core/src/main/scala/org/apache/spark/ui/JettyUtils.scala

@@ -330,7 +330,7 @@ private[spark] object JettyUtils extends Logging {
 
           // redirect the HTTP requests to HTTPS port
           httpConnector.setName(REDIRECT_CONNECTOR_NAME)
-          collection.addHandler(createRedirectHttpsHandler(securePort, scheme))
+          collection.addHandler(createRedirectHttpsHandler(connector, scheme))
           Some(connector)
 
         case None =>
@@ -378,7 +378,9 @@ private[spark] object JettyUtils extends Logging {
       server.getHandler().asInstanceOf[ContextHandlerCollection])
   }
 
-  private def createRedirectHttpsHandler(securePort: Int, scheme: String): ContextHandler = {
+  private def createRedirectHttpsHandler(
+      httpsConnector: ServerConnector,
+      scheme: String): ContextHandler = {
     val redirectHandler: ContextHandler = new ContextHandler
     redirectHandler.setContextPath("/")
     redirectHandler.setVirtualHosts(Array("@" + REDIRECT_CONNECTOR_NAME))
@@ -391,8 +393,8 @@ private[spark] object JettyUtils extends Logging {
         if (baseRequest.isSecure) {
           return
         }
-        val httpsURI = createRedirectURI(scheme, baseRequest.getServerName, securePort,
-          baseRequest.getRequestURI, baseRequest.getQueryString)
+        val httpsURI = createRedirectURI(scheme, baseRequest.getServerName,
+          httpsConnector.getLocalPort, baseRequest.getRequestURI, baseRequest.getQueryString)
         response.setContentLength(0)
         response.encodeRedirectURL(httpsURI)
         response.sendRedirect(httpsURI)

core/src/test/scala/org/apache/spark/ui/UISuite.scala

@@ -267,8 +267,12 @@ class UISuite extends SparkFunSuite {
           s"$scheme://localhost:$port/test1/root",
           s"$scheme://localhost:$port/test2/root")
         urls.foreach { url =>
-          val rc = TestUtils.httpResponseCode(new URL(url))
+          val (rc, redirectUrl) = TestUtils.httpResponseCodeAndURL(new URL(url))
           assert(rc === expected, s"Unexpected status $rc for $url")
+          if (rc == HttpServletResponse.SC_FOUND) {
+            assert(
+              TestUtils.httpResponseCode(new URL(redirectUrl.get)) === HttpServletResponse.SC_OK)
+          }
         }
       }
     } finally {