Moved push of data out of ShuffleWriter and addressed other minor comments

Author: otterc

core/src/main/scala/org/apache/spark/executor/Executor.scala

@@ -46,7 +46,7 @@ import org.apache.spark.metrics.source.JVMCPUSource
 import org.apache.spark.resource.ResourceInformation
 import org.apache.spark.rpc.RpcTimeout
 import org.apache.spark.scheduler._
-import org.apache.spark.shuffle.{FetchFailedException, ShuffleWriter}
+import org.apache.spark.shuffle.{FetchFailedException, PushShuffleSupport}
 import org.apache.spark.storage.{StorageLevel, TaskResultBlockId}
 import org.apache.spark.util._
 import org.apache.spark.util.io.ChunkedByteBuffer
@@ -318,7 +318,7 @@ private[spark] class Executor(
         case NonFatal(e) =>
           logWarning("Unable to stop heartbeater", e)
       }
-      ShuffleWriter.stop
+      PushShuffleSupport.stop()
       threadPool.shutdown()
 
       // Notify plugins that executor is shutting down so they can terminate cleanly

core/src/main/scala/org/apache/spark/internal/config/package.scala

@@ -1941,16 +1941,16 @@ package object config {
 
   private[spark] val PUSH_SHUFFLE_ENABLED =
     ConfigBuilder("spark.shuffle.push.enabled")
-      .doc("Set to 'true' to enable push based shuffle")
+      .doc("Set to 'true' to enable push based shuffle.")
       .version("3.1.0")
       .booleanConf
       .createWithDefault(false)
 
   private[spark] val PUSH_SHUFFLE_NUM_PUSH_THREADS =
     ConfigBuilder("spark.shuffle.push.numPushThreads")
-      .doc("Specify the number of threads in the block pusher pool. These threads assist  " +
+      .doc("Specify the number of threads in the block pusher pool. These threads assist " +
         "in creating connections and pushing blocks to remote shuffle services when push based " +
-        "shuffle is enabled. By default, the threadpool size is equal to the number of cores")
+        "shuffle is enabled. By default, the threadpool size is equal to the number of cores.")
       .version("3.1.0")
       .intConf
       .createOptional