          ) ++ kubernetesConf.environment).map { case (k, v) =>
            new EnvVarBuilder()
              .withName(k)
+

diff --git a/core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala b/core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala
index fa86da9ae9..ceb5dfbe6a 100644
--- a/core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala


     // Windows does not grant read permission by default to non-admin users
     // Add read permission to owner explicitly
diff --git a/python/pyspark/serializers.py b/python/pyspark/serializers.py
index 1b434d3931..286f773904 100644
--- a/python/pyspark/serializers.py
+++ b/python/pyspark/serializers.py
@@ -78,6 +78,7 @@ class SpecialLengths(object):
     END_OF_STREAM = -4
     NULL = -5
     START_ARROW_STREAM = -6
+    FINISHED = -7


 class Serializer(object):
diff --git a/python/pyspark/worker.py b/python/pyspark/worker.py
index 8ca4bb37e5..5b22911158 100644
--- a/python/pyspark/worker.py
+++ b/python/pyspark/worker.py
@@ -642,15 +642,32 @@ def main(infile, outfile):
     # check end of stream
     if read_int(infile) == SpecialLengths.END_OF_STREAM:
         write_int(SpecialLengths.END_OF_STREAM, outfile)
+        print(str(os.getpid()) + "worker's task exit with 0")
     else:
         # write a different value to tell JVM to not reuse this worker
         write_int(SpecialLengths.END_OF_DATA_SECTION, outfile)
-        sys.exit(-1)
+        #sys.exit(-1)
+        print(str(os.getpid()) + "worker's task exit with -1")
+    outfile.flush()


 if __name__ == '__main__':
     # Read information about how to connect back to the JVM from the environment.
     java_port = int(os.environ["PYTHON_WORKER_FACTORY_PORT"])
     auth_secret = os.environ["PYTHON_WORKER_FACTORY_SECRET"]
-    (sock_file, _) = local_connect_and_auth(java_port, auth_secret)
-    main(sock_file, sock_file)
+    #(sock_file, _) = local_connect_and_auth(java_port, auth_secret)
+    #main(sock_file, sock_file)
+    print("!Python worker's pid is " + str(os.getpid()) + "!!!!!!!!!!!!!")
+    (sock_file, sock) = local_connect_and_auth(java_port, auth_secret)
+    print("Python worker connected to " + str(java_port) + "!!!!!!!!!!!!!")
+    while True:
+        try:
+            data_port = read_int(sock_file)
+            print("Python new data port is " + str(data_port) + "!!!!!!!!!!!!!")
+            (data_sock_file, data_sock) = local_connect_and_auth(data_port, auth_secret)
+            main(data_sock_file, data_sock_file)
+            data_sock.close()
+            write_int(SpecialLengths.FINISHED, sock_file)
+            sock_file.flush()
+        except Exception as e:
+            print("Python waiting for connect. \n" + str(e))
diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Constants.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Constants.scala
index 543ca12594..9dfb3697f3 100644
--- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Constants.scala
