Merge remote-tracking branch 'apache/master' into job-page

Conflicts:
	core/src/main/scala/org/apache/spark/ui/jobs/JobProgressListener.scala

core/src/main/scala/org/apache/spark/ui/jobs/JobProgressListener.scala

@@ -64,11 +64,10 @@ class JobProgressListener(conf: SparkConf) extends SparkListener with Logging {
   val stageIdToData = new HashMap[(StageId, StageAttemptId), StageUIData]
   val stageIdToInfo = new HashMap[StageId, StageInfo]
   val stageIdToActiveJobIds = new HashMap[StageId, HashSet[JobId]]
-
-  // Number of completed and failed stages, may not actually equal to completedStages.size and 
-  // failedStages.size respectively due to completedStage and failedStages only maintain the latest
-  // part of the stages, the earlier ones will be removed when there are too many stages for 
-  // memory sake.
+  val poolToActiveStages = HashMap[PoolName, HashMap[StageId, StageInfo]]()
+  // Total of completed and failed stages that have ever been run.  These may be greater than
+  // `completedStages.size` and `failedStages.size` if we have run more stages or jobs than
+  // JobProgressListener's retention limits.
   var numCompletedStages = 0
   var numFailedStages = 0
 
@@ -97,7 +96,8 @@ class JobProgressListener(conf: SparkConf) extends SparkListener with Logging {
     Map(
       "activeStages" -> activeStages.size,
       "activeJobs" -> activeJobs.size,
-      "poolToActiveStages" -> poolToActiveStages.values.map(_.size).sum
+      "poolToActiveStages" -> poolToActiveStages.values.map(_.size).sum,
+      "stageIdToActiveJobIds" -> stageIdToActiveJobIds.values.map(_.size).sum
     )
   }
 
@@ -216,7 +216,7 @@ class JobProgressListener(conf: SparkConf) extends SparkListener with Logging {
     } else {
       failedStages += stage
       numFailedStages += 1
-      trimIfNecessary(failedStages)
+      trimStagesIfNecessary(failedStages)
     }
 
     for (
@@ -233,19 +233,6 @@ class JobProgressListener(conf: SparkConf) extends SparkListener with Logging {
     }
   }
 
-  /** If stages is too large, remove and garbage collect old stages */
-  private def trimIfNecessary(stages: ListBuffer[StageInfo]) = synchronized {
-    if (stages.size > retainedStages) {
-      val toRemove = math.max(retainedStages / 10, 1)
-      stages.take(toRemove).foreach { s =>
-        stageIdToData.remove((s.stageId, s.attemptId))
-        stageIdToInfo.remove(s.stageId)
-        stageIdToActiveJobIds.remove(s.stageId)
-      }
-      stages.trimStart(toRemove)
-    }
-  }
-
   /** For FIFO, all stages are contained by "default" pool but "default" pool here is meaningless */
   override def onStageSubmitted(stageSubmitted: SparkListenerStageSubmitted) = synchronized {
     val stage = stageSubmitted.stageInfo

core/src/test/scala/org/apache/spark/ui/jobs/JobProgressListenerSuite.scala

@@ -43,7 +43,10 @@ class JobProgressListenerSuite extends FunSuite with LocalSparkContext with Matc
   }
 
   private def createJobStartEvent(jobId: Int, stageIds: Seq[Int]) = {
-    SparkListenerJobStart(jobId, stageIds)
+    val stageInfos = stageIds.map { stageId =>
+      new StageInfo(stageId, 0, stageId.toString, 0, null, "")
+    }
+    SparkListenerJobStart(jobId, stageInfos, stageIds)
   }
 
   private def createJobEndEvent(jobId: Int, failed: Boolean = false) = {