[SPARK-19307][pyspark] Make sure user conf is propagated to SparkContext.

The code was failing to propagate the user conf in the case where the
JVM was already initialized, which happens when a user submits a
python script via spark-submit.

Tested with new unit test and by running a python script in a real cluster.

Author: Marcelo Vanzin

python/pyspark/context.py

@@ -132,6 +132,9 @@ def _do_init(self, master, appName, sparkHome, pyFiles, environment, batchSize,
             self._conf = conf
         else:
             self._conf = SparkConf(_jvm=SparkContext._jvm)
+            if conf is not None:
+                for k, v in conf.getAll():
+                    self._conf.set(k, v)
 
         self._batchSize = batchSize  # -1 represents an unlimited batch size
         self._unbatched_serializer = serializer

python/pyspark/tests.py

@@ -2035,6 +2035,26 @@ def test_single_script_on_cluster(self):
         self.assertEqual(0, proc.returncode)
         self.assertIn("[2, 4, 6]", out.decode('utf-8'))
 
+    def test_user_configuration(self):
+      """Make sure user configuration is respected (SPARK-19307)"""
+      script = self.createTempFile("test.py", """
+          |from pyspark import SparkConf, SparkContext
+          |
+          |conf = SparkConf().set("spark.test_config", "1")
+          |sc = SparkContext(conf = conf)
+          |try:
+          |    if sc._conf.get("spark.test_config") != "1":
+          |        raise Exception("Cannot find spark.test_config in SparkContext's conf.")
+          |finally:
+          |    sc.stop()
+          """)
+      try:
+          subprocess.check_output([self.sparkSubmit, "--master", "local", script],
+              stderr=subprocess.STDOUT)
+      except subprocess.CalledProcessError as e:
+          # This gives us a better error message for debugging.
+          self.fail("Test exited with {0}, output:\n{1}".format(e.returncode, e.output))
+
 
 class ContextTests(unittest.TestCase):