Merge branch 'ESPARK-100' into 'spark_2.1'

[ESPARK-100] 加上shuffle限流功能，单个task shuffle超过10G，就直接报错挂掉

加上shuffle限流功能，单个task shuffle超过10G，就直接报错挂掉  
resolve apache#100 

See merge request !70

Author: cenyuhai

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

@@ -72,6 +72,8 @@ final class ShuffleExternalSorter extends MemoryConsumer {
   private final BlockManager blockManager;
   private final TaskContext taskContext;
   private final ShuffleWriteMetrics writeMetrics;
+  private long totalSpillBytes = 0L;
+  private long shuffleSpillThreshold = 0L;
 
   /**
    * Force this sorter to spill when there are this many elements in memory. The default value is
@@ -117,6 +119,7 @@ final class ShuffleExternalSorter extends MemoryConsumer {
     this.numPartitions = numPartitions;
     // Use getSizeAsKb (not bytes) to maintain backwards compatibility if no units are provided
     this.fileBufferSizeBytes = (int) conf.getSizeAsKb("spark.shuffle.file.buffer", "32k") * 1024;
+    this.shuffleSpillThreshold = conf.getSizeAsBytes("spark.shuffle.spill.limit", "10g");
     this.numElementsForSpillThreshold =
       conf.getLong("spark.shuffle.spill.numElementsForceSpillThreshold", 1024 * 1024 * 1024);
     this.writeMetrics = writeMetrics;
@@ -245,6 +248,10 @@ public long spill(long size, MemoryConsumer trigger) throws IOException {
       return 0L;
     }
 
+    if (totalSpillBytes > shuffleSpillThreshold) {
+      throw new IOException("Shuffle spill exceed " + shuffleSpillThreshold);
+    }
+
     logger.info("Thread {} spilling sort data of {} to disk ({} {} so far)",
       Thread.currentThread().getId(),
       Utils.bytesToString(getMemoryUsage()),
@@ -258,6 +265,7 @@ public long spill(long size, MemoryConsumer trigger) throws IOException {
     // records. Otherwise, if the task is over allocated memory, then without freeing the memory
     // pages, we might not be able to get memory for the pointer array.
     taskContext.taskMetrics().incMemoryBytesSpilled(spillSize);
+    totalSpillBytes += spillSize;
     return spillSize;
   }
 

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

@@ -86,6 +86,7 @@ public final class UnsafeExternalSorter extends MemoryConsumer {
   private long totalSpillBytes = 0L;
   private long totalSortTimeNanos = 0L;
   private volatile SpillableIterator readingIterator = null;
+  private long shuffleSpillThreshold = 0L;
 
   public static UnsafeExternalSorter createWithExistingInMemorySorter(
       TaskMemoryManager taskMemoryManager,
@@ -150,6 +151,9 @@ private UnsafeExternalSorter(
     // TODO: Instead, separate spill metrics should be stored and reported (tracked in SPARK-3577).
     this.writeMetrics = new ShuffleWriteMetrics();
 
+    this.shuffleSpillThreshold = blockManager.conf()
+      .getSizeAsBytes("spark.shuffle.spill.limit", "10g");
+
     if (existingInMemorySorter == null) {
       this.inMemSorter = new UnsafeInMemorySorter(
         this, taskMemoryManager, recordComparator, prefixComparator, initialSize, canUseRadixSort);
@@ -188,6 +192,9 @@ public void closeCurrentPage() {
    */
   @Override
   public long spill(long size, MemoryConsumer trigger) throws IOException {
+    if (totalSpillBytes > shuffleSpillThreshold) {
+      throw new IOException("Shuffle spill exceed " + shuffleSpillThreshold);
+    }
     if (trigger != this) {
       if (readingIterator != null) {
         return readingIterator.spill();