Code review feedback

Author: pwendell

core/src/main/scala/org/apache/spark/shuffle/IndexShuffleBlockManager.scala

@@ -118,6 +118,7 @@ class IndexShuffleBlockManager(conf: SparkConf) extends ShuffleBlockResolver {
 private[spark] object IndexShuffleBlockManager {
   // No-op reduce ID used in interactions with disk store and BlockObjectWriter.
   // The disk store currently expects puts to relate to a (map, reduce) pair, but in the sort
-  // shuffle outputs from a map for several
+  // shuffle outputs for several reduces are glommed into a single file.
+  // TODO: Avoid this entirely by having the DiskBlockObjectWriter not require a BlockId.
   val NOOP_REDUCE_ID = 0
 }

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

@@ -442,7 +442,7 @@ private[spark] class BlockManager(
       val shuffleBlockManager = shuffleManager.shuffleBlockResolver
       // TODO: This should gracefully handle case where local block is not available. Currently
       // downstream code will throw an exception.
-      Some(shuffleBlockManager.getBlockData(blockId.asInstanceOf[ShuffleBlockId]).nioByteBuffer())
+      Option(shuffleBlockManager.getBlockData(blockId.asInstanceOf[ShuffleBlockId]).nioByteBuffer())
     } else {
       doGetLocal(blockId, asBlockResult = false).asInstanceOf[Option[ByteBuffer]]
     }