Address comments

Author: mccheah

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

@@ -268,8 +268,7 @@ private long[] mergeSpills(SpillInfo[] spills) throws IOException {
               mapId,
               taskContext.taskAttemptId(),
               partitioner.numPartitions());
-      mapWriter.commitAllPartitions();
-      return new long[partitioner.numPartitions()];
+      return mapWriter.commitAllPartitions();
     } else if (spills.length == 1) {
       Optional<SingleSpillShuffleMapOutputWriter> maybeSingleFileWriter =
           shuffleExecutorComponents.createSingleFileMapOutputWriter(
@@ -379,7 +378,7 @@ private void mergeSpillsWithFileStream(
           inputBufferSizeInBytes);
       }
       for (int partition = 0; partition < numPartitions; partition++) {
-        boolean copyThrewExecption = true;
+        boolean copyThrewException = true;
         ShufflePartitionWriter writer = mapWriter.getPartitionWriter(partition);
         OutputStream partitionOutput = writer.openStream();
         try {
@@ -409,9 +408,9 @@ private void mergeSpillsWithFileStream(
               }
             }
           }
-          copyThrewExecption = false;
+          copyThrewException = false;
         } finally {
-          Closeables.close(partitionOutput, copyThrewExecption);
+          Closeables.close(partitionOutput, copyThrewException);
         }
         long numBytesWritten = writer.getNumBytesWritten();
         writeMetrics.incBytesWritten(numBytesWritten);
@@ -458,10 +457,10 @@ private void mergeSpillsWithTransferTo(
             final FileChannel spillInputChannel = spillInputChannels[i];
             final long writeStartTime = System.nanoTime();
             Utils.copyFileStreamNIO(
-                    spillInputChannel,
-                    resolvedChannel.channel(),
-                    spillInputChannelPositions[i],
-                    partitionLengthInSpill);
+                spillInputChannel,
+                resolvedChannel.channel(),
+                spillInputChannelPositions[i],
+                partitionLengthInSpill);
             copyThrewExecption = false;
             spillInputChannelPositions[i] += partitionLengthInSpill;
             writeMetrics.incWriteTime(System.nanoTime() - writeStartTime);

core/src/main/java/org/apache/spark/shuffle/sort/io/LocalDiskShuffleMapOutputWriter.java

@@ -102,8 +102,7 @@ public long[] commitAllPartitions() throws IOException {
     // exception if it is incorrect. The position will not be increased to the expected length
     // after calling transferTo in kernel version 2.6.32. This issue is described at
     // https://bugs.openjdk.java.net/browse/JDK-7052359 and SPARK-3948.
-    if (outputFileChannel != null
-        && outputFileChannel.position() != bytesWrittenToMergedFile) {
+    if (outputFileChannel != null && outputFileChannel.position() != bytesWrittenToMergedFile) {
       throw new IOException(
           "Current position " + outputFileChannel.position() + " does not equal expected " +
               "position " + bytesWrittenToMergedFile + " after transferTo. Please check your " +

core/src/test/java/org/apache/spark/shuffle/sort/UnsafeShuffleWriterSuite.java

@@ -137,7 +137,7 @@ public void setUp() throws IOException {
 
     when(shuffleBlockResolver.getDataFile(anyInt(), anyInt())).thenReturn(mergedOutputFile);
 
-    Answer renameTempAnswer = invocationOnMock -> {
+    Answer<?> renameTempAnswer = invocationOnMock -> {
       partitionSizesInMergedFile = (long[]) invocationOnMock.getArguments()[2];
       File tmp = (File) invocationOnMock.getArguments()[3];
       if (!mergedOutputFile.delete()) {