Fix yarn history URL, app links.

Author: Marcelo Vanzin

core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala

@@ -153,7 +153,7 @@ private[history] class FsHistoryProvider(conf: SparkConf) extends ApplicationHis
             val conf = this.conf.clone()
             val appSecManager = new SecurityManager(conf)
             SparkUI.createHistoryUI(conf, replayBus, appSecManager, appId,
-              s"${HistoryServer.UI_PATH_PREFIX}/$appId")
+              HistoryServer.getAttemptURI(appId, attempt.attemptId))
             // Do not call ui.bind() to avoid creating a new server for each application
           }
 

core/src/main/scala/org/apache/spark/deploy/history/HistoryPage.scala

@@ -154,17 +154,12 @@ private[history] class HistoryPage(parent: HistoryServer) extends WebUIPage("")
       <a href={makePageLink(nextPage, showIncomplete)}> {nextPage} </a>)
   }
 
-  private def getAttemptURI(appId: String, attemptInfo: ApplicationAttemptInfo): String = {
-    val attemptSuffix = if (!attemptInfo.attemptId.isEmpty) s"/${attemptInfo.attemptId}" else ""
-    s"${HistoryServer.UI_PATH_PREFIX}/${appId}${attemptSuffix}"
-  }
-
   private def attemptRow(
       renderAttemptIdColumn: Boolean,
       info: ApplicationHistoryInfo,
       attempt: ApplicationAttemptInfo,
       isFirst: Boolean): Seq[Node] = {
-    val uiAddress = getAttemptURI(info.id, attempt)
+    val uiAddress = HistoryServer.getAttemptURI(info.id, attempt.attemptId)
     val startTime = UIUtils.formatDate(attempt.startTime)
     val endTime = if (attempt.endTime > 0) UIUtils.formatDate(attempt.endTime) else "-"
     val duration =
@@ -191,7 +186,8 @@ private[history] class HistoryPage(parent: HistoryServer) extends WebUIPage("")
       {
         if (renderAttemptIdColumn) {
           if (info.attempts.size > 1 && !attempt.attemptId.isEmpty) {
-            <td><a href={getAttemptURI(info.id, attempt)}>{attempt.attemptId}</a></td>
+            <td><a href={HistoryServer.getAttemptURI(info.id, attempt.attemptId)}>
+              {attempt.attemptId}</a></td>
           } else {
             <td>&nbsp;</td>
           }

core/src/main/scala/org/apache/spark/deploy/history/HistoryServer.scala

@@ -224,4 +224,9 @@ object HistoryServer extends Logging {
     }
   }
 
+  private[history] def getAttemptURI(appId: String, attemptId: String): String = {
+    val attemptSuffix = if (!attemptId.isEmpty) s"/${attemptId}" else ""
+    s"${HistoryServer.UI_PATH_PREFIX}/${appId}${attemptSuffix}"
+  }
+
 }

yarn/src/main/scala/org/apache/spark/deploy/yarn/ApplicationMaster.scala

@@ -225,9 +225,10 @@ private[spark] class ApplicationMaster(
     val sc = sparkContextRef.get()
 
     val appId = client.getAttemptId().getApplicationId().toString()
+    val attemptId = client.getAttemptId().getAttemptId().toString()
     val historyAddress =
       sparkConf.getOption("spark.yarn.historyServer.address")
-        .map { address => s"${address}${HistoryServer.UI_PATH_PREFIX}/${appId}" }
+        .map { address => s"${address}${HistoryServer.UI_PATH_PREFIX}/${appId}/${attemptId}" }
         .getOrElse("")
 
     allocator = client.register(yarnConf,