[SPARK-26175][PYTHON] Redirect the standard input of the forked child to devnull in daemon

python/pyspark/daemon.py

@@ -160,6 +160,15 @@ def handle_sigterm(*args):
                 if pid == 0:
                     # in child process
                     listen_sock.close()
+
+                    # close stdin for child process, see SPARK-26175
+                    # but we cannot just close it, because it make fileno "0" unallocated
+                    # and cause later new created file descriptor use fileno "0"
+                    # so here I close fd 0 and reopen it on "/dev/null"
+                    dummy_input = open(os.devnull, 'r')
+                    os.dup2(dummy_input.fileno(), 0)
+                    dummy_input.close()
+
                     try:
                         # Acknowledge that the fork was successful
                         outfile = sock.makefile(mode="wb")

python/pyspark/sql/tests/test_udf.py

@@ -616,6 +616,20 @@ def f():
 
         self.spark.range(1).select(f()).collect()
 
+    def test_worker_original_stdin_closed(self):
+        # Test the original stdin of worker inherit from daemon is closed
+        # and is replaced with '/dev/null'.
+        # See SPARK-26175
+        def task(iterator):
+            import sys
+            res = sys.stdin.read()
+            # Because the stdin is replaced with '/dev/null'
+            # Read data from it will get EOF
+            assert res == '', "Expect read EOF from stdin."
+            return iterator
+
+        self.sc.parallelize(range(1), 1).mapPartitions(task).count()
+
 
 class UDFInitializationTests(unittest.TestCase):
     def tearDown(self):