[SPARK-2307] SparkUI - storage tab displays incorrect RDDs

The issue here is that the `StorageTab` listens for updates from the `StorageStatusListener`, but when a block is kicked out of the cache, `StorageStatusListener` removes it from its list. Thus, there is no way for the `StorageTab` to know whether a block has been dropped.

This issue was introduced in #1080, which was itself a bug fix. Here we revert that PR and offer a different fix for the original bug (SPARK-2144).

Author: Andrew Or <[email protected]>

Closes #1249 from andrewor14/storage-ui-fix and squashes the following commits:

af019ce [Andrew Or] Fix SPARK-2307

Author: andrewor14

core/src/main/scala/org/apache/spark/storage/StorageStatusListener.scala

@@ -37,11 +37,7 @@ class StorageStatusListener extends SparkListener {
     val filteredStatus = storageStatusList.find(_.blockManagerId.executorId == execId)
     filteredStatus.foreach { storageStatus =>
       updatedBlocks.foreach { case (blockId, updatedStatus) =>
-        if (updatedStatus.storageLevel == StorageLevel.NONE) {
-          storageStatus.blocks.remove(blockId)
-        } else {
-          storageStatus.blocks(blockId) = updatedStatus
-        }
+        storageStatus.blocks(blockId) = updatedStatus
       }
     }
   }

core/src/main/scala/org/apache/spark/ui/exec/ExecutorsPage.scala

@@ -21,6 +21,7 @@ import javax.servlet.http.HttpServletRequest
 
 import scala.xml.Node
 
+import org.apache.spark.storage.StorageLevel
 import org.apache.spark.ui.{WebUIPage, UIUtils}
 import org.apache.spark.util.Utils
 
@@ -107,7 +108,9 @@ private[ui] class ExecutorsPage(parent: ExecutorsTab) extends WebUIPage("") {
     val status = listener.storageStatusList(statusId)
     val execId = status.blockManagerId.executorId
     val hostPort = status.blockManagerId.hostPort
-    val rddBlocks = status.blocks.size
+    val rddBlocks = status.blocks.count { case (_, blockStatus) =>
+      blockStatus.storageLevel != StorageLevel.NONE
+    }
     val memUsed = status.memUsed
     val maxMem = status.maxMem
     val diskUsed = status.diskUsed