You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by dt...@apache.org on 2026/08/31 19:58:43 UTC
(spark) branch master updated: [SPARK-58021][CONNECT] Reap orphaned local pool members
This is an automated email from the ASF dual-hosted git repository.
dtenedor pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new ed95123c03b8 [SPARK-58021][CONNECT] Reap orphaned local pool members
ed95123c03b8 is described below
commit ed95123c03b8b4410a0f94ca1fa9a3a1db12ad62
Author: Eric Marnadi <er...@databricks.com>
AuthorDate: Mon Aug 31 12:58:29 2026 -0700
[SPARK-58021][CONNECT] Reap orphaned local pool members
### What changes were proposed in this pull request?
This is layer 5 of the ten-PR local Connect pool stack:
#57684 -> #57685 -> #57907 -> #57686 -> #58247 ->
#58367 -> #57687 -> #58248 -> #57102 -> #57688
This layer adds recovery after a pool member has been published or claimed:
- a janitor for ready, claimed, and retiring members;
- retirement of dead, unreachable, incompatible, idle, and malformed ready members;
- retirement of claims whose client or server process generation has disappeared;
- persisted claimant process identities so PID reuse cannot strand an orphaned claim;
- recovery of independently valid server handles from malformed records; and
- bounded, PID-reuse-safe completion of retirement.
It also addresses the two follow-ups from the review of #57686: successful SIGTERM delivery is
persisted so young retirement passes do not repeat the process inspection, and the daemon-pid
fallback now has coverage for removing state after that process exits.
Pending-launch, attendant, conf-file, and unreferenced member-directory recovery remain in
#58367.
### Why are the changes needed?
Pool members outlive individual Python call frames. A client can be killed without releasing its
claim, a ready server can die or become unreachable, and persisted process IDs can be reused.
Without a conservative janitor these cases can strand JVMs or make unusable members count toward
the pool indefinitely.
Separating post-publication recovery from launch recovery gives each process-ownership model its
own review unit: this layer authorizes server cleanup from persisted process generations, while the
next layer handles attendant commands and launch process groups.
### Does this PR introduce _any_ user-facing change?
No. The pool is not wired into SparkSession in this layer.
### How was this patch tested?
The focused pool suite's 50 tests passed:
```bash
PYTHONPATH=python:python/lib/pyspark.zip:python/lib/py4j-0.10.9.9-src.zip \
SPARK_TESTING=1 \
.venv/bin/python -m unittest -v \
pyspark.sql.tests.connect.test_connect_local_server_pool
```
The changed files also passed Python AST parsing, Ruff checking and formatting, targeted mypy,
`git diff --check`, and changed-file ASCII and 100-column checks.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Fable 5) and OpenAI Codex (GPT-5)
Closes #58247 from ericm-db/local-connect-pool-recovery.
Authored-by: Eric Marnadi <er...@databricks.com>
Signed-off-by: Daniel Tenedorio <da...@databricks.com>
---
python/pyspark/sql/connect/local_server_pool.py | 207 +++++++++++++--
.../connect/test_connect_local_server_pool.py | 286 ++++++++++++++++++++-
2 files changed, 454 insertions(+), 39 deletions(-)
diff --git a/python/pyspark/sql/connect/local_server_pool.py b/python/pyspark/sql/connect/local_server_pool.py
index 7f0a07dcf61c..5ca1bcf664c2 100644
--- a/python/pyspark/sql/connect/local_server_pool.py
+++ b/python/pyspark/sql/connect/local_server_pool.py
@@ -149,6 +149,7 @@ class RetiredState(_PoolStateRecord):
pid: int
process_start_id: str
retired: float
+ signalled: bool = False
@classmethod
def pid_from_data(cls, data: Optional[Dict[str, Any]]) -> Optional[int]:
@@ -166,6 +167,14 @@ class RetiredState(_PoolStateRecord):
"""Recover a valid retirement timestamp from a malformed state record."""
return cls._timestamp(data.get("retired")) if data is not None else None
+ @staticmethod
+ def signalled_from_data(data: Optional[Dict[str, Any]]) -> Optional[bool]:
+ """Recover SIGTERM delivery state, defaulting legacy records to unsignalled."""
+ if data is None:
+ return None
+ value = data.get("signalled", False)
+ return value if isinstance(value, bool) else None
+
@classmethod
def from_data(cls, data: Optional[Dict[str, Any]]) -> Optional["RetiredState"]:
if data is None:
@@ -173,15 +182,17 @@ class RetiredState(_PoolStateRecord):
pid = cls.pid_from_data(data)
process_start_id = cls.process_start_id_from_data(data)
retired = cls.retired_from_data(data)
- if pid is None or process_start_id is None or retired is None:
+ signalled = cls.signalled_from_data(data)
+ if pid is None or process_start_id is None or retired is None or signalled is None:
return None
- return cls(pid, process_start_id, retired)
+ return cls(pid, process_start_id, retired, signalled)
def as_data(self) -> Dict[str, Any]:
return {
"pid": self.pid,
"process_start_id": self.process_start_id,
"retired": self.retired,
+ "signalled": self.signalled,
}
@@ -228,6 +239,23 @@ class PoolMember(_PoolStateRecord):
except (KeyError, TypeError, ValueError, OverflowError):
return None
+ @classmethod
+ def pid_from_data(cls, data: Optional[Dict[str, Any]]) -> Optional[int]:
+ """Recover a valid server pid even when another member field is malformed."""
+ return cls._positive_pid(data.get("pid")) if data is not None else None
+
+ @staticmethod
+ def process_start_id_from_data(data: Optional[Dict[str, Any]]) -> Optional[str]:
+ """Recover a process generation identifier from a malformed member record."""
+ value = data.get("process_start_id") if data is not None else None
+ return value if isinstance(value, str) and value else None
+
+ @staticmethod
+ def client_process_start_id_from_data(data: Optional[Dict[str, Any]]) -> Optional[str]:
+ """Recover the claiming client's process generation from a claimed record."""
+ value = data.get("client_process_start_id") if data is not None else None
+ return value if isinstance(value, str) and value else None
+
@property
def url(self) -> str:
return f"sc://{self.host}:{self.port}"
@@ -486,13 +514,14 @@ class PoolDirectory:
class ServerPool:
- """Claims and retires members of one pool directory."""
+ """Claims, reaps, and retires members of one pool directory."""
- # A retired server still alive after the grace period is hard-killed. After the give-up
- # age, tracking is removed only once the process is gone, replaced, or successfully
- # signalled.
+ # A retired server still alive after the grace period is hard-killed. With a process handle,
+ # tracking is removed only once it is gone, replaced, or successfully signalled. PID-less
+ # malformed state uses the give-up age as its bounded recovery window.
_RETIRE_KILL_AFTER_SECONDS = 30
_RETIRE_GIVE_UP_AFTER_SECONDS = 600
+ _DEFAULT_IDLE_TIMEOUT_SECONDS = 1800
_PROCESS_INSPECTION_TIMEOUT_SECONDS = 5
_PROC_STAT_START_TIME_INDEX = 19
@@ -543,13 +572,25 @@ class ServerPool:
return f"ps:{started}" if result.returncode == 0 and started else None
@classmethod
- def _same_server_instance(cls, pid: int, process_start_id: str) -> Optional[bool]:
- """Whether ``pid`` is still the recorded Connect server process generation."""
+ def _same_process_generation(
+ cls, pid: Optional[int], process_start_id: Optional[str]
+ ) -> Optional[bool]:
+ """Whether ``pid`` is alive and still has its recorded process generation."""
+ if pid is None or not _pid_alive(pid):
+ return False
+ if process_start_id is None:
+ return None
current_start_id = cls._process_start_id(pid)
if current_start_id is None:
return None
- if current_start_id != process_start_id:
- return False
+ return current_start_id == process_start_id
+
+ @classmethod
+ def _same_server_instance(cls, pid: int, process_start_id: str) -> Optional[bool]:
+ """Whether ``pid`` is still the recorded Connect server process generation."""
+ same_generation = cls._same_process_generation(pid, process_start_id)
+ if same_generation is not True:
+ return same_generation
return _is_local_connect_server(pid)
@staticmethod
@@ -568,6 +609,18 @@ class ServerPool:
"""Signal only the recorded generation of the managed Connect server."""
return cls._same_server_instance(pid, process_start_id) is True and cls._signal(pid, sig)
+ @classmethod
+ def _idle_timeout(cls) -> int:
+ """Seconds an unclaimed member may sit before it is retired.
+
+ Zero or a negative value disables idle retirement. Read the environment on each pass so
+ every reaper uses the same source of truth.
+ """
+ try:
+ return int(os.environ["SPARK_LOCAL_CONNECT_POOL_IDLE_TIMEOUT"])
+ except (KeyError, ValueError):
+ return cls._DEFAULT_IDLE_TIMEOUT_SECONDS
+
def claim(self, fingerprint: str) -> Optional[PoolMember]:
"""Claim the oldest usable member with this fingerprint, or ``None``. The rename to
``claimed-<pid>-<uid>.json`` marks the member as owned by this process; the reaping
@@ -590,24 +643,75 @@ class ServerPool:
data = self._directory.read_json(path)
member = PoolMember.from_data(data) if data is not None else None
if member is not None and member.fingerprint == fingerprint:
- candidates.append((member, uid, path))
+ assert data is not None
+ candidates.append((member, uid, path, data))
candidates.sort(key=lambda c: c[0].created)
- for member, uid, path in candidates:
+ for member, uid, path, data in candidates:
if not member.is_usable():
continue # left for the reaping rules to retire
+ client_process_start_id = self._process_start_id(os.getpid())
+ if client_process_start_id is not None:
+ claimed_data = dict(data)
+ claimed_data["client_process_start_id"] = client_process_start_id
+ # Persist ownership before the atomic rename. If this process dies between the
+ # write and rename, the extra field is harmless on the still-unclaimed record.
+ self._directory.write_json(path, claimed_data)
claim_path = self._directory.claimed_path(os.getpid(), uid)
self._directory.rename(path, claim_path)
member.claim_path = claim_path
return member
return None
+ def janitor(self) -> None:
+ """Reap unusable or orphaned pool members. Every rule is idempotent, so successive
+ passes from any process are safe."""
+ for uid in self._directory.uids():
+ self.reap(uid)
+
def reap(self, uid: str) -> bool:
- """Advance a retiring member and report whether nothing of it remains."""
+ """Apply the reaping rules to one member; ``True`` when nothing of it remains."""
states = self._directory.states(uid)
- if "retired" in states:
+ had_retired = "retired" in states
+ if "server" in states:
+ self._reap_server(uid, states["server"])
+ states = self._directory.states(uid)
+ if "claimed" in states:
+ self._reap_claimed(uid, states["claimed"])
+ states = self._directory.states(uid)
+ if had_retired and "retired" in states:
self._reap_retired(uid, states["retired"])
return not self._directory.states(uid)
+ def _reap_server(self, uid: str, path: str) -> None:
+ """A ready member that is unusable (dead, unreachable, version-mismatched after an
+ upgrade, or an unreadable record) or has sat unclaimed past the idle timeout: retire
+ it."""
+ data = self._directory.read_json(path)
+ member = PoolMember.from_data(data) if data is not None else None
+ server_pid, process_start_id = self._recover_server_handle(uid, data)
+ idle = self._idle_timeout()
+ expired = member is not None and idle > 0 and time.time() - member.created > idle
+ if member is None or expired or not member.is_usable():
+ self._retire(path, server_pid, process_start_id)
+
+ def _reap_claimed(self, uid: str, path: str) -> None:
+ """A claimed member whose client died without releasing it (e.g. SIGKILL), or whose
+ server died under its client: retire it. Claims of this live process are its own."""
+ data = self._directory.read_json(path)
+ server_pid, process_start_id = self._recover_server_handle(uid, data)
+ client_pid = self._directory.claiming_pid(path)
+ client_process_start_id = PoolMember.client_process_start_id_from_data(data)
+ if client_process_start_id is None:
+ # Records written by older clients have no process generation. Preserve their
+ # liveness-only behavior rather than risking retirement of a live claim.
+ client_alive = client_pid == os.getpid() or _pid_alive(client_pid)
+ else:
+ client_alive = (
+ self._same_process_generation(client_pid, client_process_start_id) is not False
+ )
+ if not client_alive or self._same_process_generation(server_pid, process_start_id) is False:
+ self._retire(path, server_pid, process_start_id)
+
def _reap_retired(self, uid: str, path: str) -> None:
"""A retiring member: drop it once its server is gone, hard-kill the server if it
hangs in shutdown, and stop tracking a record whose process generation was reused."""
@@ -615,6 +719,7 @@ class ServerPool:
record_pid = RetiredState.pid_from_data(data)
record_process_start_id = RetiredState.process_start_id_from_data(data)
retired = RetiredState.retired_from_data(data)
+ signalled = RetiredState.signalled_from_data(data)
server_pid = self._recover_server_pid(uid, record_pid)
# A generation id is meaningful only with the pid from the same record. The daemon pid
# file has no companion generation id, so never pair a recovered daemon pid with
@@ -622,8 +727,18 @@ class ServerPool:
process_start_id = record_process_start_id if record_pid is not None else None
now = time.time()
if server_pid is None:
- # Without a recoverable process handle, retain the state for a later read instead of
- # declaring a potentially live server gone.
+ # A daemon pid may appear after a partial publication, so retain the state for one
+ # recovery window. After that there is no process handle left to act on or observe.
+ if retired is None or retired > now:
+ self._directory.write_json(
+ path,
+ {
+ "retired": now,
+ "signalled": False,
+ },
+ )
+ elif now - retired > self._RETIRE_GIVE_UP_AFTER_SECONDS:
+ self._remove_retired(uid, path)
return
if not _pid_alive(server_pid):
self._remove_retired(uid, path)
@@ -645,7 +760,13 @@ class ServerPool:
# A crash while _retire rewrites the atomically renamed state can leave its old
# payload. Restore a shutdown clock while preserving its process identity.
self._directory.write_json(
- path, RetiredState(server_pid, process_start_id, now).as_data()
+ path,
+ RetiredState(
+ server_pid,
+ process_start_id,
+ now,
+ signalled=signalled is True,
+ ).as_data(),
)
return
age = now - retired
@@ -662,28 +783,60 @@ class ServerPool:
self._remove_retired(uid, path)
elif is_server is True:
self._signal(server_pid, signal.SIGKILL)
- else:
- # Retrying SIGTERM is idempotent and recovers a transient inspection failure during
- # _retire before the shutdown deadline escalates to SIGKILL.
- self._signal_server(server_pid, process_start_id, signal.SIGTERM)
+ elif signalled is not True:
+ # Retry a SIGTERM that was not confirmed during retirement. Persist success without
+ # refreshing the retirement clock so repeated passes remain free and escalation is
+ # still measured from the original transition.
+ if self._signal_server(server_pid, process_start_id, signal.SIGTERM):
+ self._directory.write_json(
+ path,
+ RetiredState(
+ server_pid,
+ process_start_id,
+ retired,
+ signalled=True,
+ ).as_data(),
+ )
def _remove_retired(self, uid: str, path: str) -> None:
self._directory.remove(path)
self._directory.remove_member_dir(uid)
- def _retire(self, state_path: str, server_pid: int, process_start_id: str) -> None:
+ def _retire(
+ self,
+ state_path: str,
+ server_pid: Optional[int],
+ process_start_id: Optional[str],
+ ) -> None:
"""Move a member into the retired state: signal its server and track the shutdown so
:meth:`_reap_retired` can escalate if the JVM hangs."""
_, uid = self._directory.parse_entry(os.path.basename(state_path))
assert uid is not None
- self._signal_server(server_pid, process_start_id, signal.SIGTERM)
+ signalled = False
+ if server_pid is not None and process_start_id is not None:
+ signalled = self._signal_server(server_pid, process_start_id, signal.SIGTERM)
retired_path = self._directory.retired_path(uid)
# Rename instead of removing the old state so a crash cannot leave a live server with
# no state. If rewriting is interrupted, _reap_retired preserves the recoverable pid.
self._directory.rename(state_path, retired_path)
- self._directory.write_json(
- retired_path, RetiredState(server_pid, process_start_id, time.time()).as_data()
- )
+ retired_data: Dict[str, Any] = {
+ "retired": time.time(),
+ "signalled": signalled,
+ }
+ if server_pid is not None:
+ retired_data["pid"] = server_pid
+ if process_start_id is not None:
+ retired_data["process_start_id"] = process_start_id
+ self._directory.write_json(retired_path, retired_data)
+
+ def _recover_server_handle(
+ self, uid: str, data: Optional[Dict[str, Any]]
+ ) -> Tuple[Optional[int], Optional[str]]:
+ """Recover a pid and only the process identity paired with that pid's record."""
+ record_pid = PoolMember.pid_from_data(data)
+ if record_pid is None:
+ return self._recorded_daemon_pid(uid), None
+ return record_pid, PoolMember.process_start_id_from_data(data)
def _recover_server_pid(self, uid: str, record_pid: Optional[int]) -> Optional[int]:
"""Use a record pid when present, otherwise fall back to the daemon pid file.
@@ -701,7 +854,7 @@ class ServerPool:
def release(self, member: PoolMember) -> None:
"""Retire this process's claimed member; the shutdown completes in the background,
- ready for a later lifecycle pass to finish.
+ ready for a later janitor pass to finish.
This method acquires the pool-directory lock and must not be called while the same pool
directory is already locked, including through a different ``PoolDirectory`` instance.
diff --git a/python/pyspark/sql/tests/connect/test_connect_local_server_pool.py b/python/pyspark/sql/tests/connect/test_connect_local_server_pool.py
index 02c5c831c5f2..c0c120e867ab 100644
--- a/python/pyspark/sql/tests/connect/test_connect_local_server_pool.py
+++ b/python/pyspark/sql/tests/connect/test_connect_local_server_pool.py
@@ -636,15 +636,41 @@ class LocalConnectServerPoolUnitTests(unittest.TestCase):
self.assertEqual(retired.pid, 456)
self.assertEqual(retired.process_start_id, "process-1")
self.assertEqual(retired.retired, 2.0)
+ self.assertFalse(retired.signalled)
self.assertEqual(
retired.as_data(),
- {"pid": 456, "process_start_id": "process-1", "retired": 2.0},
+ {
+ "pid": 456,
+ "process_start_id": "process-1",
+ "retired": 2.0,
+ "signalled": False,
+ },
)
+ delivered = RetiredState.from_data(
+ {
+ "pid": 456,
+ "process_start_id": "process-1",
+ "retired": 2,
+ "signalled": True,
+ }
+ )
+ assert delivered is not None
+ self.assertTrue(delivered.signalled)
self.assertIsNone(
RetiredState.from_data(
{"pid": 456, "process_start_id": "process-1", "retired": "not-a-time"}
)
)
+ self.assertIsNone(
+ RetiredState.from_data(
+ {
+ "pid": 456,
+ "process_start_id": "process-1",
+ "retired": 2,
+ "signalled": 1,
+ }
+ )
+ )
self.assertIsNone(RetiredState.from_data({"pid": 456, "retired": 2}))
def test_claim_matches_fingerprint_and_renames(self) -> None:
@@ -666,9 +692,16 @@ class LocalConnectServerPoolUnitTests(unittest.TestCase):
self.assertEqual(os.path.basename(member.claim_path), claim_name)
states = self._states("bbb")
self.assertEqual(set(states), {"claimed"})
+ with self._directory as directory:
+ claimed = directory.read_json(states["claimed"])
+ assert claimed is not None
+ self.assertEqual(
+ claimed["client_process_start_id"], ServerPool._process_start_id(os.getpid())
+ )
# The mismatched member is untouched, and a second claim finds nothing.
self.assertEqual(set(self._states("aaa")), {"server"})
with self._directory:
+ self.assertFalse(self._pool.reap("bbb"))
self.assertIsNone(self._pool.claim("my-fp"))
def test_claim_prefers_the_oldest_member(self) -> None:
@@ -837,6 +870,189 @@ class LocalConnectServerPoolUnitTests(unittest.TestCase):
for uid in records:
self.assertEqual(set(self._states(uid)), {"server"})
+ def test_idle_timeout_configuration(self) -> None:
+ for value in ("0", "-1"):
+ with self.subTest(value=value):
+ os.environ["SPARK_LOCAL_CONNECT_POOL_IDLE_TIMEOUT"] = value
+ self.assertEqual(self._pool._idle_timeout(), int(value))
+
+ os.environ["SPARK_LOCAL_CONNECT_POOL_IDLE_TIMEOUT"] = "not-an-integer"
+ self.assertEqual(self._pool._idle_timeout(), ServerPool._DEFAULT_IDLE_TIMEOUT_SECONDS)
+
+ def test_reap_server_unreachable_and_idle(self) -> None:
+ with self.subTest("unreachable member is retired"):
+ gone = self._live_process()
+ with _non_listening_socket() as port:
+ self._write_state(
+ self._directory.server_path("dead"), self._server_data(port, gone.pid)
+ )
+ with self._directory:
+ self._pool.reap("dead")
+ self.assertEqual(set(self._states("dead")), {"retired"})
+ self.assertTrue(_wait_proc_dead(gone))
+ with self.subTest("member idle past the timeout is retired"):
+ os.environ["SPARK_LOCAL_CONNECT_POOL_IDLE_TIMEOUT"] = "10"
+ with _listening_socket() as port:
+ idle = self._live_process()
+ self._write_state(
+ self._directory.server_path("1d1e"),
+ self._server_data(port, idle.pid, created=time.time() - 60),
+ )
+ with self._directory:
+ self._pool.janitor()
+ self.assertEqual(set(self._states("1d1e")), {"retired"})
+ self.assertTrue(_wait_proc_dead(idle))
+
+ def test_reap_server_does_not_expire_when_idle_retirement_is_disabled(self) -> None:
+ with _listening_socket() as port:
+ for uid, value in (("d150", "0"), ("d151", "-1")):
+ with self.subTest(value=value):
+ os.environ["SPARK_LOCAL_CONNECT_POOL_IDLE_TIMEOUT"] = value
+ server = self._live_process()
+ self._write_state(
+ self._directory.server_path(uid),
+ self._server_data(port, server.pid, created=0),
+ )
+
+ with self._directory:
+ self.assertFalse(self._pool.reap(uid))
+
+ self.assertEqual(set(self._states(uid)), {"server"})
+ self.assertIsNone(server.poll())
+
+ def test_janitor_preserves_healthy_non_idle_server(self) -> None:
+ os.environ["SPARK_LOCAL_CONNECT_POOL_IDLE_TIMEOUT"] = "10"
+ server = self._live_process()
+ with _listening_socket() as port:
+ self._write_state(
+ self._directory.server_path("f2e5"), self._server_data(port, server.pid)
+ )
+
+ with self._directory:
+ self._pool.janitor()
+
+ self.assertEqual(set(self._states("f2e5")), {"server"})
+ self.assertIsNone(server.poll())
+
+ def test_reap_does_not_signal_reused_server_pid(self) -> None:
+ unrelated = subprocess.Popen(
+ [sys.executable, "-c", "import sys; sys.stdin.buffer.read()"],
+ stdin=subprocess.PIPE,
+ stdout=subprocess.DEVNULL,
+ stderr=subprocess.DEVNULL,
+ )
+ self._procs.append(unrelated)
+ with _non_listening_socket() as port:
+ self._write_state(
+ self._directory.server_path("bad6"), self._server_data(port, unrelated.pid)
+ )
+ with self._directory:
+ self._pool.reap("bad6")
+
+ self.assertEqual(set(self._states("bad6")), {"retired"})
+ self.assertIsNone(unrelated.poll())
+
+ def test_reap_malformed_member_recovers_server_pid(self) -> None:
+ # An out-of-range created value makes the full member invalid, but its independently
+ # valid pid must still be retired and tracked rather than leaving a live JVM orphaned.
+ invalid_created = self._live_process()
+ with _non_listening_socket() as port:
+ self._write_state(
+ self._directory.server_path("bad0"),
+ self._server_data(port, invalid_created.pid, created=2**100),
+ )
+ with self._directory:
+ self._pool.reap("bad0")
+ states = self._states("bad0")
+ self.assertEqual(set(states), {"retired"})
+ with self._directory as directory:
+ self.assertEqual(directory.read_json(states["retired"])["pid"], invalid_created.pid)
+ self.assertTrue(_wait_proc_dead(invalid_created))
+
+ # Claimed records use the same recovery when their client has disappeared.
+ claimed = self._live_process()
+ with _non_listening_socket() as port:
+ self._write_state(
+ self._directory.claimed_path(2**31 - 1, "bad2"),
+ self._server_data(port, claimed.pid, created=2**100),
+ )
+ with self._directory:
+ self._pool.reap("bad2")
+ states = self._states("bad2")
+ self.assertEqual(set(states), {"retired"})
+ with self._directory as directory:
+ self.assertEqual(directory.read_json(states["retired"])["pid"], claimed.pid)
+ self.assertTrue(_wait_proc_dead(claimed))
+
+ # If the record itself has no usable pid, fall back to spark-daemon.sh's pid file. The
+ # fallback keeps the shutdown tracked but lacks the process identity needed to signal it.
+ unreadable_pid = self._live_process()
+ self._write_daemon_pid("bad1", unreadable_pid.pid)
+ self._write_state(self._directory.server_path("bad1"), {"malformed": True})
+ with self._directory:
+ self._pool.reap("bad1")
+ states = self._states("bad1")
+ self.assertEqual(set(states), {"member", "retired"})
+ with self._directory as directory:
+ retired = directory.read_json(states["retired"])
+ assert retired is not None
+ self.assertEqual(retired["pid"], unreadable_pid.pid)
+ self.assertNotIn("process_start_id", retired)
+ self.assertIsNone(unreadable_pid.poll())
+
+ def test_reap_claimed_of_dead_client(self) -> None:
+ orphan = self._live_process()
+ dead_client = 2**31 - 1
+ ours = self._live_process()
+ with _non_listening_socket() as port:
+ self._write_state(
+ self._directory.claimed_path(dead_client, "0a0a"),
+ self._server_data(port, orphan.pid),
+ )
+ self._write_state(
+ self._directory.claimed_path(os.getpid(), "0b0b"),
+ self._server_data(port, ours.pid),
+ )
+ self._write_state(
+ self._directory.claimed_path(os.getpid(), "0c0c"),
+ self._server_data(port, 2**31 - 1),
+ )
+ with self._directory:
+ self._pool.janitor()
+ # The orphaned claim and our dead server are retired; our live claim is untouched.
+ self.assertEqual(set(self._states("0a0a")), {"retired"})
+ self.assertTrue(_wait_proc_dead(orphan), "the orphaned server was not stopped")
+ self.assertEqual(set(self._states("0b0b")), {"claimed"})
+ self.assertIsNone(ours.poll())
+ self.assertEqual(set(self._states("0c0c")), {"retired"})
+
+ def test_reap_claimed_detects_a_reused_client_pid(self) -> None:
+ server = self._live_process()
+ data = self._server_data(12345, server.pid)
+ data["client_process_start_id"] = "a-different-process-generation"
+ self._write_state(self._directory.claimed_path(os.getpid(), "bad5"), data)
+
+ with self._directory:
+ self.assertFalse(self._pool.reap("bad5"))
+
+ self.assertEqual(set(self._states("bad5")), {"retired"})
+ self.assertTrue(_wait_proc_dead(server), "the orphaned server was not stopped")
+
+ def test_reap_claimed_detects_a_reused_server_pid(self) -> None:
+ unrelated = self._stubborn_process()
+ data = self._server_data(
+ 12345,
+ unrelated.pid,
+ process_start_id="a-different-process-generation",
+ )
+ self._write_state(self._directory.claimed_path(os.getpid(), "bad6"), data)
+
+ with self._directory:
+ self.assertFalse(self._pool.reap("bad6"))
+
+ self.assertEqual(set(self._states("bad6")), {"retired"})
+ self.assertIsNone(unrelated.poll())
+
def test_reap_retired_escalates_to_sigkill(self) -> None:
with self.subTest("a fresh retirement is left to shut down gracefully"):
fresh = self._stubborn_process()
@@ -875,9 +1091,11 @@ class LocalConnectServerPoolUnitTests(unittest.TestCase):
def test_reap_repairs_malformed_retired_state(self) -> None:
server = self._stubborn_process()
+ malformed = self._retired_data(server.pid, "not-a-time")
+ malformed["signalled"] = True
path = self._write_state(
self._directory.retired_path("bad4"),
- self._retired_data(server.pid, "not-a-time"),
+ malformed,
)
with self._directory as directory:
@@ -887,6 +1105,7 @@ class LocalConnectServerPoolUnitTests(unittest.TestCase):
assert repaired is not None
self.assertEqual(repaired["pid"], server.pid)
self.assertIsInstance(repaired["retired"], float)
+ self.assertTrue(repaired["signalled"])
self.assertIsNone(server.poll())
def test_reap_keeps_retired_state_when_process_inspection_fails(self) -> None:
@@ -925,6 +1144,23 @@ class LocalConnectServerPoolUnitTests(unittest.TestCase):
self.assertEqual(retained, original)
self.assertIsNone(server.poll())
+ def test_reap_eventually_removes_state_without_a_process_handle(self) -> None:
+ uid = "fade"
+ self._write_state(self._directory.server_path(uid), {"malformed": True})
+
+ with self._directory as directory:
+ self.assertFalse(self._pool.reap(uid))
+ states = self._directory.states(uid)
+ self.assertEqual(set(states), {"retired"})
+ retired = directory.read_json(states["retired"])
+ assert retired is not None
+ self.assertNotIn("pid", retired)
+ retired["retired"] = time.time() - ServerPool._RETIRE_GIVE_UP_AFTER_SECONDS - 1
+ directory.write_json(states["retired"], retired)
+ self.assertTrue(self._pool.reap(uid))
+
+ self.assertFalse(self._states(uid))
+
def test_reap_keeps_a_daemon_pid_without_a_process_identity(self) -> None:
server = self._stubborn_process()
uid = "daed"
@@ -942,6 +1178,13 @@ class LocalConnectServerPoolUnitTests(unittest.TestCase):
self.assertEqual(retained, original)
self.assertIsNone(server.poll())
+ server.kill()
+ self.assertTrue(_wait_proc_dead(server))
+ with self._directory:
+ self.assertTrue(self._pool.reap(uid))
+ self.assertFalse(os.path.exists(path))
+ self.assertFalse(os.path.exists(self._directory.member_dir(uid)))
+
def test_reap_prefers_the_record_pid_and_its_process_identity(self) -> None:
recorded_server = self._stubborn_process()
daemon_server = self._stubborn_process()
@@ -1006,8 +1249,8 @@ class LocalConnectServerPoolUnitTests(unittest.TestCase):
self.assertIsInstance(repaired["retired"], float)
self.assertIsNone(server.poll())
- def test_reap_retries_a_sigterm_missed_during_retirement(self) -> None:
- server = self._live_process()
+ def test_reap_retries_sigterm_only_until_delivered(self) -> None:
+ server = self._stubborn_process()
process_start_id = ServerPool._process_start_id(server.pid)
assert process_start_id is not None
state_path = self._write_state(
@@ -1015,16 +1258,32 @@ class LocalConnectServerPoolUnitTests(unittest.TestCase):
self._server_data(12345, server.pid),
)
- with mock.patch.object(local_server_pool, "_is_local_connect_server", return_value=None):
- with self._directory:
+ with self._directory as directory:
+ with mock.patch.object(
+ ServerPool,
+ "_signal_server",
+ side_effect=[False, False, True],
+ ) as signal_server:
self._pool._retire(state_path, server.pid, process_start_id)
+ retired_path = self._directory.retired_path("fade")
+ initial = directory.read_json(retired_path)
+ assert initial is not None
+ self.assertFalse(initial["signalled"])
+
+ self.assertFalse(self._pool.reap("fade"))
+ failed_retry = directory.read_json(retired_path)
+ self.assertEqual(failed_retry, initial)
+
+ self.assertFalse(self._pool.reap("fade"))
+ delivered = directory.read_json(retired_path)
+ assert delivered is not None
+ self.assertTrue(delivered["signalled"])
+ self.assertEqual(delivered["retired"], initial["retired"])
+
+ self.assertFalse(self._pool.reap("fade"))
+ self.assertEqual(signal_server.call_count, 3)
self.assertIsNone(server.poll())
- with self._directory:
- self.assertFalse(self._pool.reap("fade"))
- self.assertTrue(_wait_proc_dead(server), "the reaper did not retry graceful shutdown")
- with self._directory:
- self.assertTrue(self._pool.reap("fade"))
def test_release_retires_the_claimed_member(self) -> None:
server = self._live_process()
@@ -1046,7 +1305,10 @@ class LocalConnectServerPoolUnitTests(unittest.TestCase):
states = self._states("a1a1")
self.assertEqual(set(states), {"retired"})
with self._directory as directory:
- self.assertEqual(directory.read_json(states["retired"])["pid"], server.pid)
+ retired = directory.read_json(states["retired"])
+ assert retired is not None
+ self.assertEqual(retired["pid"], server.pid)
+ self.assertTrue(retired["signalled"])
self.assertTrue(_wait_proc_dead(server), "release did not stop the server")
# Releasing again is a no-op.
local_server_pool.release_pooled_local_connect_server()
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org