Logging close() in case close() fails.

Author: rxin

core/src/main/scala/org/apache/spark/network/ManagedBuffer.scala

@@ -27,7 +27,7 @@ import scala.util.Try
 import com.google.common.io.ByteStreams
 import io.netty.buffer.{ByteBufInputStream, ByteBuf}
 
-import org.apache.spark.util.ByteBufferInputStream
+import org.apache.spark.util.{ByteBufferInputStream, Utils}
 
 
 /**
@@ -83,7 +83,7 @@ final class FileSegmentManagedBuffer(val file: File, val offset: Long, val lengt
         }
     } finally {
       if (channel != null) {
-        channel.close()
+        Utils.tryLog(channel.close())
       }
     }
   }
@@ -97,7 +97,7 @@ final class FileSegmentManagedBuffer(val file: File, val offset: Long, val lengt
     } catch {
       case e: IOException =>
         if (is != null) {
-          is.close()
+          Utils.tryLog(is.close())
         }
         Try(file.length).toOption match {
           case Some(fileLen) =>
@@ -107,7 +107,7 @@ final class FileSegmentManagedBuffer(val file: File, val offset: Long, val lengt
         }
       case e: Throwable =>
         if (is != null) {
-          is.close()
+          Utils.tryLog(is.close())
         }
         throw e
     }

core/src/main/scala/org/apache/spark/util/Utils.scala

@@ -1304,6 +1304,20 @@ private[spark] object Utils extends Logging {
     }
   }
 
+  /** Executes the given block in a Try, logging any uncaught exceptions. */
+  def tryLog[T](f: => T): Try[T] = {
+    try {
+      val res = f
+      scala.util.Success(res)
+    } catch {
+      case ct: ControlThrowable =>
+        throw ct
+      case t: Throwable =>
+        logError(s"Uncaught exception in thread ${Thread.currentThread().getName}", t)
+        scala.util.Failure(t)
+    }
+  }
+
   /** Returns true if the given exception was fatal. See docs for scala.util.control.NonFatal. */
   def isFatalError(e: Throwable): Boolean = {
     e match {