Cleanup in UnsafeShuffleSorterIterator.

Author: JoshRosen

core/src/main/java/org/apache/spark/shuffle/unsafe/UnsafeShuffleInMemorySorter.java

@@ -78,35 +78,36 @@ public void insertRecord(long recordPointer, int partitionId) {
     sortBufferInsertPosition++;
   }
 
-  public static abstract class UnsafeShuffleSorterIterator {
+  /**
+   * An iterator-like class that's used instead of Java's Iterator in order to facilitate inlining.
+   */
+  public static final class UnsafeShuffleSorterIterator {
 
+    private final long[] sortBuffer;
+    private final int numRecords;
     final PackedRecordPointer packedRecordPointer = new PackedRecordPointer();
+    private int position = 0;
 
-    public abstract boolean hasNext();
+    public UnsafeShuffleSorterIterator(int numRecords, long[] sortBuffer) {
+      this.numRecords = numRecords;
+      this.sortBuffer = sortBuffer;
+    }
 
-    public abstract void loadNext();
+    public boolean hasNext() {
+      return position < numRecords;
+    }
 
+    public void loadNext() {
+      packedRecordPointer.set(sortBuffer[position]);
+      position++;
+    }
   }
 
   /**
    * Return an iterator over record pointers in sorted order.
    */
   public UnsafeShuffleSorterIterator getSortedIterator() {
     sorter.sort(sortBuffer, 0, sortBufferInsertPosition, SORT_COMPARATOR);
-    return new UnsafeShuffleSorterIterator() {
-
-      private int position = 0;
-
-      @Override
-      public boolean hasNext() {
-        return position < sortBufferInsertPosition;
-      }
-
-      @Override
-      public void loadNext() {
-        packedRecordPointer.set(sortBuffer[position]);
-        position++;
-      }
-    };
+    return new UnsafeShuffleSorterIterator(sortBufferInsertPosition, sortBuffer);
   }
 }