avoid overflow

Author: Davies Liu

core/src/main/java/org/apache/spark/shuffle/sort/ShuffleInMemorySorter.java

@@ -107,7 +107,7 @@ public void expandPointerArray(LongArray newArray) {
       array.getBaseOffset(),
       newArray.getBaseObject(),
       newArray.getBaseOffset(),
-      pos * 8
+      pos * 8L
     );
     consumer.freeArray(array);
     array = newArray;
@@ -179,8 +179,8 @@ public ShuffleSorterIterator getSortedIterator() {
     } else {
       MemoryBlock unused = new MemoryBlock(
         array.getBaseObject(),
-        array.getBaseOffset() + pos * 8,
-        (array.size() - pos) * 8);
+        array.getBaseOffset() + pos * 8L,
+        (array.size() - pos) * 8L);
       LongArray buffer = new LongArray(unused);
       Sorter<PackedRecordPointer, LongArray> sorter =
         new Sorter<>(new ShuffleSortDataFormat(buffer));

core/src/main/java/org/apache/spark/util/collection/unsafe/sort/UnsafeInMemorySorter.java

@@ -190,7 +190,7 @@ public void expandPointerArray(LongArray newArray) {
       array.getBaseOffset(),
       newArray.getBaseObject(),
       newArray.getBaseOffset(),
-      pos * 8);
+      pos * 8L);
     consumer.freeArray(array);
     array = newArray;
     capacity = calcCapacity();
@@ -289,8 +289,8 @@ public SortedIterator getSortedIterator() {
       } else {
         MemoryBlock unused = new MemoryBlock(
           array.getBaseObject(),
-          array.getBaseOffset() + pos * 8,
-          (array.size() - pos) * 8);
+          array.getBaseOffset() + pos * 8L,
+          (array.size() - pos) * 8L);
         LongArray buffer = new LongArray(unused);
         Sorter<RecordPointerAndKeyPrefix, LongArray> sorter =
           new Sorter<>(new UnsafeSortDataFormat(buffer));