[SPARK-20250][CORE] Improper OOM error when a task been killed while spilling data

Currently, when a task is calling spill() but it receives a killing request from driver (e.g., speculative task), the `TaskMemoryManager` will throw an `OOM` exception.  And we don't catch `Fatal` exception when a error caused by `Thread.interrupt`. So for `ClosedByInterruptException`, we should throw `RuntimeException` instead of `OutOfMemoryError`.

https://issues.apache.org/jira/browse/SPARK-20250?jql=project%20%3D%20SPARK

Existing unit tests.

Author: Xianyang Liu <[email protected]>

Closes #18090 from ConeyLiu/SPARK-20250.

(cherry picked from commit 731462a)
Signed-off-by: Wenchen Fan <[email protected]>

Author: ConeyLiu

core/src/main/java/org/apache/spark/memory/TaskMemoryManager.java

@@ -19,6 +19,7 @@
 
 import javax.annotation.concurrent.GuardedBy;
 import java.io.IOException;
+import java.nio.channels.ClosedByInterruptException;
 import java.util.Arrays;
 import java.util.BitSet;
 import java.util.HashSet;
@@ -156,6 +157,10 @@ public long acquireExecutionMemory(long required, MemoryConsumer consumer) {
                   break;
                 }
               }
+            } catch (ClosedByInterruptException e) {
+              // This called by user to kill a task (e.g: speculative task).
+              logger.error("error while calling spill() on " + c, e);
+              throw new RuntimeException(e.getMessage());
             } catch (IOException e) {
               logger.error("error while calling spill() on " + c, e);
               throw new OutOfMemoryError("error while calling spill() on " + c + " : "
@@ -174,6 +179,10 @@ public long acquireExecutionMemory(long required, MemoryConsumer consumer) {
               Utils.bytesToString(released), consumer);
             got += memoryManager.acquireExecutionMemory(required - got, taskAttemptId, mode);
           }
+        } catch (ClosedByInterruptException e) {
+          // This called by user to kill a task (e.g: speculative task).
+          logger.error("error while calling spill() on " + consumer, e);
+          throw new RuntimeException(e.getMessage());
         } catch (IOException e) {
           logger.error("error while calling spill() on " + consumer, e);
           throw new OutOfMemoryError("error while calling spill() on " + consumer + " : "