Revert some unwanted changes

Author: jerryshao

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

@@ -92,7 +92,7 @@ private[spark] class ShuffleMemoryManager(maxMemory: Long) extends Logging {
   }
 
   /** Release numBytes bytes for the current thread. */
-  def release(numBytes: Long): Unit =  synchronized {
+  def release(numBytes: Long): Unit = synchronized {
     val threadId = Thread.currentThread().getId
     val curMem = threadMemory.getOrElse(threadId, 0L)
     if (curMem < numBytes) {

core/src/main/scala/org/apache/spark/shuffle/hash/BlockStoreShuffleFetcher.scala

@@ -27,7 +27,7 @@ import org.apache.spark.shuffle.FetchFailedException
 import org.apache.spark.storage.{BlockId, BlockManagerId, ShuffleBlockFetcherIterator, ShuffleBlockId}
 import org.apache.spark.util.CompletionIterator
 
-private[shuffle] object BlockStoreShuffleFetcher extends Logging {
+private[hash] object BlockStoreShuffleFetcher extends Logging {
   def fetch[T](
       shuffleId: Int,
       reduceId: Int,

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

@@ -125,7 +125,7 @@ final class ShuffleRawBlockFetcherIterator(
    * Current [[FetchResult]] being processed. We track this so we can release the current buffer
    * in case of a runtime exception when processing the current buffer.
    */
-  private[spark] var currentResult: FetchResult = null
+  @volatile private[spark] var currentResult: FetchResult = null
 
   /**
    * Queue of fetch requests to issue; we'll pull requests off this gradually to make sure that