This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch branch-4.3
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.3 by this push:
new 4b0de360a920 [SPARK-57787][CONNECT][FOLLOWUP] Harden persistent local
Connect server management
4b0de360a920 is described below
commit 4b0de360a920c937608014424dafa57537cbee63
Author: Wenchen Fan <[email protected]>
AuthorDate: Tue Aug 11 17:31:15 2026 +0800
[SPARK-57787][CONNECT][FOLLOWUP] Harden persistent local Connect server
management
Followup to https://github.com/apache/spark/pull/56907.
This change hardens persistent local Spark Connect server management in two
ways. Before sending
SIGTERM, the client verifies that the PID recorded in discovery still
belongs to
SparkConnectServer. If process inspection fails transiently, discovery is
preserved so shutdown
can be retried. The launcher also removes inherited binding-address
configuration and explicitly
binds the server to 127.0.0.1.
A stale discovery file can refer to a PID that the operating system has
reused for an unrelated
process. Signaling that process would be unsafe. Separately, an inherited
binding-address setting
can expose a server that is intended to remain local.
No. The persistent local server remains opt-in. The change only makes its
shutdown and network
binding behavior safer.
Added regression tests covering a reused PID, an inconclusive process
probe, and forced loopback
binding. Ran:
build/sbt -Phive package
python/run-tests --testnames
pyspark.sql.tests.connect.test_connect_local_server
All 16 tests passed.
Generated-by: Codex (GPT-5)
Closes #57854 from cloud-fan/SPARK-57787-followup.
Authored-by: Wenchen Fan <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
(cherry picked from commit 28a200d908c9681638c62539ae533e95605f0a44)
Signed-off-by: Wenchen Fan <[email protected]>
---
python/pyspark/sql/connect/local_server.py | 54 +++++++++++----
.../sql/tests/connect/test_connect_local_server.py | 79 +++++++++++++++++++++-
2 files changed, 120 insertions(+), 13 deletions(-)
diff --git a/python/pyspark/sql/connect/local_server.py
b/python/pyspark/sql/connect/local_server.py
index 4f06d884aff4..aed82f0c87d9 100644
--- a/python/pyspark/sql/connect/local_server.py
+++ b/python/pyspark/sql/connect/local_server.py
@@ -28,9 +28,9 @@ own server-side session, so session-local state does not leak
between runs.
The discovery file, the daemon's pid file, and the logs live in a per-user
``0700`` directory
under the system temp dir; ``SPARK_LOCAL_CONNECT_DISCOVERY`` overrides the
discovery file
-location. The auth token is stored with ``0600`` and the server binds
localhost, so other
-users on the machine can neither read the token nor reach the server.
Processes of the same
-user share the server by design.
+location. The auth token is stored with ``0600`` and the server always binds
IPv4 loopback,
+overriding any configured binding address, so other users on the machine can
neither read the
+token nor authenticate to the server. Processes of the same user share the
server by design.
The server runs until stopped with ``python -m
pyspark.sql.connect.local_server --stop``.
(A plain ``sbin/stop-connect-server.sh`` cannot find it: the daemon runs with
a custom pid
@@ -75,6 +75,24 @@ def _pid_alive(pid: int) -> bool:
return True
+def _is_local_connect_server(pid: int) -> Optional[bool]:
+ """Whether ``pid`` is still the managed Connect server recorded in
discovery.
+
+ Returns ``None`` when the process cannot be inspected, so callers do not
discard the
+ discovery information needed to retry later.
+ """
+ try:
+ result = subprocess.run(
+ ["ps", "-ww", "-p", str(pid), "-o", "command="],
+ capture_output=True,
+ text=True,
+ timeout=5,
+ )
+ e1xcept (OSError, subprocess.SubprocessError):
+ return None
+ return result.returncode == 0 and _SERVER_CLASS in result.stdout
+
+
class Discovery:
"""Reads and writes the discovery file recording the persistent local
server.
@@ -225,14 +243,18 @@ class LocalConnectServer:
os.environ["SPARK_CONNECT_AUTHENTICATE_TOKEN"] = self.token
return self.url
- def stop(self) -> bool:
+ def stop(self) -> Optional[bool]:
stopped = False
if self.pid is not None:
- try:
- os.kill(self.pid, signal.SIGTERM)
- stopped = True
- except OSError:
- pass
+ is_server = _is_local_connect_server(self.pid)
+ if is_server is None:
+ return None
+ if is_server:
+ try:
+ os.kill(self.pid, signal.SIGTERM)
+ stopped = True
+ except OSError:
+ pass
self._discovery.clear()
return stopped
@@ -363,6 +385,8 @@ class ServerLauncher:
"--master",
self._master,
"--conf",
+ "spark.connect.grpc.binding.address=127.0.0.1",
+ "--conf",
"spark.connect.grpc.binding.port={}".format(port),
]
if conf_file is not None:
@@ -466,9 +490,11 @@ def reuse_or_start_local_connect_server(master: str, opts:
Dict[str, Any]) -> st
return LocalConnectServer(discovery).reuse_or_start(master, opts)
-def stop_local_connect_server() -> bool:
+def stop_local_connect_server() -> Optional[bool]:
"""Stop the recorded persistent local Connect server, if any; safe to call
when none is
- running. Also available as ``python -m pyspark.sql.connect.local_server
--stop``.
+ running. Returns ``True`` when the server was signalled, ``False`` when no
matching server
+ was found, and ``None`` when the process could not be inspected. Also
available as
+ ``python -m pyspark.sql.connect.local_server --stop``.
"""
with Discovery() as discovery:
return LocalConnectServer(discovery).stop()
@@ -488,8 +514,12 @@ def main() -> None:
if not args.stop:
parser.print_help(sys.stderr)
sys.exit(2)
- if stop_local_connect_server():
+ stopped = stop_local_connect_server()
+ if stopped:
print("Stopped the persistent local Spark Connect server.")
+ elif stopped is None:
+ print("Could not verify the persistent local Spark Connect server; try
again later.")
+ sys.exit(1)
else:
print("No running persistent local Spark Connect server found.")
diff --git a/python/pyspark/sql/tests/connect/test_connect_local_server.py
b/python/pyspark/sql/tests/connect/test_connect_local_server.py
index eba26972ae8c..e091c2e6c29b 100644
--- a/python/pyspark/sql/tests/connect/test_connect_local_server.py
+++ b/python/pyspark/sql/tests/connect/test_connect_local_server.py
@@ -192,11 +192,77 @@ class LocalConnectServerReuseTests(unittest.TestCase):
discovery.save(
{k: getattr(server, k) for k in ("host", "port", "token",
"pid", "spark_version")}
)
- with mock.patch.object(os, "kill") as kill:
+ # Avoid inspecting or signaling a real process while exercising the
stop path.
+ ps_result = subprocess.CompletedProcess([], 0,
stdout=local_server._SERVER_CLASS)
+ with (
+ mock.patch.object(subprocess, "run", return_value=ps_result) as
run,
+ mock.patch.object(os, "kill") as kill,
+ ):
self.assertTrue(local_server.stop_local_connect_server())
+ run.assert_called_once_with(
+ ["ps", "-ww", "-p", "12345", "-o", "command="],
+ capture_output=True,
+ text=True,
+ timeout=5,
+ )
kill.assert_called_once_with(12345, signal.SIGTERM)
self.assertIsNone(self._discovered_server().pid)
+ def test_stop_does_not_signal_reused_pid(self) -> None:
+ from unittest import mock
+
+ with Discovery() as discovery:
+ server = self._server(pid=12345)
+ discovery.save(
+ {k: getattr(server, k) for k in ("host", "port", "token",
"pid", "spark_version")}
+ )
+ # Model a recycled pid without depending on host process state.
+ ps_result = subprocess.CompletedProcess([], 0, stdout="unrelated
process")
+ with (
+ mock.patch.object(subprocess, "run", return_value=ps_result),
+ mock.patch.object(os, "kill") as kill,
+ ):
+ self.assertFalse(local_server.stop_local_connect_server())
+ kill.assert_not_called()
+ self.assertIsNone(self._discovered_server().pid)
+
+ def test_stop_preserves_discovery_when_process_cannot_be_inspected(self)
-> None:
+ from unittest import mock
+
+ with Discovery() as discovery:
+ server = self._server(pid=12345)
+ discovery.save(
+ {k: getattr(server, k) for k in ("host", "port", "token",
"pid", "spark_version")}
+ )
+ with (
+ mock.patch.object(subprocess, "run",
side_effect=subprocess.TimeoutExpired("ps", 5)),
+ mock.patch.object(os, "kill") as kill,
+ ):
+ self.assertIsNone(local_server.stop_local_connect_server())
+ kill.assert_not_called()
+ self.assertEqual(self._discovered_server().pid, 12345)
+ with Discovery() as discovery:
+ discovery.clear()
+
+ def test_server_launcher_binds_to_loopback(self) -> None:
+ from unittest import mock
+
+ with Discovery() as discovery:
+ launcher = local_server.ServerLauncher("local[2]", {}, discovery)
+ # Capture the launcher argv without starting an external daemon.
+ with (
+ mock.patch.dict(os.environ, {"SPARK_HOME": self._tmpdir}),
+ mock.patch.object(os.path, "isfile", return_value=True) as
isfile,
+ mock.patch.object(
+ subprocess, "run",
return_value=subprocess.CompletedProcess([], 0)
+ ) as run,
+ ):
+ launcher._run_script(15002, "token", None)
+ isfile.assert_called_once_with(
+ os.path.join(self._tmpdir, "sbin", "start-connect-server.sh")
+ )
+ self.assertIn("spark.connect.grpc.binding.address=127.0.0.1",
run.call_args.args[0])
+
def test_stop_cli_reports_when_no_server(self) -> None:
result = subprocess.run(
[sys.executable, "-m", "pyspark.sql.connect.local_server",
"--stop"],
@@ -208,6 +274,17 @@ class LocalConnectServerReuseTests(unittest.TestCase):
self.assertEqual(result.returncode, 0, result.stderr)
self.assertIn("No running persistent local Spark Connect server",
result.stdout)
+ def test_stop_cli_fails_when_process_cannot_be_inspected(self) -> None:
+ from unittest import mock
+
+ with (
+ mock.patch.object(sys, "argv", ["local_server", "--stop"]),
+ mock.patch.object(local_server, "stop_local_connect_server",
return_value=None),
+ self.assertRaises(SystemExit) as raised,
+ ):
+ local_server.main()
+ self.assertEqual(raised.exception.code, 1)
+
def test_reuse_or_start_requires_posix(self) -> None:
from unittest import mock
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
