diff --git a/mkdocs/docs/configuration.md b/mkdocs/docs/configuration.md index 54d33dd00e..ae845d3ef1 100644 --- a/mkdocs/docs/configuration.md +++ b/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: diff --git a/pyiceberg/catalog/hive.py b/pyiceberg/catalog/hive.py index 181f9d4661..ee3c4b3625 100644 --- a/pyiceberg/catalog/hive.py +++ b/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 diff --git a/tests/catalog/test_hive.py b/tests/catalog/test_hive.py index f594aa876e..2900b979e7 100644 --- a/tests/catalog/test_hive.py +++ b/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"