Unconditionally update master's block status when handling invalid request.

JoshRosen · JoshRosen · commit 6609c2a9595a · 2016-09-13T12:49:48.000-07:00
diff --git a/core/src/main/scala/org/apache/spark/storage/BlockManager.scala b/core/src/main/scala/org/apache/spark/storage/BlockManager.scala
@@ -280,7 +280,12 @@ private[spark] class BlockManager(
     } else {
       getLocalBytes(blockId) match {
         case Some(buffer) => new BlockManagerManagedBuffer(blockInfoManager, blockId, buffer)
-        case None => throw new BlockNotFoundException(blockId.toString)
+        case None =>
+          // If this block manager receives a request for a block that it doesn't have then it's
+          // likely that the master has outdated block statuses for this block. Therefore, we send
+          // an RPC so that this block is marked as being unavailable from this block manager.
+          reportBlockStatus(blockId, BlockStatus.empty)
+          throw new BlockNotFoundException(blockId.toString)
       }
     }
   }

Original file line number	Diff line number	Diff line change
`@@ -280,7 +280,12 @@ private[spark] class BlockManager(`
`280`	`280`	`} else {`
`281`	`281`	`getLocalBytes(blockId) match {`
`282`	`282`	`case Some(buffer) => new BlockManagerManagedBuffer(blockInfoManager, blockId, buffer)`
`283`		`- case None => throw new BlockNotFoundException(blockId.toString)`
	`283`	`+ case None =>`
	`284`	`+ // If this block manager receives a request for a block that it doesn't have then it's`
	`285`	`+ // likely that the master has outdated block statuses for this block. Therefore, we send`
	`286`	`+ // an RPC so that this block is marked as being unavailable from this block manager.`
	`287`	`+ reportBlockStatus(blockId, BlockStatus.empty)`
	`288`	`+ throw new BlockNotFoundException(blockId.toString)`
`284`	`289`	`}`
`285`	`290`	`}`
`286`	`291`	`}`