Review Comments: Part 7

Minor changes to variable names and comments

Author: ankuriitg

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

@@ -584,7 +584,9 @@ final class ShuffleBlockFetcherIterator(
   }
 
   private[storage] def throwFetchFailedException(
-      blockId: BlockId, address: BlockManagerId, e: Throwable) = {
+      blockId: BlockId,
+      address: BlockManagerId,
+      e: Throwable) = {
     blockId match {
       case ShuffleBlockId(shufId, mapId, reduceId) =>
         throw new FetchFailedException(address, shufId.toInt, mapId.toInt, reduceId, e)
@@ -605,15 +607,15 @@ private class BufferReleasingInputStream(
     private val iterator: ShuffleBlockFetcherIterator,
     private val blockId: BlockId,
     private val address: BlockManagerId,
-    private val streamCompressedOrEncrypted: Boolean)
+    private val detectCorruption: Boolean)
   extends InputStream {
   private[this] var closed = false
 
   override def read(): Int = {
     try {
       delegate.read()
     } catch {
-      case e: IOException if streamCompressedOrEncrypted =>
+      case e: IOException if detectCorruption =>
         IOUtils.closeQuietly(this)
         iterator.throwFetchFailedException(blockId, address, e)
     }
@@ -635,7 +637,7 @@ private class BufferReleasingInputStream(
     try {
       delegate.skip(n)
     } catch {
-      case e: IOException if streamCompressedOrEncrypted =>
+      case e: IOException if detectCorruption =>
         IOUtils.closeQuietly(this)
         iterator.throwFetchFailedException(blockId, address, e)
     }
@@ -647,7 +649,7 @@ private class BufferReleasingInputStream(
     try {
       delegate.read(b)
     } catch {
-      case e: IOException if streamCompressedOrEncrypted =>
+      case e: IOException if detectCorruption =>
         IOUtils.closeQuietly(this)
         iterator.throwFetchFailedException(blockId, address, e)
     }
@@ -657,7 +659,7 @@ private class BufferReleasingInputStream(
     try {
       delegate.read(b, off, len)
     } catch {
-      case e: IOException if streamCompressedOrEncrypted =>
+      case e: IOException if detectCorruption =>
         IOUtils.closeQuietly(this)
         iterator.throwFetchFailedException(blockId, address, e)
     }

core/src/main/scala/org/apache/spark/util/Utils.scala

@@ -347,7 +347,7 @@ private[spark] object Utils extends Logging {
   def copyStreamUpTo(in: InputStream, maxSize: Long): (Boolean, InputStream) = {
     var count = 0L
     val out = new ChunkedByteBufferOutputStream(64 * 1024, ByteBuffer.allocate)
-    val streamCopied = tryWithSafeFinally {
+    val fullyCopied = tryWithSafeFinally {
       val bufSize = Math.min(8192L, maxSize)
       val buf = new Array[Byte](bufSize.toInt)
       var n = 0
@@ -368,10 +368,10 @@ private[spark] object Utils extends Logging {
         out.close()
       }
     }
-    if (streamCopied) {
-      (streamCopied, out.toChunkedByteBuffer.toInputStream(dispose = true))
+    if (fullyCopied) {
+      (fullyCopied, out.toChunkedByteBuffer.toInputStream(dispose = true))
     } else {
-      (streamCopied, new SequenceInputStream(
+      (fullyCopied, new SequenceInputStream(
         out.toChunkedByteBuffer.toInputStream(dispose = true), in))
     }
   }

core/src/test/scala/org/apache/spark/storage/ShuffleBlockFetcherIteratorSuite.scala

@@ -495,8 +495,8 @@ class ShuffleBlockFetcherIteratorSuite extends SparkFunSuite with PrivateMethodT
       true,
       taskContext.taskMetrics.createTempShuffleReadMetrics())
 
-    // Only one block should be returned which has corruption after maxBytesInFlight/3 because the
-    // other block will detect corruption on first fetch, and then get added to the queue again for
+    // We'll get back the block which has corruption after maxBytesInFlight/3 because the other
+    // block will detect corruption on first fetch, and then get added to the queue again for
     // a retry
     val (id, st) = iterator.next()
     assert(id === shuffleBlockId2)
@@ -507,7 +507,8 @@ class ShuffleBlockFetcherIteratorSuite extends SparkFunSuite with PrivateMethodT
     }
 
     // Following will succeed as it reads part of the stream which is not corrupt. This will read
-    // maxBytesInFlight/3 bytes from first stream and remaining from the second stream
+    // maxBytesInFlight/3 bytes from the portion copied into memory, and remaining from the
+    // underlying stream
     new DataInputStream(st).readFully(
       new Array[Byte](streamNotCorruptTill), 0, streamNotCorruptTill)
 

core/src/test/scala/org/apache/spark/util/UtilsSuite.scala

@@ -220,7 +220,7 @@ class UtilsSuite extends SparkFunSuite with ResetSystemProperties with Logging {
 
     val limit = 1000
     // testing for inputLength less than, equal to and greater than limit
-    List(998, 999, 1000, 1001, 1002).foreach { inputLength =>
+    (limit - 2 to limit + 2).foreach { inputLength =>
       val in = new ByteArrayInputStream(bytes.take(inputLength))
       val (fullyCopied: Boolean, mergedStream: InputStream) = Utils.copyStreamUpTo(in, limit)
       try {