Fix Windows spark shell stdin issue

It turns out that java.lang.Process reads directly from the
parent process' stdin on Windows. This means we should avoid
spawning a thread that also attempts to redirect System.in to
the subprocess (in vain) and contends with the subprocess in
reading System.in.

This raises an issue with knowing when to terminate the JVM
in the PySpark shell, however, where Java itself is a python
subprocess. We previously relied on the Java process killing
itself on broken pipe, but this mechanism is not available on
Windows since we no longer read from System.in for the EOF.
Instead, in this environment we rely on python's shutdown
hook to kill the child process.

Author: andrewor14

core/src/main/scala/org/apache/spark/deploy/SparkSubmitDriverBootstrapper.scala

@@ -133,17 +133,24 @@ private[spark] object SparkSubmitDriverBootstrapper {
     val process = builder.start()
 
     // Redirect stdin, stdout, and stderr to/from the child JVM
-    val stdinThread = new RedirectThread(System.in, process.getOutputStream, "redirect stdin")
     val stdoutThread = new RedirectThread(process.getInputStream, System.out, "redirect stdout")
     val stderrThread = new RedirectThread(process.getErrorStream, System.err, "redirect stderr")
-    stdinThread.start()
     stdoutThread.start()
     stderrThread.start()
 
-    // Terminate on broken pipe, which signals that the parent process has exited. This is
-    // important for the PySpark shell, where Spark submit itself is a python subprocess.
-    stdinThread.join()
-    process.destroy()
+    // In Windows, the subprocess reads directly from our stdin, so we should avoid spawning
+    // a thread that also reads from stdin and contends with the subprocess.
+    if (Utils.isWindows) {
+      // For the PySpark shell, the termination of this process is handled in java_gateway.py
+      process.waitFor()
+    } else {
+      // Terminate on broken pipe, which signals that the parent process has exited. This is
+      // important for the PySpark shell, where Spark submit itself is a python subprocess.
+      val stdinThread = new RedirectThread(System.in, process.getOutputStream, "redirect stdin")
+      stdinThread.start()
+      stdinThread.join()
+      process.destroy()
+    }
   }
 
 }

python/pyspark/java_gateway.py

@@ -15,6 +15,7 @@
 # limitations under the License.
 #
 
+import atexit
 import os
 import sys
 import signal
@@ -69,6 +70,13 @@ def preexec_func():
                 error_msg += "--------------------------------------------------------------\n"
             raise Exception(error_msg)
 
+        # Ensure the Java subprocess does not linger after python exists. Note that this is best
+        # effort and intended mainly for Windows. In UNIX-based systems, the child process can kill
+        # itself on broken pipe (i.e. when the parent process' stdin sends an EOF). In Windows,
+        # however, this is not possible because java.lang.Process reads directly from the parent
+        # process' stdin, contending with any opportunity to read an EOF from the parent.
+        atexit.register(lambda: proc.kill())
+
         # Create a thread to echo output from the GatewayServer, which is required
         # for Java log output to show up:
         class EchoOutputThread(Thread):