fix(hygiene): eliminate gRPC fork/ReleaseExecute hazards and clean up logging and notebook output noise

README.md

@@ -124,6 +124,11 @@ The `ManagedSparkSession.builder` provides a fluent API to configure the session
 | `subnetwork(subnet)` | Sets the subnetwork URI for the session. |
 | `ttl(duration)` | Sets the time-to-live (TTL) for the session using a `datetime.timedelta` object. |
 | `instrumentation(inst)` | Records what the library does on your behalf; see [Instrumentation (opt-in)](#instrumentation-opt-in). |
+| `showOperationLinks(enabled)` | Show each query's Spark UI link (default `True`). Links only appear in notebooks and interactive terminals; `False` hides them there too. |
+| `showProgressBar(enabled)` | Show session-creation and query progress bars (default `True`). Like links, they only appear in notebooks and interactive terminals. |
+| `progressBarLeave(leave)` | Keep a query's progress bar on screen once it finishes (default `False`). |
+
+The three display settings apply when `getOrCreate()` creates or attaches to a session, not to a session it reuses from this process.
 
 ### Configuration (environment variables)
 
@@ -134,6 +139,12 @@ These environment variables change how the client behaves.
 | `MANAGED_SPARK_CONNECT_REPLAY_ON_POSITION_LOST` | `1` | When a read query (`collect`, `toPandas`, `count`, ...) fails with `INVALID_CURSOR.POSITION_NOT_AVAILABLE` because the server no longer buffers the result the client has to resume from, re-run it once under a new operation id and log a `WARNING`. Commands (writes, DDL) are never re-run. Set to `0` to disable, e.g. for queries that are expensive or not deterministic. Read when the session is created. |
 | `MANAGED_SPARK_CONNECT_WS_SILENCE_LIMIT_S` | `300` | Cut the proxy WebSocket connection if an RPC is in flight and the server stays silent (no frames received) for this many seconds, so a wedged stream fails instead of hanging. The check runs on each 60 s WebSocket receive poll, so the cut happens between this limit and the limit plus 60 s. Set to `0` to disable. |
 | `MANAGED_SPARK_CONNECT_LIVENESS_TTL_SECONDS` | unset | Before an RPC, the client checks that the Dataproc session is still alive. Once a check finds it `ACTIVE`, that answer is reused for this many seconds: unset or negative reuses it until an RPC fails, `0` checks before every RPC. A failed RPC always triggers a fresh check. The session only counts as gone when Dataproc reports it not found or in a terminal state; when the check itself fails (for example during a control-plane outage), the RPC proceeds. |
+| `GRPC_VERBOSITY` | `ERROR` | gRPC's own variable. If it is unset when this package is imported, and gRPC hasn't been imported yet, the package sets it to `ERROR` for the whole process to silence gRPC's informational fork messages. Set it yourself to override. |
+| `MANAGED_SPARK_CONNECT_LOG_LEVEL` | unset | A level name (`DEBUG`, `INFO`, ...) or a positive number for the `google.cloud.managed_spark_connect` loggers. If the root logger has no handlers when the package is imported, the library logs to stderr itself, and its records then don't reach root handlers configured later: configure logging before importing the package to send them elsewhere. The library never configures the root logger. Setting it also lets pyspark print its query-context JSON for failed queries (see `SPARK_CONNECT_LOG_LEVEL`). |
+| `SPARK_CONNECT_LOG_LEVEL` | unset | pyspark's own variable. Unless it or `MANAGED_SPARK_CONNECT_LOG_LEVEL` is set, creating a session silences pyspark's `SQLQueryContextLogger` and `DataFrameQueryContextLogger`, which otherwise print a JSON dump to stderr for every failed query. This applies to the whole process. |
+| `MANAGED_SPARK_CONNECT_SHOW_OPERATION_LINKS` | `true` | Default for `showOperationLinks()`. |
+| `MANAGED_SPARK_CONNECT_PROGRESS_BAR` | `true` | Default for `showProgressBar()`. |
+| `MANAGED_SPARK_CONNECT_PROGRESS_BAR_LEAVE` | `false` | Default for `progressBarLeave()`. |
 
 ### Reusing Named Sessions Across Notebooks
 
@@ -177,6 +188,12 @@ result conversions — as timed *spans*. It is off by default; opt in via
 `MANAGED_SPARK_CONNECT_INSTRUMENTATION_FILE` and
 `MANAGED_SPARK_CONNECT_INSTRUMENTATION_LOG` environment variables.
 
+Without instrumentation the session still hooks a few pyspark client methods
+on its own client (operation ids and links, progress bars, the
+`INVALID_CURSOR.POSITION_NOT_AVAILABLE` replay; see
+[Configuration](#configuration-environment-variables)) and coalesces the
+`ReleaseExecute` RPCs of its result streams.
+
 See [docs/instrumentation.md](https://github.com/GoogleCloudDataproc/managed-spark-connect-python/blob/main/docs/instrumentation.md)
 for usage, the API, the record schema, and the span and event catalogues.
 

docs/instrumentation.md

@@ -76,6 +76,10 @@ JsonlFileSink(path, *, append=True)   InMemorySink()   LoggingSink(logger=None,
 they set operation ids, display operation links, close a query's progress
 bar when it ends and re-run a read query once on
 `INVALID_CURSOR.POSITION_NOT_AVAILABLE` (see `reattach.position_lost` below).
+Every session also coalesces the `ReleaseExecute(until)` RPCs of its result
+streams, through a process-wide patch of pyspark's
+`ExecutePlanResponseReattachableIterator._release_until` that leaves the
+iterators of other clients alone.
 
 ## Record schema
 

google/cloud/managed_spark_connect/__init__.py

@@ -13,9 +13,20 @@
 # limitations under the License.
 import importlib.metadata
 import importlib.util
+import os
 import sys
 import warnings
 
+# gRPC C-core logs fork-handler noise at INFO; default it to ERROR. gRPC reads
+# the variable once, when it is first imported, so this only applies when this
+# package is imported before grpc; an explicit GRPC_VERBOSITY always wins.
+if "grpc" not in sys.modules:
+    os.environ.setdefault("GRPC_VERBOSITY", "ERROR")
+
+from . import _logging
+
+_logging.configure_package_logger()
+
 
 def _in_ipython() -> bool:
     # Not environment.is_interactive(), which is also true in a plain
@@ -92,7 +103,8 @@ def _check_pyspark_version(pyspark_version: str) -> None:
         warnings.warn(
             f"Package '{old_package_name}' is already installed in your environment. "
             f"This might cause conflicts with '{current_package_name}'. "
-            f"Consider uninstalling '{old_package_name}' and only install '{current_package_name}'."
+            f"Consider uninstalling '{old_package_name}' and only install '{current_package_name}'.",
+            stacklevel=2,
         )
     except Exception:
         pass

google/cloud/managed_spark_connect/_logging.py

@@ -0,0 +1,78 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Logging setup. Never configures the root logger."""
+
+import logging
+import os
+from typing import Optional
+
+PACKAGE_LOGGER_NAME = "google.cloud.managed_spark_connect"
+ENV_LOG_LEVEL = "MANAGED_SPARK_CONNECT_LOG_LEVEL"
+_PYSPARK_QUERY_CONTEXT_LOGGERS = (
+    "SQLQueryContextLogger",
+    "DataFrameQueryContextLogger",
+)
+
+
+def _parse_level(raw: str) -> Optional[int]:
+    """A positive logging level from a name (``DEBUG``) or a number."""
+    raw = raw.strip()
+    level = int(raw) if raw.isdigit() else getattr(logging, raw.upper(), None)
+    return level if isinstance(level, int) and level > 0 else None
+
+
+def configure_package_logger() -> logging.Logger:
+    """Give the package logger a NullHandler and apply ``LOG_LEVEL``.
+
+    With ``MANAGED_SPARK_CONNECT_LOG_LEVEL`` set and no handler on the root
+    logger, the package logger gets its own ``StreamHandler`` and stops
+    propagating, so a later ``logging.basicConfig()`` doesn't print every
+    record twice.
+    """
+    pkg_logger = logging.getLogger(PACKAGE_LOGGER_NAME)
+    if not any(isinstance(h, logging.NullHandler) for h in pkg_logger.handlers):
+        pkg_logger.addHandler(logging.NullHandler())
+    level = _parse_level(os.environ.get(ENV_LOG_LEVEL, ""))
+    if level is None:
+        return pkg_logger
+    pkg_logger.setLevel(level)
+    has_stream_handler = any(
+        not isinstance(h, logging.NullHandler) for h in pkg_logger.handlers
+    )
+    if not has_stream_handler and not logging.getLogger().handlers:
+        pkg_logger.addHandler(logging.StreamHandler())
+        pkg_logger.propagate = False
+    return pkg_logger
+
+
+def quiet_pyspark_query_context_loggers() -> None:
+    """Stop pyspark from dumping every failed query's context to stderr.
+
+    pyspark's ``SQLQueryContextLogger`` and ``DataFrameQueryContextLogger``
+    write a JSON record with the server stack trace whenever an
+    ``AnalysisException`` or similar is built, on top of the exception
+    itself. Unless ``SPARK_CONNECT_LOG_LEVEL`` or
+    ``MANAGED_SPARK_CONNECT_LOG_LEVEL`` is set, raise both loggers above
+    ``CRITICAL``. They are process-wide, so this applies to every pyspark
+    session in the process.
+    """
+    if (
+        os.environ.get("SPARK_CONNECT_LOG_LEVEL", "").strip()
+        or os.environ.get(ENV_LOG_LEVEL, "").strip()
+    ):
+        return
+    from pyspark.logger.logger import PySparkLogger
+
+    for name in _PYSPARK_QUERY_CONTEXT_LOGGERS:
+        PySparkLogger.getLogger(name).setLevel(logging.CRITICAL + 1)

google/cloud/managed_spark_connect/client/core.py

@@ -401,45 +401,54 @@ class ProxiedChannel(grpc.Channel):
 
     def __init__(self, target_host, is_active_callback, instrumentation=None):
         recorder = instrumentation_lib.get_recorder(instrumentation)
-        self._supports_error_guard = (
-            getattr(is_active_callback, "_supports_error_guard", False) is True
-        )
         if recorder.enabled:
             is_active_callback = instrumentation_lib.timed_guard(
                 is_active_callback
             )
         self._is_active_callback = is_active_callback
         self._guard_errors = _OutageLog(logger)
+        # Discover and refresh credentials on the calling thread, before the
+        # proxy and gRPC threads start, so a credential helper that forks
+        # (gcloud) never does so while a gRPC thread holds an active ExecCtx.
+        # The result is cached for the proxy's own connections.
+        try:
+            proxy.get_refreshed_credentials(recorder=recorder)
+        except Exception:
+            logger.debug("Credential pre-warm failed", exc_info=True)
         self._proxy = proxy.ManagedSparkSessionProxy(
             0, target_host, recorder=recorder
         )
         self._proxy.start()
-        self._proxied_connect_url = f"sc://localhost:{self._proxy.port}"
-        self._wrapped = instrumentation_lib.wrap_channel(
-            DefaultChannelBuilder(self._proxied_connect_url).toChannel(),
-            recorder,
-            instrumentation_lib.INNER,
-        )
+        try:
+            self._proxied_connect_url = f"sc://localhost:{self._proxy.port}"
+            self._wrapped = instrumentation_lib.wrap_channel(
+                DefaultChannelBuilder(self._proxied_connect_url).toChannel(),
+                recorder,
+                instrumentation_lib.INNER,
+            )
+        except BaseException:
+            self._proxy.stop()
+            raise
 
     def __enter__(self):
         return self
 
     def __exit__(self, *args):
-        ret = self._wrapped.__exit__(*args)
-        self._proxy.stop()
-        return ret
+        try:
+            return self._wrapped.__exit__(*args)
+        finally:
+            self._proxy.stop()
 
     def close(self):
-        ret = self._wrapped.close()
-        self._proxy.stop()
-        return ret
+        try:
+            return self._wrapped.close()
+        finally:
+            self._proxy.stop()
 
     def _check_active(self):
         if self._is_active_callback is None:
             return
         mode = _GUARD_MODE.get()
-        if mode == "on_error" and not self._supports_error_guard:
-            return
         try:
             active = self._is_active_callback()
         except Exception as exc:
@@ -457,7 +466,7 @@ def _check_active(self):
             return
         self._guard_errors.success()
         if not active:
-            logger.warning(f"Session is no longer active")
+            logger.debug("Session is no longer active")
             raise RuntimeError(
                 "Session not active. Please create a new session"
             )