[dborowitz]

Note

Stacked on #216 (dborowitz-resilience), #211 (dborowitz-optimization), and #210 (dborowitz-timing).

Summary

Eliminates C-core gRPC fork/threading hazards (I0000 ... fork_posix.cc:71] Other threads are currently calling into gRPC, skipping fork() handlers and post-query background ReleaseExecute storms) and cleans up log, stderr, and notebook cell-output noise across google-cloud-spark-connect, with sensible defaults and explicit Builder / environment-variable controls.

1. Proper gRPC Lifecycle & Fork-Safety Fixes

• Why fork_posix.cc:71 happened:
  1. Post-query ReleaseExecute storm (reattach.py): PySpark's ExecutePlanResponseReattachableIterator submits a background ReleaseExecute(ReleaseUntil) RPC to a shared ThreadPoolExecutor for every streamed Arrow batch plus ReleaseAll on result_complete, while returning to the caller immediately as soon as result_complete arrives. On a 500-batch df.toPandas() call, 500+ background ReleaseExecute RPCs continued firing into C-core gRPC for up to 13 seconds after toPandas() returned, colliding with any subsequent subprocess (!shell, gcloud, %pip, matplotlib) in the same or next notebook cell.
  2. Lazy credential/client subprocesses during in-flight gRPC calls: When connect_tcp_bridge() initialized credentials lazily on the proxy thread while the gRPC thread was already inside C-core waiting on the loopback socket, any gcloud or /opt/google/endpoint-verification/bin/apihelper subprocess forked while a thread held an active gRPC ExecCtx.
  3. Lingering proxy sockets on teardown: ManagedSparkSessionProxy.stop() only closed the listener socket without shutting down active client/bridged sockets or joining forwarding threads.
• How this PR fixes it:
  • Coalesces concurrent ReleaseUntil RPCs and quiesces on completion (client/reattach_patch.py): Allows at most 1 in-flight ReleaseExecute per iterator at a time, coalescing intermediate until_response_id updates to the latest 1-batch-slack ID, aborting redundant ReleaseUntils once _result_complete or _release_all is reached, and draining in-flight release futures (bounded by MANAGED_SPARK_CONNECT_RELEASE_DRAIN_TIMEOUT_S, default 2.0s) before _execute_and_fetch / _execute_and_fetch_as_iterator / stop() returns. In live Dataproc benchmarks (1,000,000 rows, 514 Arrow batches), this reduced ReleaseExecute RPCs from 517 to 3–5 (99%+ reduction) with 0 background gRPC threads remaining after toPandas() returns.
  • Pre-warms credentials and SessionControllerClient before starting the proxy or gRPC channel (session.py, client/core.py): Runs get_refreshed_credentials() and _create_session_controller_client() on the calling thread before spinning up ManagedSparkSessionProxy or the C-core gRPC channel.
  • Cleanly shuts down active proxy sockets and threads (client/proxy.py, client/core.py, session.py): Tracks active client sockets, bridged WebSocket wrappers, and forwarding threads in ManagedSparkSessionProxy and shuts them down (SHUT_RDWR) and joins threads on stop() / ProxiedChannel.close().
  • Defaults GRPC_VERBOSITY to ERROR (__init__.py): Calls os.environ.setdefault("GRPC_VERBOSITY", "ERROR") before any transitive grpc import so benign C-core I0000 info logs from fork()+exec() subprocesses are suppressed while preserving genuine ERROR logs and respecting user-set GRPC_VERBOSITY.

2. Log, Warning, and Notebook Output Hygiene

• No more root-logger hijacking: Removes logging.basicConfig(level=logging.INFO) from session.py (which previously attached a root StreamHandler(sys.stderr) at INFO for the entire Python process at import time) and attaches a logging.NullHandler() to logging.getLogger("google.cloud.managed_spark_connect") in __init__.py. Setting MANAGED_SPARK_CONNECT_LOG_LEVEL=<LEVEL> configures only the google.cloud.managed_spark_connect logger hierarchy.
• No more non-interactive tqdm threads or stray print()s:
  • Gates the session-creation tqdm progress-bar thread (create_session_pbar) and its trailing newline on interactivity (environment.is_interactive() or MANAGED_SPARK_CONNECT_FORCE_PROGRESS_BAR=1).
  • Demotes bare print(f"... Managed Spark Session deleted: {e}") calls in get_active_s8s_session_response() (which fired on routine NotFound checks when creating a custom sessionId) and duplicate logger.info calls in _get_session_by_id() to logger.debug(...).
  • Routes first-time session attach messages through _output_element_or_message() and keeps repeated in-process getOrCreate() singleton returns silent.
