This is an automated email from the ASF dual-hosted git repository. diqiu50 pushed a commit to branch trino-irc-1.3 in repository https://gitbox.apache.org/repos/asf/gravitino.git
commit bc220aa9897696fe120bce7bbe50d15269e99af8 Author: yuhui <[email protected]> AuthorDate: Thu Aug 27 07:08:36 2026 +0000 fix(spark-connector): align Iceberg REST routing auth --- docs/spark-connector/spark-catalog-iceberg.md | 51 ++++++- docs/spark-connector/spark-connector.md | 2 + .../spark/connector/GravitinoSparkConfig.java | 4 + .../connector/catalog/GravitinoCatalogManager.java | 15 +-- .../connector/iceberg/GravitinoIcebergCatalog.java | 53 +++++++- .../connector/iceberg/IcebergRestOAuthConfig.java | 78 ++++++++--- .../TestGravitinoIcebergCatalogRestRouting.java | 150 +++++++++++++++++++++ .../iceberg/TestIcebergRestOAuthConfig.java | 82 ++++++++--- 8 files changed, 377 insertions(+), 58 deletions(-) diff --git a/docs/spark-connector/spark-catalog-iceberg.md b/docs/spark-connector/spark-catalog-iceberg.md index ee8a5080a1..6d79cad215 100644 --- a/docs/spark-connector/spark-catalog-iceberg.md +++ b/docs/spark-connector/spark-catalog-iceberg.md @@ -160,10 +160,11 @@ instead of the static, long-lived secrets (JDBC password, S3/OSS access keys) th placed in the Spark Iceberg connector configuration. See [Credential vending](../security/credential-vending.md) for how to enable credential vending on the Gravitino server. -The endpoint is discovered once, when a catalog is first initialized in a Spark session; it is not -re-checked afterward. If the server does not expose a discoverable endpoint (for example, the `iceberg-rest` -auxiliary service is disabled, or not configured with `catalog-config-provider=dynamic-config-provider`), the -catalog falls back to the existing hive/jdbc translation described above. +REST routing is enabled by default. The endpoint is discovered once, when a catalog is first initialized +in a Spark session; it is not re-checked afterward. If the server does not expose a discoverable endpoint +(for example, the `iceberg-rest` auxiliary service is disabled or not configured with +`catalog-config-provider=dynamic-config-provider`), catalog initialization fails instead of silently using +the native Hive/JDBC backend. To force a specific endpoint instead of relying on auto-discovery, set: @@ -171,6 +172,12 @@ To force a specific endpoint instead of relying on auto-discovery, set: spark.sql.gravitino.iceberg.rest-uri http://<gravitino-host>:9001/iceberg ``` +To retain the legacy Hive/JDBC translation and skip endpoint discovery, disable routing explicitly: + +```properties +spark.sql.gravitino.iceberg.rest-routing-enabled false +``` + If Gravitino requires authentication on the IRC endpoint, pass the Iceberg REST client's own auth properties using the `spark.sql.gravitino.iceberg.rest.` prefix. For example, for Basic authentication: @@ -183,6 +190,36 @@ spark.sql.gravitino.iceberg.rest.rest.auth.basic.password <password> See [Connect Spark to Iceberg REST](../iceberg-rest-engine/spark.md) for the full set of supported `rest.auth.*` properties and how to configure them when connecting directly to the IRC endpoint. +When the Gravitino client uses OAuth2, the connector reuses its OAuth2 client configuration for IRC by +default. This avoids duplicating configuration when both endpoints accept the same client identity. The +equivalent explicit setting is: + +```properties +spark.sql.gravitino.iceberg.reuseOAuth2 true +``` + +Gravitino and IRC may use different OAuth2 clients even when IRC runs as a Gravitino auxiliary service. +Override any reused value with an IRC-specific Iceberg REST property; unspecified values continue to come +from the Gravitino client configuration: + +```properties +# Gravitino metadata API client +spark.sql.gravitino.authType oauth2 +spark.sql.gravitino.oauth2.serverUri https://identity.example.com +spark.sql.gravitino.oauth2.tokenPath /oauth/token +spark.sql.gravitino.oauth2.credential <gravitino-client-id>:<gravitino-client-secret> +spark.sql.gravitino.oauth2.scope gravitino + +# IRC data-plane client override +spark.sql.gravitino.iceberg.rest.credential <irc-client-id>:<irc-client-secret> +spark.sql.gravitino.iceberg.rest.scope iceberg +``` + +The IRC properties take precedence field by field. If an explicit legacy OAuth2 configuration cannot be +completed from the reused Gravitino configuration, catalog initialization fails and identifies the missing +properties. Set `spark.sql.gravitino.iceberg.reuseOAuth2=false` when supplying a complete, independent IRC +authentication configuration or when IRC does not require OAuth2. + Because vended credentials are only consumed by Iceberg's native `FileIO` implementations, make sure the warehouse storage jars listed under [Storage](#storage) below are on the Spark classpath; the connector derives `io-impl` automatically from the warehouse location's scheme (`s3`/`s3a`/`s3n`, `gs`, @@ -194,7 +231,11 @@ Spark connector could convert storage properties in the Gravitino catalog to Spa ### S3 -Please downloading the [Iceberg AWS bundle](https://mvnrepository.com/artifact/org.apache.iceberg/iceberg-aws-bundle) and place it in the classpath of Spark. +Download the [Iceberg AWS bundle](https://mvnrepository.com/artifact/org.apache.iceberg/iceberg-aws-bundle) +that matches the Iceberg runtime version and place it on the Spark driver and executor classpaths. This is +required for `S3FileIO` even when the Spark image already includes the AWS SDK v1 used by Hadoop S3A; +`S3FileIO` uses AWS SDK v2 from `iceberg-aws-bundle`. If the bundle is absent, initialization can fail with +a `NoClassDefFoundError` that does not identify the missing bundle directly. ### OSS diff --git a/docs/spark-connector/spark-connector.md b/docs/spark-connector/spark-connector.md index 8f24b1d9b3..d3466c20bb 100644 --- a/docs/spark-connector/spark-connector.md +++ b/docs/spark-connector/spark-connector.md @@ -34,7 +34,9 @@ The Apache Gravitino Spark connector leverages the Spark DataSourceV2 interface | spark.sql.gravitino.enableIcebergSupport | string | `false` | Set to `true` to use Iceberg catalog. | No | | spark.sql.gravitino.enablePaimonSupport | string | `false` | Set to `true` to use Paimon catalog. | No | | spark.sql.gravitino.client. | string | (none) | The configuration key prefix for the Gravitino client config. | No | +| spark.sql.gravitino.iceberg.rest-routing-enabled | boolean | `true` | Whether `hive` and `jdbc` backed Iceberg catalogs must be routed through the Gravitino Iceberg REST server. Set to `false` to retain legacy native backend translation. | No | | spark.sql.gravitino.iceberg.rest-uri | string | (none) | Overrides the auto-discovered Gravitino Iceberg REST server endpoint. See [Iceberg catalog](spark-catalog-iceberg.md#routing-through-the-gravitino-iceberg-rest-server). | No | +| spark.sql.gravitino.iceberg.reuseOAuth2 | boolean | `true` | Reuses the Gravitino OAuth2 client configuration for routed Iceberg REST catalogs. Explicit IRC OAuth2 properties override individual reused values. | No | | spark.sql.gravitino.iceberg.rest. | string | (none) | The configuration key prefix for the Iceberg REST client config (e.g. `rest.auth.type`), applied when a catalog is routed through the Gravitino Iceberg REST server. | No | To configure the Gravitino client, use properties prefixed with `spark.sql.gravitino.client.`. These properties will be passed to the Gravitino client after removing the `spark.sql.` prefix. diff --git a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/GravitinoSparkConfig.java b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/GravitinoSparkConfig.java index 119c2f9501..1fdf8377f7 100644 --- a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/GravitinoSparkConfig.java +++ b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/GravitinoSparkConfig.java @@ -36,6 +36,10 @@ public class GravitinoSparkConfig { // lakehouse-iceberg catalogs are routed through; takes precedence over auto-discovery. public static final String GRAVITINO_ICEBERG_REST_URI = GRAVITINO_PREFIX + "iceberg.rest-uri"; + /** Whether hive/jdbc backed Iceberg catalogs must be routed through Iceberg REST. */ + public static final String GRAVITINO_ICEBERG_REST_ROUTING_ENABLED = + GRAVITINO_PREFIX + "iceberg.rest-routing-enabled"; + // Reuses the Gravitino OAuth2 client configuration for automatically routed Iceberg REST // catalogs. Iceberg obtains and refreshes its own access token with the same client identity. public static final String GRAVITINO_ICEBERG_REUSE_OAUTH2 = diff --git a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/catalog/GravitinoCatalogManager.java b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/catalog/GravitinoCatalogManager.java index 0d92b96ee8..d82876a3fe 100644 --- a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/catalog/GravitinoCatalogManager.java +++ b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/catalog/GravitinoCatalogManager.java @@ -96,8 +96,8 @@ public class GravitinoCatalogManager { /** * Resolves the Gravitino Iceberg REST server endpoint for this manager's metalake, if the server - * exposes one. The lookup is performed once and the result, including a lookup failure, is cached - * for the lifetime of this manager. + * exposes one. The lookup is performed once and a successful response, including a response with + * no endpoint, is cached for the lifetime of this manager. Request failures are propagated. * * @return the discovered Iceberg REST endpoint, or empty if none is available */ @@ -113,16 +113,7 @@ public class GravitinoCatalogManager { } private Optional<String> resolveIcebergRestUri() { - try { - return gravitinoClient.icebergRestServiceUri(metalakeName); - } catch (Exception e) { - LOG.debug( - "No Iceberg REST server endpoint is available for metalake {}, " - + "falling back to native catalog backend routing.", - metalakeName, - e); - return Optional.empty(); - } + return gravitinoClient.icebergRestServiceUri(metalakeName); } private Catalog loadCatalog(String catalogName) { diff --git a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/iceberg/GravitinoIcebergCatalog.java b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/iceberg/GravitinoIcebergCatalog.java index 1fc5b5b031..a2a042db4b 100644 --- a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/iceberg/GravitinoIcebergCatalog.java +++ b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/iceberg/GravitinoIcebergCatalog.java @@ -25,6 +25,8 @@ import java.util.HashMap; import java.util.Locale; import java.util.Map; import java.util.Optional; +import java.util.function.Supplier; +import java.util.function.UnaryOperator; import java.util.stream.Collectors; import java.util.stream.Stream; import org.apache.commons.lang3.StringUtils; @@ -81,7 +83,11 @@ public class GravitinoIcebergCatalog extends BaseCatalog } } String catalogBackendName = IcebergPropertiesUtils.getCatalogBackendName(properties); - Optional<String> icebergRestUri = resolveIcebergRestUri(properties); + Optional<String> icebergRestUri = + resolveIcebergRestUri( + properties, + key -> SparkSession.active().conf().get(key, null), + () -> GravitinoCatalogManager.get().getIcebergRestUri()); Map<String, String> all; if (icebergRestUri.isPresent()) { all = buildAutoRoutedIcebergRestProperties(name, options, properties, icebergRestUri.get()); @@ -100,7 +106,10 @@ public class GravitinoIcebergCatalog extends BaseCatalog * backed catalogs are eligible; a catalog already configured with {@code catalog-backend=rest} or * {@code custom} is left untouched. */ - private Optional<String> resolveIcebergRestUri(Map<String, String> properties) { + static Optional<String> resolveIcebergRestUri( + Map<String, String> properties, + UnaryOperator<String> sessionConfig, + Supplier<Optional<String>> endpointDiscovery) { String backend = properties.get(IcebergConstants.CATALOG_BACKEND); if (backend == null) { return Optional.empty(); @@ -114,12 +123,46 @@ public class GravitinoIcebergCatalog extends BaseCatalog return Optional.empty(); } - String manualUri = - SparkSession.active().conf().get(GravitinoSparkConfig.GRAVITINO_ICEBERG_REST_URI, null); + String routingEnabled = + sessionConfig.apply(GravitinoSparkConfig.GRAVITINO_ICEBERG_REST_ROUTING_ENABLED); + if (StringUtils.isNotBlank(routingEnabled) + && !"true".equalsIgnoreCase(routingEnabled) + && !"false".equalsIgnoreCase(routingEnabled)) { + throw new IllegalArgumentException( + GravitinoSparkConfig.GRAVITINO_ICEBERG_REST_ROUTING_ENABLED + + " must be true or false, but was: " + + routingEnabled); + } + if ("false".equalsIgnoreCase(routingEnabled)) { + return Optional.empty(); + } + + String manualUri = sessionConfig.apply(GravitinoSparkConfig.GRAVITINO_ICEBERG_REST_URI); if (StringUtils.isNotBlank(manualUri)) { return Optional.of(manualUri); } - return GravitinoCatalogManager.get().getIcebergRestUri(); + + Optional<String> discoveredUri; + try { + discoveredUri = endpointDiscovery.get(); + } catch (RuntimeException e) { + throw new IllegalStateException( + "Failed to discover the Iceberg REST endpoint. Configure " + + GravitinoSparkConfig.GRAVITINO_ICEBERG_REST_URI + + ", use a Gravitino server that supports /api/system/iceberg-rest, or set " + + GravitinoSparkConfig.GRAVITINO_ICEBERG_REST_ROUTING_ENABLED + + "=false to use legacy Hive/JDBC backend translation.", + e); + } + if (!discoveredUri.isPresent()) { + throw new IllegalStateException( + "No Iceberg REST endpoint is available. Configure " + + GravitinoSparkConfig.GRAVITINO_ICEBERG_REST_URI + + ", use a Gravitino server that supports /api/system/iceberg-rest, or set " + + GravitinoSparkConfig.GRAVITINO_ICEBERG_REST_ROUTING_ENABLED + + "=false to use legacy Hive/JDBC backend translation."); + } + return discoveredUri; } private Map<String, String> buildAutoRoutedIcebergRestProperties( diff --git a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/iceberg/IcebergRestOAuthConfig.java b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/iceberg/IcebergRestOAuthConfig.java index 92b548ca00..b7ba2e619c 100644 --- a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/iceberg/IcebergRestOAuthConfig.java +++ b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/iceberg/IcebergRestOAuthConfig.java @@ -19,9 +19,10 @@ package org.apache.gravitino.spark.connector.iceberg; -import com.google.common.base.Preconditions; import com.google.common.collect.ImmutableSet; +import java.util.ArrayList; import java.util.HashMap; +import java.util.List; import java.util.Map; import java.util.Set; import org.apache.commons.lang3.StringUtils; @@ -33,13 +34,14 @@ class IcebergRestOAuthConfig { static final String AUTH_TYPE = "rest.auth.type"; static final String AUTH_TYPE_OAUTH2 = "oauth2"; + static final String TOKEN = "token"; static final String CREDENTIAL = "credential"; static final String OAUTH2_SERVER_URI = "oauth2-server-uri"; static final String SCOPE = "scope"; private static final Set<String> LEGACY_OAUTH_PROPERTIES = ImmutableSet.of( - "token", + TOKEN, CREDENTIAL, SCOPE, OAUTH2_SERVER_URI, @@ -52,35 +54,75 @@ class IcebergRestOAuthConfig { static Map<String, String> resolve(SparkConf sparkConf, Map<String, String> explicitRestConfig) { Map<String, String> result = new HashMap<>(explicitRestConfig); - if (hasExplicitAuthentication(result) - || !sparkConf.getBoolean(GravitinoSparkConfig.GRAVITINO_ICEBERG_REUSE_OAUTH2, true)) { + String explicitAuthType = result.get(AUTH_TYPE); + if (StringUtils.isNotBlank(explicitAuthType) + && !AUTH_TYPE_OAUTH2.equalsIgnoreCase(explicitAuthType)) { + return result; + } + + // A bearer token is a complete OAuth2 authentication mechanism by itself. Preserve the legacy + // behavior and do not require client-credential properties alongside it. + if (StringUtils.isNotBlank(result.get(TOKEN))) { + return result; + } + + // Namespaced authentication settings are self-contained, including token-only OAuth2. Do not + // mix the legacy OAuth2 client properties used by the reuse path into that configuration. + if (result.keySet().stream() + .anyMatch(key -> key.startsWith("rest.auth.") && !AUTH_TYPE.equals(key))) { return result; } String authType = sparkConf.get(GravitinoSparkConfig.GRAVITINO_AUTH_TYPE, AuthProperties.SIMPLE_AUTH_TYPE); - if (!AuthProperties.isOAuth2(authType)) { + boolean reuseOAuth2 = + sparkConf.getBoolean(GravitinoSparkConfig.GRAVITINO_ICEBERG_REUSE_OAUTH2, true) + && AuthProperties.isOAuth2(authType); + boolean hasExplicitLegacyOAuth2 = + StringUtils.equalsIgnoreCase(explicitAuthType, AUTH_TYPE_OAUTH2) + || result.keySet().stream().anyMatch(LEGACY_OAUTH_PROPERTIES::contains); + if (!reuseOAuth2 && !hasExplicitLegacyOAuth2) { return result; } - String serverUri = required(sparkConf, GravitinoSparkConfig.GRAVITINO_OAUTH2_URI); - String tokenPath = required(sparkConf, GravitinoSparkConfig.GRAVITINO_OAUTH2_PATH); - result.put(AUTH_TYPE, AUTH_TYPE_OAUTH2); - result.put(CREDENTIAL, required(sparkConf, GravitinoSparkConfig.GRAVITINO_OAUTH2_CREDENTIAL)); - result.put(SCOPE, required(sparkConf, GravitinoSparkConfig.GRAVITINO_OAUTH2_SCOPE)); - result.put(OAUTH2_SERVER_URI, joinUri(serverUri, tokenPath)); + result.putIfAbsent(AUTH_TYPE, AUTH_TYPE_OAUTH2); + if (reuseOAuth2) { + putIfConfigured( + result, + CREDENTIAL, + sparkConf.get(GravitinoSparkConfig.GRAVITINO_OAUTH2_CREDENTIAL, null)); + putIfConfigured( + result, SCOPE, sparkConf.get(GravitinoSparkConfig.GRAVITINO_OAUTH2_SCOPE, null)); + if (!result.containsKey(OAUTH2_SERVER_URI)) { + String serverUri = sparkConf.get(GravitinoSparkConfig.GRAVITINO_OAUTH2_URI, null); + String tokenPath = sparkConf.get(GravitinoSparkConfig.GRAVITINO_OAUTH2_PATH, null); + if (StringUtils.isNotBlank(serverUri) && StringUtils.isNotBlank(tokenPath)) { + result.put(OAUTH2_SERVER_URI, joinUri(serverUri, tokenPath)); + } + } + } + + validateLegacyOAuth2(result); return result; } - private static boolean hasExplicitAuthentication(Map<String, String> restConfig) { - return restConfig.keySet().stream() - .anyMatch(key -> key.startsWith("rest.auth.") || LEGACY_OAUTH_PROPERTIES.contains(key)); + private static void putIfConfigured(Map<String, String> config, String key, String value) { + if (!config.containsKey(key) && StringUtils.isNotBlank(value)) { + config.put(key, value); + } } - private static String required(SparkConf sparkConf, String key) { - String value = sparkConf.get(key, null); - Preconditions.checkArgument(StringUtils.isNotBlank(value), key + " should not be empty"); - return value; + private static void validateLegacyOAuth2(Map<String, String> config) { + List<String> missing = new ArrayList<>(); + for (String key : ImmutableSet.of(AUTH_TYPE, CREDENTIAL, SCOPE, OAUTH2_SERVER_URI)) { + if (StringUtils.isBlank(config.get(key))) { + missing.add(key); + } + } + if (!missing.isEmpty()) { + throw new IllegalArgumentException( + "Incomplete Iceberg REST OAuth2 configuration; missing: " + String.join(", ", missing)); + } } private static String joinUri(String serverUri, String tokenPath) { diff --git a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/iceberg/TestGravitinoIcebergCatalogRestRouting.java b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/iceberg/TestGravitinoIcebergCatalogRestRouting.java new file mode 100644 index 0000000000..412352da06 --- /dev/null +++ b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/iceberg/TestGravitinoIcebergCatalogRestRouting.java @@ -0,0 +1,150 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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. + */ + +package org.apache.gravitino.spark.connector.iceberg; + +import com.google.common.collect.ImmutableMap; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.atomic.AtomicBoolean; +import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants; +import org.apache.gravitino.spark.connector.GravitinoSparkConfig; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +/** Tests Iceberg REST endpoint selection for Spark catalogs. */ +public class TestGravitinoIcebergCatalogRestRouting { + + @Test + void testRoutingDisabledSkipsDiscovery() { + Map<String, String> sessionConfig = + ImmutableMap.of(GravitinoSparkConfig.GRAVITINO_ICEBERG_REST_ROUTING_ENABLED, "false"); + AtomicBoolean discoveryCalled = new AtomicBoolean(); + + Optional<String> result = + GravitinoIcebergCatalog.resolveIcebergRestUri( + hiveProperties(), + sessionConfig::get, + () -> { + discoveryCalled.set(true); + return Optional.of("http://discovered/iceberg";); + }); + + Assertions.assertFalse(result.isPresent()); + Assertions.assertFalse(discoveryCalled.get()); + } + + @Test + void testManualUriTakesPrecedenceOverDiscovery() { + Map<String, String> sessionConfig = + ImmutableMap.of( + GravitinoSparkConfig.GRAVITINO_ICEBERG_REST_URI, "http://configured/iceberg";); + AtomicBoolean discoveryCalled = new AtomicBoolean(); + + Optional<String> result = + GravitinoIcebergCatalog.resolveIcebergRestUri( + hiveProperties(), + sessionConfig::get, + () -> { + discoveryCalled.set(true); + return Optional.of("http://discovered/iceberg";); + }); + + Assertions.assertEquals("http://configured/iceberg";, result.get()); + Assertions.assertFalse(discoveryCalled.get()); + } + + @Test + void testRoutingEnabledRequiresEndpoint() { + IllegalStateException exception = + Assertions.assertThrows( + IllegalStateException.class, + () -> + GravitinoIcebergCatalog.resolveIcebergRestUri( + hiveProperties(), key -> null, Optional::empty)); + + Assertions.assertTrue( + exception + .getMessage() + .contains(GravitinoSparkConfig.GRAVITINO_ICEBERG_REST_ROUTING_ENABLED)); + } + + @Test + void testInvalidRoutingSettingFails() { + Map<String, String> sessionConfig = + ImmutableMap.of(GravitinoSparkConfig.GRAVITINO_ICEBERG_REST_ROUTING_ENABLED, "yes"); + + IllegalArgumentException exception = + Assertions.assertThrows( + IllegalArgumentException.class, + () -> + GravitinoIcebergCatalog.resolveIcebergRestUri( + hiveProperties(), sessionConfig::get, Optional::empty)); + + Assertions.assertTrue( + exception + .getMessage() + .contains(GravitinoSparkConfig.GRAVITINO_ICEBERG_REST_ROUTING_ENABLED)); + } + + @Test + void testDiscoveryFailureIncludesEscapeHatches() { + RuntimeException discoveryFailure = new RuntimeException("connection refused"); + + IllegalStateException exception = + Assertions.assertThrows( + IllegalStateException.class, + () -> + GravitinoIcebergCatalog.resolveIcebergRestUri( + hiveProperties(), + key -> null, + () -> { + throw discoveryFailure; + })); + + Assertions.assertSame(discoveryFailure, exception.getCause()); + Assertions.assertTrue( + exception + .getMessage() + .contains(GravitinoSparkConfig.GRAVITINO_ICEBERG_REST_ROUTING_ENABLED)); + Assertions.assertTrue( + exception.getMessage().contains(GravitinoSparkConfig.GRAVITINO_ICEBERG_REST_URI)); + } + + @Test + void testNonRoutableBackendSkipsDiscovery() { + AtomicBoolean discoveryCalled = new AtomicBoolean(); + + Optional<String> result = + GravitinoIcebergCatalog.resolveIcebergRestUri( + ImmutableMap.of(IcebergConstants.CATALOG_BACKEND, "rest"), + key -> null, + () -> { + discoveryCalled.set(true); + return Optional.of("http://discovered/iceberg";); + }); + + Assertions.assertFalse(result.isPresent()); + Assertions.assertFalse(discoveryCalled.get()); + } + + private static ImmutableMap<String, String> hiveProperties() { + return ImmutableMap.of(IcebergConstants.CATALOG_BACKEND, "hive"); + } +} diff --git a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/iceberg/TestIcebergRestOAuthConfig.java b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/iceberg/TestIcebergRestOAuthConfig.java index eff21194ba..16ac74e454 100644 --- a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/iceberg/TestIcebergRestOAuthConfig.java +++ b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/iceberg/TestIcebergRestOAuthConfig.java @@ -20,7 +20,6 @@ package org.apache.gravitino.spark.connector.iceberg; import com.google.common.collect.ImmutableMap; -import com.google.common.collect.ImmutableSet; import java.util.Collections; import java.util.Map; import org.apache.gravitino.spark.connector.GravitinoSparkConfig; @@ -57,25 +56,72 @@ public class TestIcebergRestOAuthConfig { } @Test - void testLegacyExplicitOAuthPropertiesDisableAutomaticReuse() { + void testExplicitOAuthCredentialOverridesReusedCredential() { SparkConf sparkConf = oauthSparkConf("https://identity.example.com";, "oauth/token"); + Map<String, String> explicit = ImmutableMap.of("credential", "irc:secret"); - for (String property : - ImmutableSet.of( - "token", - "credential", - "scope", - "oauth2-server-uri", - "audience", - "resource", - "token-refresh-enabled", - "token-exchange-enabled")) { - Map<String, String> explicit = ImmutableMap.of(property, "explicit-value"); - - Map<String, String> result = IcebergRestOAuthConfig.resolve(sparkConf, explicit); - - Assertions.assertEquals(explicit, result, property); - } + Map<String, String> result = IcebergRestOAuthConfig.resolve(sparkConf, explicit); + + Assertions.assertEquals("oauth2", result.get(IcebergRestOAuthConfig.AUTH_TYPE)); + Assertions.assertEquals("irc:secret", result.get(IcebergRestOAuthConfig.CREDENTIAL)); + Assertions.assertEquals("openid", result.get(IcebergRestOAuthConfig.SCOPE)); + Assertions.assertEquals( + "https://identity.example.com/oauth/token";, + result.get(IcebergRestOAuthConfig.OAUTH2_SERVER_URI)); + } + + @Test + void testAllExplicitOAuthPropertiesOverrideReusedProperties() { + SparkConf sparkConf = oauthSparkConf("https://identity.example.com";, "oauth/token"); + Map<String, String> explicit = + ImmutableMap.of( + IcebergRestOAuthConfig.AUTH_TYPE, + "oauth2", + IcebergRestOAuthConfig.CREDENTIAL, + "irc:secret", + IcebergRestOAuthConfig.SCOPE, + "irc-scope", + IcebergRestOAuthConfig.OAUTH2_SERVER_URI, + "https://irc-identity.example.com/token";); + + Map<String, String> result = IcebergRestOAuthConfig.resolve(sparkConf, explicit); + + Assertions.assertEquals(explicit, result); + } + + @Test + void testPartialOAuthConfigFailsWhenReuseIsUnavailable() { + SparkConf sparkConf = new SparkConf(false); + Map<String, String> explicit = ImmutableMap.of("credential", "irc:secret"); + + IllegalArgumentException exception = + Assertions.assertThrows( + IllegalArgumentException.class, + () -> IcebergRestOAuthConfig.resolve(sparkConf, explicit)); + + Assertions.assertTrue(exception.getMessage().contains("scope")); + Assertions.assertTrue(exception.getMessage().contains("oauth2-server-uri")); + } + + @Test + void testNamespacedOAuthTokenIsKeptIndependent() { + SparkConf sparkConf = oauthSparkConf("https://identity.example.com";, "oauth/token"); + Map<String, String> explicit = + ImmutableMap.of("rest.auth.type", "oauth2", "rest.auth.oauth2.token", "irc-token"); + + Map<String, String> result = IcebergRestOAuthConfig.resolve(sparkConf, explicit); + + Assertions.assertEquals(explicit, result); + } + + @Test + void testLegacyBearerTokenIsKeptIndependent() { + SparkConf sparkConf = oauthSparkConf("https://identity.example.com";, "oauth/token"); + Map<String, String> explicit = ImmutableMap.of("token", "irc-token"); + + Map<String, String> result = IcebergRestOAuthConfig.resolve(sparkConf, explicit); + + Assertions.assertEquals(explicit, result); } @Test
