fix

common/network-common/src/main/java/org/apache/spark/network/buffer/FileSegmentManagedBuffer.java

@@ -36,7 +36,7 @@
 /**
  * A {@link ManagedBuffer} backed by a segment in a file.
  */
-public class FileSegmentManagedBuffer extends ManagedBuffer {
+public final class FileSegmentManagedBuffer extends ManagedBuffer {
   private final TransportConf conf;
   private final File file;
   private final long offset;

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

@@ -65,7 +65,8 @@ final class ShuffleBlockFetcherIterator(
     maxBytesInFlight: Long,
     maxReqsInFlight: Int,
     maxReqSizeShuffleToMem: Long,
-    detectCorrupt: Boolean) extends Iterator[(BlockId, InputStream)] with Logging {
+    detectCorrupt: Boolean)
+  extends Iterator[(BlockId, InputStream)] with Logging {
 
   import ShuffleBlockFetcherIterator._
 
@@ -161,7 +162,7 @@ final class ShuffleBlockFetcherIterator(
     while (iter.hasNext) {
       val result = iter.next()
       result match {
-        case SuccessFetchResult(bId, address, _, buf, _) =>
+        case SuccessFetchResult(_, address, _, buf, _) =>
           if (address != blockManager.blockManagerId) {
             shuffleMetrics.incRemoteBytesRead(buf.size)
             shuffleMetrics.incRemoteBlocksFetched(1)

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

@@ -18,6 +18,7 @@
 package org.apache.spark.storage
 
 import java.io.{File, InputStream, IOException}
+import java.util.UUID
 import java.util.concurrent.Semaphore
 
 import scala.concurrent.ExecutionContext.Implicits.global
@@ -132,6 +133,7 @@ class ShuffleBlockFetcherIteratorSuite extends SparkFunSuite with PrivateMethodT
       verify(mockBuf, times(1)).release()
       verify(wrappedInputStream.invokePrivate(delegateAccess()), times(1)).close()
     }
+
     // 3 local blocks, and 2 remote blocks
     // (but from the same block manager so one call to fetchBlocks)
     verify(blockManager, times(3)).getBlockData(any())
@@ -418,7 +420,10 @@ class ShuffleBlockFetcherIteratorSuite extends SparkFunSuite with PrivateMethodT
     doReturn(localBmId).when(blockManager).blockManagerId
 
     val diskBlockManager = mock(classOf[DiskBlockManager])
-    doReturn(new File("shuffle-read-file")).when(diskBlockManager).getFile(any(classOf[String]))
+    doReturn{
+      var blockId = new TempLocalBlockId(UUID.randomUUID())
+      (blockId, new File(blockId.name))
+    }.when(diskBlockManager).createTempLocalBlock()
     doReturn(diskBlockManager).when(blockManager).diskBlockManager
 
     val remoteBmId = BlockManagerId("test-client-1", "test-client-1", 2)

docs/configuration.md

@@ -521,7 +521,7 @@ Apart from these, the following properties are also available, and may be useful
 </tr>
 <tr>
   <td><code>spark.reducer.maxReqSizeShuffleToMem</code></td>
-  <td>200 * 1024 * 1024</td>
+  <td>200m</td>
   <td>
     The blocks of a shuffle request will be fetched to disk when size of the request is above
     this threshold. This is to avoid a giant request takes too much memory.