Add hive.kerberos-service-host configuration option

mkdocs/docs/configuration.md

@@ -749,6 +749,7 @@ catalog:
 | hive.hive2-compatible        | true    | Using Hive 2.x compatibility mode    |
 | hive.kerberos-authentication | true    | Using authentication via Kerberos    |
 | hive.kerberos-service-name   | hive    | Kerberos service name (default hive) |
+| hive.kerberos-service-host   | hive-host | Kerberos service host (default URI host) |
 | ugi                 | t-1234:secret                    | Hadoop UGI for Hive client.                                                                        |
 
 When using Hive 2.x, make sure to set the compatibility flag:

pyiceberg/catalog/hive.py

@@ -129,6 +129,7 @@
 HIVE_KERBEROS_AUTH_DEFAULT = False
 HIVE_KERBEROS_SERVICE_NAME = "hive.kerberos-service-name"
 HIVE_KERBEROS_SERVICE_NAME_DEFAULT = "hive"
+HIVE_KERBEROS_SERVICE_HOST = "hive.kerberos-service-host"
 
 LOCK_CHECK_MIN_WAIT_TIME = "lock-check-min-wait-time"
 LOCK_CHECK_MAX_WAIT_TIME = "lock-check-max-wait-time"
@@ -155,10 +156,12 @@ def __init__(
         ugi: str | None = None,
         kerberos_auth: bool | None = HIVE_KERBEROS_AUTH_DEFAULT,
         kerberos_service_name: str | None = HIVE_KERBEROS_SERVICE_NAME,
+        kerberos_service_host: str | None = None,
     ):
         self._uri = uri
         self._kerberos_auth = kerberos_auth
         self._kerberos_service_name = kerberos_service_name
+        self._kerberos_service_host = kerberos_service_host
         self._ugi = ugi.split(":") if ugi else None
         self._transport = self._init_thrift_transport()
         self._was_opened = False
@@ -169,7 +172,8 @@ def _init_thrift_transport(self) -> TTransport:
         if not self._kerberos_auth:
             return TTransport.TBufferedTransport(socket)
         else:
-            return TTransport.TSaslClientTransport(socket, host=url_parts.hostname, service=self._kerberos_service_name)
+            host = self._kerberos_service_host or url_parts.hostname
+            return TTransport.TSaslClientTransport(socket, host=host, service=self._kerberos_service_name)
 
     def _client(self) -> Client:
         protocol = TBinaryProtocol.TBinaryProtocol(self._transport)
@@ -316,6 +320,7 @@ def _create_hive_client(properties: dict[str, str]) -> _HiveClient:
                     properties.get("ugi"),
                     property_as_bool(properties, HIVE_KERBEROS_AUTH, HIVE_KERBEROS_AUTH_DEFAULT),
                     properties.get(HIVE_KERBEROS_SERVICE_NAME, HIVE_KERBEROS_SERVICE_NAME_DEFAULT),
+                    properties.get(HIVE_KERBEROS_SERVICE_HOST),
                 )
             except BaseException as e:
                 last_exception = e

tests/catalog/test_hive.py

@@ -48,6 +48,7 @@
     DO_NOT_UPDATE_STATS,
     DO_NOT_UPDATE_STATS_DEFAULT,
     HIVE_KERBEROS_AUTH,
+    HIVE_KERBEROS_SERVICE_HOST,
     HIVE_KERBEROS_SERVICE_NAME,
     LOCK_CHECK_MAX_WAIT_TIME,
     LOCK_CHECK_MIN_WAIT_TIME,
@@ -1330,7 +1331,7 @@ def test_create_hive_client_success() -> None:
 
     with patch("pyiceberg.catalog.hive._HiveClient", return_value=MagicMock()) as mock_hive_client:
         client = HiveCatalog._create_hive_client(properties)
-        mock_hive_client.assert_called_once_with("thrift://localhost:10000", "user", False, "hive")
+        mock_hive_client.assert_called_once_with("thrift://localhost:10000", "user", False, "hive", None)
         assert client is not None
 
 
@@ -1343,7 +1344,21 @@ def test_create_hive_client_with_kerberos_success() -> None:
     }
     with patch("pyiceberg.catalog.hive._HiveClient", return_value=MagicMock()) as mock_hive_client:
         client = HiveCatalog._create_hive_client(properties)
-        mock_hive_client.assert_called_once_with("thrift://localhost:10000", "user", True, "hiveuser")
+        mock_hive_client.assert_called_once_with("thrift://localhost:10000", "user", True, "hiveuser", None)
+        assert client is not None
+
+
+def test_create_hive_client_with_kerberos_service_host() -> None:
+    properties = {
+        "uri": "thrift://localhost:10000",
+        "ugi": "user",
+        HIVE_KERBEROS_AUTH: "true",
+        HIVE_KERBEROS_SERVICE_NAME: "hiveuser",
+        HIVE_KERBEROS_SERVICE_HOST: "hive-host.example.com",
+    }
+    with patch("pyiceberg.catalog.hive._HiveClient", return_value=MagicMock()) as mock_hive_client:
+        client = HiveCatalog._create_hive_client(properties)
+        mock_hive_client.assert_called_once_with("thrift://localhost:10000", "user", True, "hiveuser", "hive-host.example.com")
         assert client is not None
 
 
@@ -1356,7 +1371,10 @@ def test_create_hive_client_multiple_uris() -> None:
         client = HiveCatalog._create_hive_client(properties)
         assert mock_hive_client.call_count == 2
         mock_hive_client.assert_has_calls(
-            [call("thrift://localhost:10000", "user", False, "hive"), call("thrift://localhost:10001", "user", False, "hive")]
+            [
+                call("thrift://localhost:10000", "user", False, "hive", None),
+                call("thrift://localhost:10001", "user", False, "hive", None),
+            ]
         )
         assert client is not None
 
@@ -1445,3 +1463,23 @@ def test_kerberized_client_uses_fresh_transport_on_reuse(
             second_transport_id = id(client._transport)
 
         assert first_transport_id != second_transport_id
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="Kerberos/puresasl not available on Windows")
+def test_kerberized_client_uses_configured_service_host() -> None:
+    """The SASL host must come from hive.kerberos-service-host, or the URI host when unset."""
+    configured = _HiveClient(
+        uri="thrift://metastore-lb:9083",
+        kerberos_auth=True,
+        kerberos_service_name="hive",
+        kerberos_service_host="metastore-host",
+    )
+    assert configured._transport.sasl.host == "metastore-host"
+    assert configured._transport.sasl.service == "hive"
+
+    default = _HiveClient(
+        uri="thrift://metastore-lb:9083",
+        kerberos_auth=True,
+        kerberos_service_name="hive",
+    )
+    assert default._transport.sasl.host == "metastore-lb"