• Clean per-query notebook output & new Builder / env knobs:
  • Defaults execution progress bar leave to False so transient df.columns / df.count() bars clean up after completion instead of stacking in notebook cells.
  • Adds Builder.showOperationLinks(bool) (MANAGED_SPARK_CONNECT_SHOW_OPERATION_LINKS), Builder.showProgressBar(bool) (MANAGED_SPARK_CONNECT_PROGRESS_BAR), and Builder.progressBarLeave(bool) (MANAGED_SPARK_CONNECT_PROGRESS_BAR_LEAVE).
• Additional noise & warning fixes:
  • Normalizes three-part runtime_version strings (e.g. "2.3.10" -> "2.3") before looking up in RUNTIME_PYTHON_MAP in _check_python_version_compatibility() so patch-level runtime versions do not emit false "Unrecognized Spark runtime version" warnings.
  • Silences PySpark 4.0's SQLQueryContextLogger / DataFrameQueryContextLogger raw JSON stderr dumps on query exceptions unless SPARK_CONNECT_LOG_LEVEL or MANAGED_SPARK_CONNECT_LOG_LEVEL is set.
  • Filters PySpark 4.0's internal Pandas4Warning: The copy keyword is deprecated emitted inside toPandas().
  • Demotes unpinned-package recommendations in pypi_artifacts.py and ProxiedChannel._check_active() inactive-session log lines (right before raising RuntimeError) to DEBUG, adds stacklevel=2 to the deprecated package warning in __init__.py, and replaces assert all(...) with a clean early return in _display_operation_link().

---

Commits in this PR

1. fix(grpc): set default GRPC_VERBOSITY=ERROR, pre-warm credentials, and close proxy sockets cleanly
2. fix(reattach): coalesce concurrent ReleaseUntil RPCs and quiesce release pool on completion
3. fix(logging): remove root basicConfig and gate session/query output on interactivity and builder knobs

Testing

• Added unit and subprocess regression tests in tests/unit/test_grpc_fork_noise.py, tests/unit/test_reattach_patch.py, and tests/unit/test_logging_hygiene.py.
• Verified live against Dataproc Serverless Spark Connect sessions across both the 96-trial resilience suite and subprocess/notebook hygiene probes (0 fork_posix.cc lines, 0 leaked tqdm bars, 0 stray stdout/stderr lines in non-interactive mode).

[gemini-code-assist]

Code Review

This pull request introduces several enhancements to the managed Spark Connect client, including credential pre-warming to avoid gRPC fork-safety issues, coalesced ReleaseUntil management to prevent background thread build-up, and improved proxy socket and thread lifecycle management to ensure clean teardown. Feedback on the changes highlights a concurrency race condition in _ensure_iterator_release_state where attributes may be accessed before initialization, a potential session leak if self.client.close() raises an exception during session stop, and cross-platform test failures on Windows due to Unix-specific commands and os.fork() usage.

[dborowitz]

/gemini review

[gemini-code-assist]

Code Review

This pull request introduces several enhancements to the Managed Spark Connect client, including ReleaseExecute RPC coalescing to prevent queuing stale release calls, improved logging hygiene (such as silencing gRPC fork noise and PySpark query context dumps), and customizable display settings for operation links and progress bars. It also refactors session lifecycle management to prevent reference cycles and ensure clean teardown. The review feedback highlights critical compatibility issues with older PySpark versions (such as potential ModuleNotFoundError and AttributeError during cleanup), a thread-safety concern when initializing the coalesced release state, and a potential AttributeError if the release thread pool is uninitialized.