ericm-db commented on code in PR #57684:
URL: https://github.com/apache/spark/pull/57684#discussion_r3693081529
##########
python/pyspark/sql/connect/local_server.py:
##########
@@ -303,20 +355,9 @@ def _seed_conf(self) -> Dict[str, Any]:
opt-in keys. Only the run that starts the server can seed static
confs; later runs
find the JVM already warm.
"""
- conf: Dict[str, Any] = {}
- for i in range(int(os.environ.get("PYSPARK_REMOTE_INIT_CONF_LEN",
"0"))):
- conf =
json.loads(os.environ["PYSPARK_REMOTE_INIT_CONF_{}".format(i)])
- conf.update(self._opts)
- for k in list(conf):
- if k in (
- "spark.remote",
- "spark.api.mode",
- "spark.master",
- "spark.connect.authenticate.token",
- "spark.connect.grpc.binding.port",
- ) or k.startswith("spark.local.connect."):
- conf.pop(k)
- return conf
+ if self._seed_override is not None:
+ return dict(self._seed_override)
Review Comment:
Yeah, better to keep the invariant in one place. Pulled the strip into
`_strip_launcher_conf` and run the override through it too. It's idempotent so
#57102's path is unchanged, but now raw opts passed as a seed can't leak
`spark.master`/port/token/`spark.local.connect.*` into `--properties-file`.
`startup_seed_conf` uses the same helper.
##########
python/pyspark/sql/connect/local_server.py:
##########
@@ -269,20 +321,20 @@ def _token(self) -> str:
)
def _pick_port(self) -> int:
- """Under SPARK_TESTING always use an OS-assigned free port so suites
can run in
- parallel; otherwise honor the configured/default port, falling back to
a free one if
- another process holds it. (A live stale server of ours also holds the
port, but that
- start fails later at spark-daemon.sh's pid-file check regardless of
port.) The sbin
- script cannot report an ephemeral port back, so the free port is
picked and released
- here, with a small race until the server binds it.
+ """Use an OS-assigned free port when requested or under SPARK_TESTING
so suites can
+ run in parallel. Otherwise honor the configured/default port, falling
back to a free
+ one if another process holds it. (A live stale server of ours also
holds the port, but
+ that start fails later at spark-daemon.sh's pid-file check regardless
of port.) The
+ sbin script cannot report an ephemeral port back, so the free port is
picked and
+ released here, with a small race until the server binds it.
"""
def free_port() -> int:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
sock.bind(("localhost", 0))
return sock.getsockname()[1]
- if "SPARK_TESTING" in os.environ:
+ if self._use_ephemeral_port or "SPARK_TESTING" in os.environ:
Review Comment:
Added a test that pops `SPARK_TESTING`, sets `use_ephemeral_port=True` with
a non-integer configured port, and asserts `_pick_port()` still returns a real
port - the configured branch would've raised on `int(...)`, so a clean return
proves it took the free-port path. Also added one for the
`use_ephemeral_port=False` case honoring the configured port.
##########
python/pyspark/sql/connect/local_server.py:
##########
@@ -303,20 +355,9 @@ def _seed_conf(self) -> Dict[str, Any]:
opt-in keys. Only the run that starts the server can seed static
confs; later runs
find the JVM already warm.
"""
Review Comment:
Updated it to cover both paths - the env+opts merge and the verbatim
override - and that either way the result goes through `_strip_launcher_conf`.
##########
python/pyspark/sql/tests/connect/test_connect_local_server.py:
##########
@@ -114,6 +114,65 @@ def test_discovery_location(self) -> None:
self.assertIn("spark-connect-{}".format(getpass.getuser()),
default.directory)
self.assertEqual(os.stat(default.directory).st_mode & 0o777, 0o700)
+ def test_startup_seed_conf(self) -> None:
+ from unittest import mock
+
+ initial = {
+ "spark.sql.shuffle.partitions": "8",
+ "spark.master": "local[1]",
+ }
+ opts = {
+ "spark.sql.warehouse.dir": os.path.join(self._tmpdir, "warehouse"),
+ "spark.local.connect.reuse": "true",
+ "spark.connect.grpc.binding.port": "0",
+ }
+ env = {
+ "PYSPARK_REMOTE_INIT_CONF_LEN": "1",
+ "PYSPARK_REMOTE_INIT_CONF_0": json.dumps(initial),
+ }
+ with mock.patch.dict(os.environ, env):
+ self.assertEqual(
+ local_server.startup_seed_conf(opts),
+ {
+ "spark.sql.shuffle.partitions": "8",
+ "spark.sql.warehouse.dir": opts["spark.sql.warehouse.dir"],
+ },
+ )
+
+ def test_start_delegates_launch_options(self) -> None:
Review Comment:
Added `ServerLauncher` tests for `_seed_conf`/`_seed_properties_file`:
override (used minus stripped keys), `{}` (stays empty, doesn't fall through to
`PYSPARK_REMOTE_INIT_CONF_*`), and `None` (env+opts merge). Empty seed yields
no properties file; non-empty writes a 0600 file with the confs.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
