KE-37170 fix broadcastManager NPE (apache#480)

Author: yabola

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

@@ -363,26 +363,28 @@ class BlockManagerMasterEndpoint(
     while (iterator.hasNext) {
       val blockId = iterator.next
       val locations = blockLocations.get(blockId)
-      locations -= blockManagerId
-      // De-register the block if none of the block managers have it. Otherwise, if pro-active
-      // replication is enabled, and a block is either an RDD or a test block (the latter is used
-      // for unit testing), we send a message to a randomly chosen executor location to replicate
-      // the given block. Note that we ignore other block types (such as broadcast/shuffle blocks
-      // etc.) as replication doesn't make much sense in that context.
-      if (locations.size == 0) {
-        blockLocations.remove(blockId)
-        logWarning(s"No more replicas available for $blockId !")
-      } else if (proactivelyReplicate && (blockId.isRDD || blockId.isInstanceOf[TestBlockId])) {
-        // As a heuristic, assume single executor failure to find out the number of replicas that
-        // existed before failure
-        val maxReplicas = locations.size + 1
-        val i = (new Random(blockId.hashCode)).nextInt(locations.size)
-        val blockLocations = locations.toSeq
-        val candidateBMId = blockLocations(i)
-        blockManagerInfo.get(candidateBMId).foreach { bm =>
-          val remainingLocations = locations.toSeq.filter(bm => bm != candidateBMId)
-          val replicateMsg = ReplicateBlock(blockId, remainingLocations, maxReplicas)
-          bm.storageEndpoint.ask[Boolean](replicateMsg)
+      if(locations != null) {
+        locations -= blockManagerId
+        // De-register the block if none of the block managers have it. Otherwise, if pro-active
+        // replication is enabled, and a block is either an RDD or a test block (the latter is used
+        // for unit testing), we send a message to a randomly chosen executor location to replicate
+        // the given block. Note that we ignore other block types (such as broadcast/shuffle blocks
+        // etc.) as replication doesn't make much sense in that context.
+        if (locations.size == 0) {
+          blockLocations.remove(blockId)
+          logWarning(s"No more replicas available for $blockId !")
+        } else if (proactivelyReplicate && (blockId.isRDD || blockId.isInstanceOf[TestBlockId])) {
+          // As a heuristic, assume single executor failure to find out the number of replicas that
+          // existed before failure
+          val maxReplicas = locations.size + 1
+          val i = (new Random(blockId.hashCode)).nextInt(locations.size)
+          val blockLocations = locations.toSeq
+          val candidateBMId = blockLocations(i)
+          blockManagerInfo.get(candidateBMId).foreach { bm =>
+            val remainingLocations = locations.toSeq.filter(bm => bm != candidateBMId)
+            val replicateMsg = ReplicateBlock(blockId, remainingLocations, maxReplicas)
+            bm.storageEndpoint.ask[Boolean](replicateMsg)
+          }
         }
       }
     }