fix: avoid TSaslClientTransport reuse to eliminate server-side SASL noise

Author: barking-code

pyiceberg/catalog/hive.py

@@ -142,9 +142,17 @@
 
 
 class _HiveClient:
-    """Helper class to nicely open and close the transport."""
+    """Helper class to nicely open and close the transport.
 
-    _transport: TTransport
+    ``TSaslClientTransport`` instances are single-use: ``close()`` disposes
+    the SASL session and the same instance cannot be reopened. The
+    transport is therefore created lazily in ``__enter__`` rather than held
+    on the class, which lets the ``_HiveClient`` itself live for the
+    duration of a ``HiveCatalog`` while still honouring the SASL transport
+    lifecycle.
+    """
+
+    _transport: "TTransport | None"
     _ugi: list[str] | None
 
     def __init__(
@@ -158,7 +166,7 @@ def __init__(
         self._kerberos_auth = kerberos_auth
         self._kerberos_service_name = kerberos_service_name
         self._ugi = ugi.split(":") if ugi else None
-        self._transport = self._init_thrift_transport()
+        self._transport = None
 
     def _init_thrift_transport(self) -> TTransport:
         url_parts = urlparse(self._uri)
@@ -169,31 +177,28 @@ def _init_thrift_transport(self) -> TTransport:
             return TTransport.TSaslClientTransport(socket, host=url_parts.hostname, service=self._kerberos_service_name)
 
     def _client(self) -> Client:
+        assert self._transport is not None  # set by __enter__
         protocol = TBinaryProtocol.TBinaryProtocol(self._transport)
         client = Client(protocol)
         if self._ugi:
             client.set_ugi(*self._ugi)
         return client
 
     def __enter__(self) -> Client:
-        """Make sure the transport is initialized and open."""
-        if not self._transport.isOpen():
-            try:
-                self._transport.open()
-            except (TypeError, TTransport.TTransportException):
-                # Close the old transport before reinitializing to prevent resource leaks
-                try:
-                    self._transport.close()
-                except Exception:
-                    pass
-                # reinitialize _transport
-                self._transport = self._init_thrift_transport()
-                self._transport.open()
-        return self._client()  # recreate the client
+        """Initialize and open a fresh transport for this entry.
+
+        A new ``TSaslClientTransport`` is created on every entry because
+        the underlying SASL session cannot be reused after ``close()``.
+        Reusing a closed transport would leave a half-opened TCP socket
+        and SASL handshake EOF noise on the server.
+        """
+        self._transport = self._init_thrift_transport()
+        self._transport.open()
+        return self._client()
 
     def __exit__(self, exctype: type[BaseException] | None, excinst: BaseException | None, exctb: TracebackType | None) -> None:
         """Close transport if it was opened."""
-        if self._transport.isOpen():
+        if self._transport is not None and self._transport.isOpen():
             self._transport.close()
 
 

tests/catalog/test_hive.py

@@ -1410,3 +1410,42 @@ def test_create_hive_client_with_kerberos_using_context_manager(
         # closing and re-opening work as expected.
         with client as open_client:
             assert open_client._iprot.trans.isOpen()
+
+
+def test_kerberized_client_uses_fresh_transport_per_entry(
+    kerberized_hive_metastore_fake_url: str,
+) -> None:
+    """TSaslClientTransport is single-use — each entry must create a new one.
+
+    The underlying SASL session cannot be reused after close(). Each
+    context-manager entry must therefore produce a fresh transport
+    instance rather than reopening the previous one (which would leave
+    a half-opened TCP socket and SASL handshake EOF noise on the server).
+    """
+    client = _HiveClient(
+        uri=kerberized_hive_metastore_fake_url,
+        kerberos_auth=True,
+    )
+    # No transport should exist until the context manager is entered.
+    assert client._transport is None
+    with (
+        patch(
+            "puresasl.mechanisms.kerberos.authGSSClientStep",
+            return_value=None,
+        ),
+        patch(
+            "puresasl.mechanisms.kerberos.authGSSClientResponse",
+            return_value=base64.b64encode(b"Some Response"),
+        ),
+        patch(
+            "puresasl.mechanisms.GSSAPIMechanism.complete",
+            return_value=True,
+        ),
+    ):
+        with client:
+            first_transport_id = id(client._transport)
+
+        with client:
+            second_transport_id = id(client._transport)
+
+        assert first_transport_id != second_transport_id