fix style

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

@@ -46,7 +46,7 @@ private[spark] class ShuffleMemoryManager(maxMemory: Long) extends Logging {
   /**
    * release other Spillable's memory of current thread until freeMemory >= requestedMemory
    */
-  private[spark] def releaseReservedMemory(toGrant: Long, requestMemory: Long): Long =
+  private[this] def releaseReservedMemory(toGrant: Long, requestMemory: Long): Long =
     synchronized {
     val threadId = Thread.currentThread().getId
     if (toGrant >= requestMemory || !threadReservedList.contains(threadId)){

core/src/main/scala/org/apache/spark/util/collection/CollectionSpillable.scala

@@ -42,16 +42,16 @@ private[spark] trait CollectionSpillable[C] extends Logging with Spillable{
   protected val shuffleMemoryManager = SparkEnv.get.shuffleMemoryManager
 
   // Threshold for this collection's size in bytes before we start tracking its memory usage
-  protected var myMemoryThreshold = 0L
+  private[this] var myMemoryThreshold = 0L
 
   // Number of elements read from input since last spill
-  protected var _elementsRead = 0L
+  private[this] var _elementsRead = 0L
 
   // Number of bytes spilled in total
-  protected var _memoryBytesSpilled = 0L
+  private[this] var _memoryBytesSpilled = 0L
 
   // Number of spills
-  protected var _spillCount = 0
+  private[this] var _spillCount = 0
 
   /**
    * Spills the current in-memory collection to disk if needed. Attempts to acquire more
@@ -110,4 +110,18 @@ private[spark] trait CollectionSpillable[C] extends Logging with Spillable{
       .format(threadId, org.apache.spark.util.Utils.bytesToString(size),
         _spillCount, if (_spillCount > 1) "s" else ""))
   }
+
+  /**
+   * log ForceSpill and return collection's size
+   */
+  protected def logForceSpill(currentMemory: Long): Long = {
+    _spillCount += 1
+    logSpillage(currentMemory)
+
+    _elementsRead = 0
+    _memoryBytesSpilled += currentMemory
+    val freeMemory = myMemoryThreshold
+    myMemoryThreshold = 0L
+    freeMemory
+  }
 }

core/src/main/scala/org/apache/spark/util/collection/ExternalAppendOnlyMap.scala

@@ -237,17 +237,9 @@ class ExternalAppendOnlyMap[K, V, C](
   override def forceSpill(): Long = {
     var freeMemory = 0L
     if (memoryOrDiskIter.isDefined) {
-      _spillCount += 1
-      logSpillage(currentMap.estimateSize())
-
+      freeMemory = logForceSpill(currentMap.estimateSize())
       memoryOrDiskIter.get.spill()
-
-      _elementsRead = 0
-      _memoryBytesSpilled += currentMap.estimateSize()
-      freeMemory = myMemoryThreshold
-      myMemoryThreshold = 0L
     }
-
     freeMemory
   }
 

core/src/main/scala/org/apache/spark/util/collection/ExternalSorter.scala

@@ -617,7 +617,9 @@ private[spark] class ExternalSorter[K, V, C](
    * Spill in-memory inMemory to a temporary file on disk.
    * Return on-disk iterator over a temporary file.
    */
-  private[this] def spillMemoryToDisk(iterator: Iterator[((Int, K), C)]): Iterator[((Int, K), C)] = {
+  private[this] def spillMemoryToDisk(iterator: Iterator[((Int, K), C)])
+    : Iterator[((Int, K), C)] = {
+
     val it = new WritablePartitionedIterator {
       private[this] var cur = if (iterator.hasNext) iterator.next() else null
 
@@ -643,7 +645,8 @@ private[spark] class ExternalSorter[K, V, C](
    * An iterator that read elements from in-memory iterator or disk iterator when in-memory
    * iterator have spilled to disk.
    */
-  case class MemoryOrDiskIterator(memIter: Iterator[((Int, K), C)]) extends Iterator[((Int, K), C)] {
+  case class MemoryOrDiskIterator(memIter: Iterator[((Int, K), C)])
+    extends Iterator[((Int, K), C)] {
 
     var currentIter = memIter
 
@@ -671,26 +674,13 @@ private[spark] class ExternalSorter[K, V, C](
   override def forceSpill(): Long = {
     var freeMemory = 0L
     if (memoryOrDiskIter.isDefined) {
-      // has memory buffer that can be spilled
-      _spillCount += 1
-
       val shouldCombine = aggregator.isDefined
       if (shouldCombine) {
-        logSpillage(map.estimateSize())
+        freeMemory = logForceSpill(map.estimateSize())
       } else {
-        logSpillage(buffer.estimateSize())
+        freeMemory = logForceSpill(buffer.estimateSize())
       }
-
       memoryOrDiskIter.get.spill()
-
-      _elementsRead = 0
-      if (shouldCombine) {
-        _memoryBytesSpilled += map.estimateSize()
-      } else {
-        _memoryBytesSpilled += buffer.estimateSize()
-      }
-      freeMemory = myMemoryThreshold
-      myMemoryThreshold = 0L
     }
     freeMemory
   }

core/src/test/scala/org/apache/spark/shuffle/ShuffleMemoryManagerSuite.scala

@@ -25,7 +25,7 @@ import java.util.concurrent.CountDownLatch
 import org.apache.spark.SparkFunSuite
 import org.apache.spark.Spillable
 
-class FakeSpillable extends Spillable {
+private[this] class FakeSpillable extends Spillable {
 
   var myMemoryThreshold: Long = 0L