[SPARK-58021][CONNECT] Add local server pool acquisition engine

[ericm-db]

What changes were proposed in this pull request?

This is layer 4 of the local Connect pool stack: #57684 -> #57685 -> #57686 -> this PR. Until the lower layers merge, GitHub shows their cumulative diff; the new review unit here is commit f3a02502293.

This layer adds the acquisition engine:

• claim, refill, retry, and timeout handling;
• detached member attendants;
• ephemeral local Connect server boot and publication;
• per-fingerprint pool sizing;
• attendant supervision and cleanup;
• process-exit release registration; and
• internal --attend plus user-invoked --purge entry points.

SparkSession integration and JIT warmup remain in later PRs.

Why are the changes needed?

The storage and lifecycle layers define safe state transitions, but a pool also needs one owner for launching and supervising each server. Isolating that orchestration makes the attendant boundary, environment sanitization, retry behavior, and refill concurrency independently reviewable.

Does this PR introduce any user-facing change?

No. The acquisition engine is not selected by SparkSession in this layer.

How was this patch tested?

Expanded LocalConnectServerPoolUnitTests with acquisition coverage for fingerprint-aware refill, re-entrant acquisition of an existing claim, and the POSIX platform gate, in addition to all storage and lifecycle cases.

python -m unittest -v pyspark.sql.tests.connect.test_connect_local_server_pool.LocalConnectServerPoolUnitTests

All 17 tests passed. Ruff check, Ruff format check, and git diff --check also passed.

Was this patch authored or co-authored using generative AI tooling?

Generated-by: Claude Code (Fable 5) and OpenAI Codex (GPT-5)

[HyukjinKwon]

0 blocking, 0 non-blocking, 0 nits.
Correctly designed concurrent acquisition; the lock discipline avoids both deadlock and over-spawning. No issues found.

Verification

Verified: acquire mutates only under with self._directory: and deliberately releases the lock between polls so attendants can publish (holding it would deadlock the wait); refill counts and spawns entirely under the lock, so concurrent cold-starters share one complement of launches rather than each spawning target (no over-spawn); a dead launch is recounted only after janitor reaps it, and janitor runs first each pass; the loop is bounded by _ACQUIRE_TIMEOUT so a never-ready pool fails loudly. Build/test: python/run-tests --testnames 'pyspark.sql.tests.connect.test_connect_local_server_pool'. (Note: the contract-claim scanner zombied on infra and was skipped after a retry; the acquire/refill concurrency claims were verified manually against the code.)