[SPARK-27468][Core][WEBUI] BlockUpdate replication event shouldn't overwrite storage level description in the UI

core/src/main/scala/org/apache/spark/status/AppStatusListener.scala

@@ -917,8 +917,24 @@ private[spark] class AppStatusListener(
     // Update the block entry in the RDD info, keeping track of the deltas above so that we
     // can update the executor information too.
     liveRDDs.get(block.rddId).foreach { rdd =>
+
       if (updatedStorageLevel.isDefined) {
-        rdd.setStorageLevel(updatedStorageLevel.get)
+        // Replicated block update events will have `storageLevel.replication=1`.
+        // To avoid overwriting the block replicated event in the store, we need to
+        // have a check for whether the event is block replication or not.
+        // Default value of  `storageInfo.replication = 1` and hence if
+        // `storeLevel.replication = 2`, the replicated events won't overwrite in the store.
+        val storageInfo = rdd.storageInfo
+        val isReplicatedBlockUpdateEvent = storageLevel.replication < storageInfo.replication &&
+          (storageInfo.useDisk == storageLevel.useDisk &&
+            storageInfo.useMemory == storageLevel.useMemory &&
+            storageInfo.deserialized == storageLevel.deserialized &&
+            storageInfo.useOffHeap == storageLevel.useOffHeap)
+
+        if (!isReplicatedBlockUpdateEvent) {
+          rdd.storageInfo = storageLevel
+          rdd.setStorageLevel(updatedStorageLevel.get)
+        }
       }
 
       val partition = rdd.partition(block.name)

core/src/main/scala/org/apache/spark/status/LiveEntity.scala

@@ -29,7 +29,7 @@ import org.apache.spark.JobExecutionStatus
 import org.apache.spark.executor.{ExecutorMetrics, TaskMetrics}
 import org.apache.spark.scheduler.{AccumulableInfo, StageInfo, TaskInfo}
 import org.apache.spark.status.api.v1
-import org.apache.spark.storage.RDDInfo
+import org.apache.spark.storage.{RDDInfo, StorageLevel}
 import org.apache.spark.ui.SparkUI
 import org.apache.spark.util.AccumulatorContext
 import org.apache.spark.util.collection.OpenHashSet
@@ -510,6 +510,7 @@ private class LiveRDD(val info: RDDInfo) extends LiveEntity {
   var storageLevel: String = weakIntern(info.storageLevel.description)
   var memoryUsed = 0L
   var diskUsed = 0L
+  var storageInfo: StorageLevel = info.storageLevel
 
   private val partitions = new HashMap[String, LiveRDDPartition]()
   private val partitionSeq = new RDDPartitionSeq()

core/src/test/scala/org/apache/spark/status/AppStatusListenerSuite.scala

@@ -1520,6 +1520,46 @@ class AppStatusListenerSuite extends SparkFunSuite with BeforeAndAfter {
     }
   }
 
+  test("storage description should display correct replication if storage replication is 2") {
+    val listener = new AppStatusListener(store, conf, true)
+    // Register a couple of block managers.
+    val bm1 = BlockManagerId("1", "host-1", 1234)
+    val bm2 = BlockManagerId("2", "host-2", 2345)
+
+    Seq(bm1, bm2).foreach { bm =>
+      listener.onExecutorAdded(SparkListenerExecutorAdded(1L, bm.executorId,
+        new ExecutorInfo(bm.host, 1, Map.empty, Map.empty)))
+      listener.onBlockManagerAdded(SparkListenerBlockManagerAdded(1L, bm, 42L))
+    }
+    val rddBlock = RddBlock(1, 1, 1L, 2L)
+
+    val level = StorageLevel.MEMORY_AND_DISK_2
+    // `replication` value  of the replicated block will be 1.
+    val levelBlockReplica = StorageLevel.MEMORY_AND_DISK
+
+    // Submit a stage and make sure the RDDs are recorded.
+    val rdd1Info = new RDDInfo(rddBlock.rddId, "rdd1", 2, level, Nil)
+    val stage = new StageInfo(1, 0, "stage1", 4, Seq(rdd1Info), Nil, "details1")
+    listener.onStageSubmitted(SparkListenerStageSubmitted(stage, new Properties()))
+
+    // Block update event, where replication = 2
+    listener.onBlockUpdated(SparkListenerBlockUpdated(
+      BlockUpdatedInfo(bm1, rddBlock.blockId, level, rddBlock.memSize, rddBlock.diskSize)))
+    // Block update event, where replication = 1
+    listener.onBlockUpdated(SparkListenerBlockUpdated(
+      BlockUpdatedInfo(bm2, rddBlock.blockId, levelBlockReplica, rddBlock.memSize,
+        rddBlock.diskSize)))
+
+    check[RDDStorageInfoWrapper](rddBlock.rddId) { wrapper =>
+      val partitionInfo = wrapper.info.partitions.get.find(_.blockName === rddBlock.blockId.name)
+        .get
+      assert(partitionInfo.storageLevel === level.description)
+      assert(partitionInfo.memoryUsed === 2 * rddBlock.memSize)
+      assert(partitionInfo.diskUsed === 2 * rddBlock.diskSize)
+      assert(partitionInfo.executors === Seq(bm1.executorId, bm2.executorId))
+    }
+  }
+
   test("storage information on executor lost/down") {
     val listener = new AppStatusListener(store, conf, true)
     val maxMemory = 42L