[SPARK-2532] Consolidated shuffle fixes

core/src/main/scala/org/apache/spark/SparkEnv.scala

@@ -146,6 +146,9 @@ object SparkEnv extends Logging {
     // Listener bus is only used on the driver
     if (isDriver) {
       assert(listenerBus != null, "Attempted to create driver SparkEnv with null listener bus!")
+      // When running tests in local mode, previous shutdown of sc could have marked it as
+      // VM shutdown. recheck and disable shutdown flag.
+      // Utils.doShutdownCheck()
     }
 
     val securityManager = new SecurityManager(conf)

core/src/main/scala/org/apache/spark/serializer/JavaSerializer.scala

@@ -40,7 +40,7 @@ private[spark] class JavaSerializationStream(out: OutputStream, counterReset: In
    */
   def writeObject[T: ClassTag](t: T): SerializationStream = {
     objOut.writeObject(t)
-    if (counterReset > 0 && counter >= counterReset) {
+    if (counterReset >= 0 && counter >= counterReset) {
       objOut.reset()
       counter = 0
     } else {

core/src/main/scala/org/apache/spark/shuffle/hash/HashShuffleWriter.scala

@@ -61,7 +61,8 @@ class HashShuffleWriter[K, V](
   }
 
   /** Close this writer, passing along whether the map completed */
-  override def stop(success: Boolean): Option[MapStatus] = {
+  override def stop(successInput: Boolean): Option[MapStatus] = {
+    var success = successInput
     try {
       if (stopping) {
         return None
@@ -71,7 +72,8 @@ class HashShuffleWriter[K, V](
         try {
           return Some(commitWritesAndBuildStatus())
         } catch {
-          case e: Exception =>
+          case e: Throwable =>
+            success = false // for finally block
             revertWrites()
             throw e
         }
@@ -96,9 +98,9 @@ class HashShuffleWriter[K, V](
     var totalBytes = 0L
     var totalTime = 0L
     val compressedSizes = shuffle.writers.map { writer: BlockObjectWriter =>
-      writer.commit()
-      writer.close()
+      writer.commitAndClose()
       val size = writer.fileSegment().length
+      assert(size >= 0)
       totalBytes += size
       totalTime += writer.timeWriting()
       MapOutputTracker.compressSize(size)
@@ -116,8 +118,13 @@ class HashShuffleWriter[K, V](
   private def revertWrites(): Unit = {
     if (shuffle != null && shuffle.writers != null) {
       for (writer <- shuffle.writers) {
-        writer.revertPartialWrites()
-        writer.close()
+        try {
+          writer.revertPartialWritesAndClose()
+        } catch {
+          // Ensure that all revert's get done - log exception and continue
+          case ex: Exception =>
+            logError("Exception reverting/closing writers", ex)
+        }
       }
     }
   }