Added SparkConf to the ctor argument of IndexShuffleBlockManager.

Author: rxin

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

@@ -22,7 +22,7 @@ import java.nio.ByteBuffer
 
 import com.google.common.io.ByteStreams
 
-import org.apache.spark.SparkEnv
+import org.apache.spark.{SparkConf, SparkEnv}
 import org.apache.spark.network.buffer.{FileSegmentManagedBuffer, ManagedBuffer}
 import org.apache.spark.network.netty.SparkTransportConf
 import org.apache.spark.storage._
@@ -39,11 +39,11 @@ import org.apache.spark.storage._
 // Note: Changes to the format in this file should be kept in sync with
 // org.apache.spark.network.shuffle.StandaloneShuffleBlockManager#getSortBasedShuffleBlockData().
 private[spark]
-class IndexShuffleBlockManager extends ShuffleBlockManager {
+class IndexShuffleBlockManager(conf: SparkConf) extends ShuffleBlockManager {
 
   private lazy val blockManager = SparkEnv.get.blockManager
 
-  private val transportConf = SparkTransportConf.fromSparkConf(SparkEnv.get.conf)
+  private val transportConf = SparkTransportConf.fromSparkConf(conf)
 
   /**
    * Mapping to a single shuffleBlockId with reduce ID 0.

core/src/main/scala/org/apache/spark/shuffle/sort/SortShuffleManager.scala

@@ -25,7 +25,7 @@ import org.apache.spark.shuffle.hash.HashShuffleReader
 
 private[spark] class SortShuffleManager(conf: SparkConf) extends ShuffleManager {
 
-  private val indexShuffleBlockManager = new IndexShuffleBlockManager()
+  private val indexShuffleBlockManager = new IndexShuffleBlockManager(conf)
   private val shuffleMapNumber = new ConcurrentHashMap[Int, Int]()
 
   /**