This is an automated email from the ASF dual-hosted git repository.
diqiu50 pushed a commit to branch branch-1.3
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/branch-1.3 by this push:
new 0b99555a41 [Cherry-pick to branch-1.3] [#12709]
improvement(spark-connector): Route lakehouse-iceberg catalogs through the
Iceberg REST server (#12710) (#12864)
0b99555a41 is described below
commit 0b99555a410f11966528446e193bb49d4945a37c
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Fri Sep 4 17:22:52 2026 +0800
[Cherry-pick to branch-1.3] [#12709] improvement(spark-connector): Route
lakehouse-iceberg catalogs through the Iceberg REST server (#12710) (#12864)
**Cherry-pick Information:**
- Original commit: 5b666b322b0da5f2ded3e0f23591c3e8f71e0ea8
- Target branch: `branch-1.3`
- Status: ⚠️ **Has conflicts - manual resolution required**
---------
Co-authored-by: Yuhui <[email protected]>
Co-authored-by: Claude Sonnet 5 <[email protected]>
Co-authored-by: diqiu50 <[email protected]>
---
.../workflows/spark-integration-test-action.yml | 2 +-
docs/spark-connector/spark-catalog-iceberg.md | 99 ++++++++-
docs/spark-connector/spark-connector.md | 20 +-
.../spark/connector/GravitinoSparkConfig.java | 24 ++
.../connector/catalog/GravitinoCatalogManager.java | 32 ++-
.../connector/iceberg/GravitinoIcebergCatalog.java | 154 ++++++++++++-
.../iceberg/IcebergPropertiesConstants.java | 18 ++
.../iceberg/IcebergPropertiesConverter.java | 205 +++++++++++++++++
.../connector/iceberg/IcebergRestOAuthConfig.java | 135 +++++++++++
.../connector/plugin/GravitinoDriverPlugin.java | 1 +
.../catalog/TestGravitinoCatalogManager.java | 107 +++++++++
.../connector/glue/TestGravitinoGlueCatalog.java | 2 +-
.../TestGravitinoIcebergCatalogRestRouting.java | 246 +++++++++++++++++++++
.../iceberg/TestIcebergPropertiesConverter.java | 190 ++++++++++++++++
.../iceberg/TestIcebergRestOAuthConfig.java | 205 +++++++++++++++++
.../connector/integration/test/SparkEnvIT.java | 71 ++++--
.../iceberg/SparkIcebergCatalogHiveBackendIT.java | 10 +
.../iceberg/SparkIcebergCatalogRestRoutingIT.java | 76 +++++++
.../SparkIcebergCatalogRestRoutingIT33.java | 62 ++++++
.../SparkIcebergCatalogRestRoutingIT34.java | 25 +++
spark-connector/v3.5/spark/build.gradle.kts | 2 +
.../SparkIcebergCatalogRestRoutingIT35.java | 25 +++
...kIcebergCatalogRestS3CredentialVendingIT35.java | 239 ++++++++++++++++++++
23 files changed, 1914 insertions(+), 36 deletions(-)
diff --git a/.github/workflows/spark-integration-test-action.yml
b/.github/workflows/spark-integration-test-action.yml
index c0f7a1e712..39d679fda9 100644
--- a/.github/workflows/spark-integration-test-action.yml
+++ b/.github/workflows/spark-integration-test-action.yml
@@ -25,7 +25,7 @@ jobs:
start-runner:
name: JDK${{ inputs.java-version }}-${{ inputs.test-mode }}-Scala${{
inputs.scala-version }}
runs-on: ubuntu-latest
- timeout-minutes: 60
+ timeout-minutes: 120
env:
PLATFORM: ${{ inputs.architecture }}
steps:
diff --git a/docs/spark-connector/spark-catalog-iceberg.md
b/docs/spark-connector/spark-catalog-iceberg.md
index fd3210e4eb..056c70dbd4 100644
--- a/docs/spark-connector/spark-catalog-iceberg.md
+++ b/docs/spark-connector/spark-catalog-iceberg.md
@@ -148,13 +148,110 @@ Gravitino catalog property names with the prefix
`spark.bypass.` are passed to S
Iceberg catalog property `cache-enabled` is setting to `false` internally and
not allowed to change.
:::
+## Routing Through the Gravitino Iceberg REST Server
+
+If the Gravitino server exposes an [Iceberg REST
catalog](../iceberg-rest-service.md) (IRC) endpoint for the
+current metalake, the Spark connector automatically routes `hive` and `jdbc`
backed Iceberg catalogs through
+that endpoint instead of talking to the Hive metastore or JDBC database
directly. This has no effect on
+catalogs whose `catalog-backend` is already `rest` or `custom`.
+
+Routing through the IRC server is the only way to receive short-lived,
per-table **vended credentials**
+that Iceberg's native REST protocol refreshes automatically. The non-REST path
can still inject a single
+vended credential fetched once at catalog initialization (see
+[Credential vending](../security/credential-vending.md)), but it is not
refreshed per table access.
+
+REST routing is enabled by default. The endpoint is discovered once per Spark
application — for the life
+of the Gravitino Spark plugin, not per `SparkSession` — and is not re-checked
afterward.
+
+- If no discoverable endpoint is found (for example, the `iceberg-rest`
auxiliary service is disabled or
+ not configured with `catalog-config-provider=dynamic-config-provider`),
catalog initialization fails
+ with an actionable error. Set
`spark.sql.gravitino.iceberg.rest-routing-enabled=false` to retain the
+ native Hive/JDBC backend instead.
+- A catalog whose warehouse uses a scheme with a native Iceberg FileIO
(`s3://`, `gs://`, `abfs://`, etc.)
+ must have [credential vending](../security/credential-vending.md) configured
(`credential-providers`)
+ before it can be routed: routing replaces any static storage credentials for
that FileIO with vended
+ ones, and without credential vending the catalog would lose storage access.
Catalog initialization
+ fails with an actionable error if this is not configured. Set
`rest-routing-enabled=false` for that
+ catalog to keep using the legacy Hive/JDBC backend instead.
+
+To force a specific endpoint instead of relying on auto-discovery, set:
+
+```properties
+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:
+
+```properties
+spark.sql.gravitino.iceberg.rest.rest.auth.type basic
+spark.sql.gravitino.iceberg.rest.rest.auth.basic.username <username>
+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. This validation also
applies when reusing the Gravitino
+configuration by default with no IRC-specific override at all: if the reused
configuration itself is
+incomplete, 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.
+
+:::caution
+Spark's UI redacts environment values whose property name matches
`secret|password|token`, which does not
+match `credential`. Both `spark.sql.gravitino.oauth2.credential` and
`spark.sql.gravitino.iceberg.rest.credential`
+are shown in plain text on the Spark UI's environment page; set
`spark.redaction.regex` to also match
+`credential` if this is a concern.
+:::
+
+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`,
+`abfs`/`abfss`/`wasb`/`wasbs`, `oss`) unless `io-impl` is already set
explicitly on the catalog.
+
## Storage
Spark connector could convert storage properties in the Gravitino catalog to
Spark Iceberg connector automatically, No extra configuration is needed for
`S3`, `ADLS`, `OSS`, `GCS`.
### 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 5a3d8562c1..36bd7e384e 100644
--- a/docs/spark-connector/spark-connector.md
+++ b/docs/spark-connector/spark-connector.md
@@ -26,14 +26,18 @@ The Apache Gravitino Spark connector leverages the Spark
DataSourceV2 interface
1. [Build](../how-to-build.md) or download the package
([gravitino-spark-connector-runtime-3.3](https://mvnrepository.com/artifact/org.apache.gravitino/gravitino-spark-connector-runtime-3.3),
[gravitino-spark-connector-runtime-3.4](https://mvnrepository.com/artifact/org.apache.gravitino/gravitino-spark-connector-runtime-3.4),
[gravitino-spark-connector-runtime-3.5](https://mvnrepository.com/artifact/org.apache.gravitino/gravitino-spark-connector-runtime-3.5)),
and place it to the classpa [...]
2. Configure the Spark session to use the Gravitino spark connector.
-| Property | Type | Default Value |
Description
| Required |
-|------------------------------------------|--------|---------------|-------------------------------------------------------------------------------------------------|----------|
-| spark.plugins | string | (none) |
Gravitino spark plugin name,
`org.apache.gravitino.spark.connector.plugin.GravitinoSparkPlugin` | Yes |
-| spark.sql.gravitino.metalake | string | (none) | The
metalake name that spark connector used to request to Gravitino.
| Yes |
-| spark.sql.gravitino.uri | string | (none) | The uri
of Gravitino server address.
| Yes |
-| 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 |
+| Property | Type | Default Value |
Description
| Required |
+|---------------------------------------------------|---------|---------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------|----------|
+| spark.plugins | string | (none) |
Gravitino spark plugin name,
`org.apache.gravitino.spark.connector.plugin.GravitinoSparkPlugin`
| Yes |
+| spark.sql.gravitino.metalake | string | (none) |
The metalake name that spark connector used to request to Gravitino.
| Yes |
+| spark.sql.gravitino.uri | string | (none) |
The uri of Gravitino server address.
| Yes |
+| 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 be108c9af1..e67c122176 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
@@ -32,6 +32,30 @@ public class GravitinoSparkConfig {
GRAVITINO_PREFIX + "enablePaimonSupport";
public static final String GRAVITINO_CLIENT_CONFIG_PREFIX = GRAVITINO_PREFIX
+ "client.";
+ /**
+ * Manually overrides the Gravitino Iceberg REST server endpoint that
hive/jdbc backed
+ * 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 are 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 =
+ GRAVITINO_PREFIX + "iceberg.reuseOAuth2";
+
+ /**
+ * Pass-through prefix for the Iceberg REST client config (e.g.
rest.auth.type,
+ * rest.auth.basic.username), applied when a catalog is routed through the
Iceberg REST server.
+ */
+ public static final String GRAVITINO_ICEBERG_REST_CONFIG_PREFIX =
+ GRAVITINO_PREFIX + "iceberg.rest.";
+
public static final String GRAVITINO_AUTH_TYPE =
GRAVITINO_PREFIX + AuthProperties.GRAVITINO_CLIENT_AUTH_TYPE;
public static final String GRAVITINO_OAUTH2_URI =
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 a484972656..9c5a4569bf 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
@@ -24,6 +24,7 @@ import com.google.common.base.Preconditions;
import com.google.common.base.Supplier;
import java.util.Arrays;
import java.util.Map;
+import java.util.Optional;
import org.apache.gravitino.Catalog;
import org.apache.gravitino.client.GravitinoClient;
import org.slf4j.Logger;
@@ -35,19 +36,26 @@ public class GravitinoCatalogManager {
private static GravitinoCatalogManager gravitinoCatalogManager;
private volatile boolean isClosed = false;
+ private final String metalakeName;
private final Cache<String, Catalog> gravitinoCatalogs;
private final GravitinoClient gravitinoClient;
- private GravitinoCatalogManager(Supplier<GravitinoClient> clientBuilder) {
+ // Resolved lazily on first access and cached for the life of this manager;
Spark catalogs are
+ // initialized once, so there is no refresh path if the Iceberg REST
endpoint changes later.
+ private volatile Optional<String> icebergRestUri;
+
+ private GravitinoCatalogManager(String metalakeName,
Supplier<GravitinoClient> clientBuilder) {
+ this.metalakeName = metalakeName;
this.gravitinoClient = clientBuilder.get();
// Will not evict catalog by default
this.gravitinoCatalogs = Caffeine.newBuilder().build();
}
- public static GravitinoCatalogManager create(Supplier<GravitinoClient>
clientBuilder) {
+ public static GravitinoCatalogManager create(
+ String metalakeName, Supplier<GravitinoClient> clientBuilder) {
Preconditions.checkState(
gravitinoCatalogManager == null, "Should not create duplicate
GravitinoCatalogManager");
- gravitinoCatalogManager = new GravitinoCatalogManager(clientBuilder);
+ gravitinoCatalogManager = new GravitinoCatalogManager(metalakeName,
clientBuilder);
return gravitinoCatalogManager;
}
@@ -86,6 +94,24 @@ public class GravitinoCatalogManager {
return gravitinoCatalogs.asMap();
}
+ /**
+ * Resolves the Gravitino Iceberg REST server endpoint for this manager's
metalake, if the server
+ * 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
+ */
+ public Optional<String> getIcebergRestUri() {
+ if (icebergRestUri == null) {
+ synchronized (this) {
+ if (icebergRestUri == null) {
+ icebergRestUri = gravitinoClient.icebergRestServiceUri(metalakeName);
+ }
+ }
+ }
+ return icebergRestUri;
+ }
+
private Catalog loadCatalog(String catalogName) {
Catalog catalog = gravitinoClient.loadCatalog(catalogName);
Preconditions.checkArgument(
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 0fa99d3ab7..b97cdeeef4 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
@@ -21,21 +21,33 @@ package org.apache.gravitino.spark.connector.iceberg;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
+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;
import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants;
import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergPropertiesUtils;
+import org.apache.gravitino.credential.CredentialConstants;
import org.apache.gravitino.credential.CredentialPropertyUtils;
import org.apache.gravitino.rel.Table;
+import org.apache.gravitino.spark.connector.GravitinoSparkConfig;
import org.apache.gravitino.spark.connector.PropertiesConverter;
import org.apache.gravitino.spark.connector.SparkTransformConverter;
import org.apache.gravitino.spark.connector.SparkTypeConverter;
import org.apache.gravitino.spark.connector.catalog.BaseCatalog;
+import org.apache.gravitino.spark.connector.catalog.GravitinoCatalogManager;
import org.apache.iceberg.catalog.Catalog;
import org.apache.iceberg.spark.SparkCatalog;
import org.apache.iceberg.spark.procedures.SparkProcedures;
import org.apache.iceberg.spark.source.HasIcebergCatalog;
import org.apache.iceberg.spark.source.SparkTable;
+import org.apache.spark.SparkConf;
+import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.catalyst.analysis.NoSuchFunctionException;
import org.apache.spark.sql.catalyst.analysis.NoSuchNamespaceException;
import org.apache.spark.sql.catalyst.analysis.NoSuchProcedureException;
@@ -72,15 +84,149 @@ public class GravitinoIcebergCatalog extends BaseCatalog
}
}
String catalogBackendName =
IcebergPropertiesUtils.getCatalogBackendName(properties);
- Map<String, String> all =
- getPropertiesConverter().toSparkCatalogProperties(options, properties);
- CredentialPropertyUtils.applyIcebergCredentials(
- CredentialPropertyUtils.getCredentials(gravitinoCatalogClient), all);
+ SparkConf sparkConf = SparkSession.active().sparkContext().conf();
+ Optional<String> icebergRestUri =
+ resolveIcebergRestUri(
+ properties,
+ key -> sparkConf.get(key, null),
+ () -> GravitinoCatalogManager.get().getIcebergRestUri());
+ Map<String, String> all;
+ if (icebergRestUri.isPresent()) {
+ all =
+ buildAutoRoutedIcebergRestProperties(
+ name, options, properties, icebergRestUri.get(), sparkConf);
+ } else {
+ all = getPropertiesConverter().toSparkCatalogProperties(options,
properties);
+ CredentialPropertyUtils.applyIcebergCredentials(
+ CredentialPropertyUtils.getCredentials(gravitinoCatalogClient), all);
+ }
TableCatalog icebergCatalog = new SparkCatalog();
icebergCatalog.initialize(catalogBackendName, new
CaseInsensitiveStringMap(all));
return icebergCatalog;
}
+ /**
+ * Resolves the Iceberg REST server endpoint to route this catalog through,
if any. Only hive/jdbc
+ * backed catalogs are eligible; a catalog already configured with {@code
catalog-backend=rest} or
+ * {@code custom} is left untouched, and so is a catalog with no {@code
catalog-backend} property
+ * at all.
+ *
+ * <p>An eligible catalog whose warehouse has a native Iceberg FileIO
(s3/gs/abfs-family schemes)
+ * fails immediately unless {@code credential-providers} is configured:
routing replaces any
+ * static storage credentials for that FileIO with vended ones, so such a
catalog would silently
+ * lose storage access once routed. A warehouse scheme with no native FileIO
(e.g. {@code
+ * hdfs://}, {@code file://}) carries no such risk and is unaffected.
+ */
+ 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();
+ }
+ String normalizedBackend = backend.toLowerCase(Locale.ROOT);
+ boolean eligible =
+
IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_BACKEND_HIVE.equals(normalizedBackend)
+ ||
IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_BACKEND_JDBC.equals(
+ normalizedBackend);
+ if (!eligible) {
+ return Optional.empty();
+ }
+
+ 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();
+ }
+
+ boolean warehouseHasNativeFileIo =
+
IcebergPropertiesConverter.deriveFileIoImpl(properties.get(IcebergConstants.WAREHOUSE))
+ != null;
+ if (warehouseHasNativeFileIo
+ &&
StringUtils.isBlank(properties.get(CredentialConstants.CREDENTIAL_PROVIDERS))) {
+ throw new IllegalStateException(
+ "Catalog's warehouse has a native Iceberg FileIO but no
credential-providers "
+ + "configured; routing through the Iceberg REST server replaces
any static storage "
+ + "credentials for that FileIO with vended ones, so this catalog
would lose storage "
+ + "access once routed. Configure credential-providers on the
catalog, or set "
+ + GravitinoSparkConfig.GRAVITINO_ICEBERG_REST_ROUTING_ENABLED
+ + "=false to use legacy Hive/JDBC backend translation.");
+ }
+
+ String manualUri =
sessionConfig.apply(GravitinoSparkConfig.GRAVITINO_ICEBERG_REST_URI);
+ if (StringUtils.isNotBlank(manualUri)) {
+ return Optional.of(manualUri);
+ }
+
+ Optional<String> discoveredUri;
+ try {
+ discoveredUri = endpointDiscovery.get();
+ } catch (RuntimeException e) {
+ // endpointDiscovery can throw for reasons other than connectivity (e.g.
a bug in client
+ // bootstrapping), so the exception's own type is included rather than
assuming it is
+ // always a reachability/config problem.
+ throw new IllegalStateException(
+ "Failed to discover the Iceberg REST endpoint ("
+ + e.getClass().getSimpleName()
+ + "). 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(
+ String gravitinoCatalogName,
+ CaseInsensitiveStringMap options,
+ Map<String, String> properties,
+ String restUri,
+ SparkConf sparkConf) {
+ IcebergPropertiesConverter converter = (IcebergPropertiesConverter)
getPropertiesConverter();
+ Map<String, String> all =
+ new HashMap<>(
+ converter.buildIcebergRestProperties(
+ gravitinoCatalogName,
+ restUri,
+ properties,
+ getAutoRoutedIcebergRestClientConfig(sparkConf)));
+ if (options != null) {
+ all.putAll(options);
+ // options can re-introduce a reserved routing key (e.g. a Spark-level
`uri`/`prefix`
+ // catalog option); re-derive them so options can never redirect a
routed catalog.
+ converter.reapplyReservedRestProperties(gravitinoCatalogName, restUri,
all);
+ }
+ return all;
+ }
+
+ static Map<String, String> getAutoRoutedIcebergRestClientConfig(SparkConf
sparkConf) {
+ Map<String, String> explicitRestConfig =
+ Stream.of(
+ sparkConf.getAllWithPrefix(
+ GravitinoSparkConfig.GRAVITINO_ICEBERG_REST_CONFIG_PREFIX))
+ .collect(Collectors.toMap(t -> t._1, t -> t._2, (oldVal, newVal)
-> newVal));
+ return IcebergRestOAuthConfig.resolve(sparkConf, explicitRestConfig);
+ }
+
@Override
protected org.apache.spark.sql.connector.catalog.Table createSparkTable(
Identifier identifier,
diff --git
a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/iceberg/IcebergPropertiesConstants.java
b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/iceberg/IcebergPropertiesConstants.java
index 3d6ea418b4..dbd6292951 100644
---
a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/iceberg/IcebergPropertiesConstants.java
+++
b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/iceberg/IcebergPropertiesConstants.java
@@ -81,5 +81,23 @@ public class IcebergPropertiesConstants {
static final String GRAVITINO_ICEBERG_CATALOG_BACKEND_NAME =
IcebergConstants.CATALOG_BACKEND_NAME;
+ // Only used when routing a hive/jdbc backed catalog through the Gravitino
Iceberg REST server;
+ // not part of the Gravitino <-> Iceberg property mapping in
IcebergPropertiesUtils.
+ static final String ICEBERG_REST_CATALOG_PREFIX = "prefix";
+
+ static final String ICEBERG_IO_IMPL = IcebergConstants.IO_IMPL;
+ static final String ICEBERG_S3_ENDPOINT =
IcebergConstants.ICEBERG_S3_ENDPOINT;
+ static final String ICEBERG_S3_PATH_STYLE_ACCESS =
IcebergConstants.ICEBERG_S3_PATH_STYLE_ACCESS;
+ static final String ICEBERG_AWS_S3_REGION = IcebergConstants.AWS_S3_REGION;
+ static final String ICEBERG_OSS_ENDPOINT =
IcebergConstants.ICEBERG_OSS_ENDPOINT;
+
+ static final String ICEBERG_S3_FILE_IO_IMPL =
"org.apache.iceberg.aws.s3.S3FileIO";
+ static final String ICEBERG_GCS_FILE_IO_IMPL =
"org.apache.iceberg.gcp.gcs.GCSFileIO";
+ static final String ICEBERG_ADLS_FILE_IO_IMPL =
"org.apache.iceberg.azure.adlsv2.ADLSFileIO";
+ static final String ICEBERG_OSS_FILE_IO_IMPL =
"org.apache.iceberg.aliyun.oss.OSSFileIO";
+
+ static final String ICEBERG_ACCESS_DELEGATION =
IcebergConstants.ICEBERG_ACCESS_DELEGATION;
+ static final String ICEBERG_ACCESS_DELEGATION_VENDED_CREDENTIALS =
"vended-credentials";
+
private IcebergPropertiesConstants() {}
}
diff --git
a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/iceberg/IcebergPropertiesConverter.java
b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/iceberg/IcebergPropertiesConverter.java
index 9f7011aed5..f8fa15eb3a 100644
---
a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/iceberg/IcebergPropertiesConverter.java
+++
b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/iceberg/IcebergPropertiesConverter.java
@@ -20,17 +20,34 @@
package org.apache.gravitino.spark.connector.iceberg;
import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
import java.util.Map;
import org.apache.commons.lang3.StringUtils;
import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergCatalogBackend;
import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants;
import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergPropertiesUtils;
+import org.apache.gravitino.credential.CredentialConstants;
import org.apache.gravitino.spark.connector.PropertiesConverter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/** Transform Apache Iceberg catalog properties between Apache Spark and
Apache Gravitino. */
public class IcebergPropertiesConverter implements PropertiesConverter {
+ private static final Logger LOG =
LoggerFactory.getLogger(IcebergPropertiesConverter.class);
+
+ /** Routing keys the connector always derives itself; they cannot be
overridden. */
+ private static final List<String> RESERVED_REST_PROPERTIES =
+ ImmutableList.of(
+ IcebergPropertiesConstants.ICEBERG_CATALOG_TYPE,
+ IcebergPropertiesConstants.ICEBERG_CATALOG_URI,
+ IcebergPropertiesConstants.ICEBERG_CATALOG_WAREHOUSE,
+ IcebergPropertiesConstants.ICEBERG_REST_CATALOG_PREFIX);
+
public static class IcebergPropertiesConverterHolder {
private static final IcebergPropertiesConverter INSTANCE = new
IcebergPropertiesConverter();
}
@@ -68,6 +85,194 @@ public class IcebergPropertiesConverter implements
PropertiesConverter {
return all;
}
+ /**
+ * Builds Spark Iceberg catalog properties that route requests through the
Gravitino Iceberg REST
+ * server, regardless of the catalog's actual storage backend (hive/jdbc).
This is the only path
+ * on which temporary credentials work: the Iceberg REST protocol vends a
fresh credential per
+ * table access, so static secrets are intentionally not carried over from
{@code
+ * gravitinoProperties}. Static-key credentials for the non-REST path are
unaffected; they are
+ * still injected via {@link
org.apache.gravitino.credential.CredentialPropertyUtils}.
+ *
+ * @param gravitinoCatalogName the Gravitino catalog name, used as both
{@code warehouse} (for the
+ * initial catalog discovery request) and {@code prefix} (for every
subsequent request path
+ * segment) so the REST server resolves the same Gravitino catalog
+ * @param restUri the Iceberg REST server endpoint to route through,
resolved by the caller
+ * @param gravitinoProperties the Gravitino catalog properties
+ * @param icebergRestClientConfig operator-level Iceberg REST client config
(e.g. {@code
+ * rest.auth.type}), applied after catalog-level {@code spark.bypass.}
overrides since it is a
+ * cluster-wide operational setting
+ * @return the Spark Iceberg catalog properties
+ */
+ Map<String, String> buildIcebergRestProperties(
+ String gravitinoCatalogName,
+ String restUri,
+ Map<String, String> gravitinoProperties,
+ Map<String, String> icebergRestClientConfig) {
+ Preconditions.checkArgument(StringUtils.isNotBlank(restUri), "restUri
should not be blank");
+
+ Map<String, String> all = new HashMap<>();
+ // Later put/putAll calls override earlier ones for the same key; this is
call-order
+ // precedence, unrelated to HashMap's (unspecified) iteration order.
+ all.putAll(buildStorageProperties(gravitinoProperties));
+ all.put(
+ IcebergPropertiesConstants.ICEBERG_ACCESS_DELEGATION,
+
IcebergPropertiesConstants.ICEBERG_ACCESS_DELEGATION_VENDED_CREDENTIALS);
+ // The catalog's own spark.bypass properties override the defaults above,
so a renamed
+ // Iceberg client property can be worked around without a connector change.
+ all.putAll(extractSparkBypassProperties(gravitinoProperties));
+ all.putAll(icebergRestClientConfig);
+
+ reapplyReservedRestProperties(gravitinoCatalogName, restUri, all);
+ all.put(IcebergPropertiesConstants.ICEBERG_CATALOG_CACHE_ENABLED, "FALSE");
+ return all;
+ }
+
+ /**
+ * Re-derives the reserved routing keys ({@code type}/{@code uri}/{@code
warehouse}/{@code
+ * prefix}) on {@code all}, so that a source merged in after {@link
#buildIcebergRestProperties}
+ * (e.g. Spark catalog {@code options}) can never redirect a routed catalog.
+ *
+ * @param gravitinoCatalogName the Gravitino catalog name
+ * @param restUri the Iceberg REST server endpoint to route through
+ * @param all the properties to re-derive the reserved keys on, mutated in
place
+ */
+ void reapplyReservedRestProperties(
+ String gravitinoCatalogName, String restUri, Map<String, String> all) {
+ Map<String, String> derived =
+ ImmutableMap.of(
+ IcebergPropertiesConstants.ICEBERG_CATALOG_TYPE,
+ IcebergPropertiesConstants.ICEBERG_CATALOG_BACKEND_REST,
+ IcebergPropertiesConstants.ICEBERG_CATALOG_URI,
+ restUri,
+ IcebergPropertiesConstants.ICEBERG_CATALOG_WAREHOUSE,
+ gravitinoCatalogName,
+ IcebergPropertiesConstants.ICEBERG_REST_CATALOG_PREFIX,
+ gravitinoCatalogName);
+ warnOnReservedRestPropertyOverrides(gravitinoCatalogName, all, derived);
+ all.putAll(derived);
+ }
+
+ private Map<String, String> extractSparkBypassProperties(Map<String, String>
properties) {
+ Map<String, String> bypass = new HashMap<>();
+ if (properties != null) {
+ properties.forEach(
+ (k, v) -> {
+ if (k.startsWith(SPARK_PROPERTY_PREFIX)) {
+ bypass.put(k.substring(SPARK_PROPERTY_PREFIX.length()), v);
+ }
+ });
+ }
+ return bypass;
+ }
+
+ private void warnOnReservedRestPropertyOverrides(
+ String gravitinoCatalogName, Map<String, String> config, Map<String,
String> derived) {
+ for (String reserved : RESERVED_REST_PROPERTIES) {
+ String current = config.get(reserved);
+ // A reserved key already set to its own derived value (from a prior
call on the same
+ // catalog) is not an override; only a value that actually differs is
worth warning about.
+ if (current != null && !current.equals(derived.get(reserved))) {
+ LOG.warn(
+ "Property '{}' set on catalog '{}' is ignored; the connector
always derives it when "
+ + "routing through the Iceberg REST server.",
+ reserved,
+ gravitinoCatalogName);
+ }
+ }
+ }
+
+ /**
+ * Derives the storage-related Iceberg client config from the catalog's
warehouse location. Only
+ * non-secret settings (native FileIO impl, custom endpoint, region,
path-style access) are
+ * carried over; static access keys and JDBC credentials are deliberately
excluded since the REST
+ * path vends its own temporary credentials.
+ */
+ private Map<String, String> buildStorageProperties(Map<String, String>
gravitinoProperties) {
+ Map<String, String> icebergProperties =
+ IcebergPropertiesUtils.toIcebergCatalogProperties(gravitinoProperties);
+ Map<String, String> storageProperties = new HashMap<>();
+ copyIfPresent(icebergProperties,
IcebergPropertiesConstants.ICEBERG_IO_IMPL, storageProperties);
+
+ String warehouse = gravitinoProperties.get(IcebergConstants.WAREHOUSE);
+ String fileIoImpl = deriveFileIoImpl(warehouse);
+ if (fileIoImpl != null) {
+
storageProperties.putIfAbsent(IcebergPropertiesConstants.ICEBERG_IO_IMPL,
fileIoImpl);
+ } else if
(!storageProperties.containsKey(IcebergPropertiesConstants.ICEBERG_IO_IMPL)) {
+ warnOnSchemeWithoutNativeFileIo(gravitinoProperties, warehouse);
+ }
+
+ copyIfPresent(
+ icebergProperties, IcebergPropertiesConstants.ICEBERG_S3_ENDPOINT,
storageProperties);
+ copyIfPresent(
+ icebergProperties, IcebergPropertiesConstants.ICEBERG_AWS_S3_REGION,
storageProperties);
+ copyIfPresent(
+ icebergProperties,
+ IcebergPropertiesConstants.ICEBERG_S3_PATH_STYLE_ACCESS,
+ storageProperties);
+ copyIfPresent(
+ icebergProperties, IcebergPropertiesConstants.ICEBERG_OSS_ENDPOINT,
storageProperties);
+ return storageProperties;
+ }
+
+ /**
+ * Derives the native Iceberg FileIO implementation for a warehouse
location, or {@code null} if
+ * the scheme has none (e.g. {@code hdfs://}, {@code file://}).
Package-private so the routing
+ * decision can check whether a warehouse has a native FileIO without
duplicating the scheme list.
+ */
+ static String deriveFileIoImpl(String warehouse) {
+ if (StringUtils.isBlank(warehouse) || !warehouse.contains("://")) {
+ return null;
+ }
+ String scheme = StringUtils.substringBefore(warehouse,
"://").toLowerCase(Locale.ROOT);
+ switch (scheme) {
+ case "s3":
+ case "s3a":
+ case "s3n":
+ return IcebergPropertiesConstants.ICEBERG_S3_FILE_IO_IMPL;
+ case "gs":
+ return IcebergPropertiesConstants.ICEBERG_GCS_FILE_IO_IMPL;
+ case "abfs":
+ case "abfss":
+ case "wasb":
+ case "wasbs":
+ return IcebergPropertiesConstants.ICEBERG_ADLS_FILE_IO_IMPL;
+ case "oss":
+ return IcebergPropertiesConstants.ICEBERG_OSS_FILE_IO_IMPL;
+ default:
+ return null;
+ }
+ }
+
+ /**
+ * Warns when a warehouse scheme has no native Iceberg FileIO but the
catalog vends credentials.
+ * Vended credentials are only consumed by Iceberg's native FileIO
implementations, so table
+ * access would otherwise fail at read time with a storage authentication
error far from its
+ * cause, unless {@code io-impl} is set explicitly on the catalog.
+ */
+ private void warnOnSchemeWithoutNativeFileIo(
+ Map<String, String> gravitinoProperties, String warehouse) {
+ if (StringUtils.isBlank(warehouse) || !warehouse.contains("://")) {
+ // hdfs and file legitimately have no native FileIO here, and neither
vends credentials.
+ return;
+ }
+ if
(StringUtils.isBlank(gravitinoProperties.get(CredentialConstants.CREDENTIAL_PROVIDERS)))
{
+ return;
+ }
+ LOG.warn(
+ "Warehouse '{}' has no native Iceberg FileIO for the credentials
vended by '{}' to be "
+ + "applied to; table access may fail to authenticate unless
'io-impl' is set "
+ + "explicitly. Schemes with a native FileIO: s3/s3a/s3n, gs,
abfs/abfss/wasb/wasbs, oss.",
+ warehouse,
+ gravitinoProperties.get(CredentialConstants.CREDENTIAL_PROVIDERS));
+ }
+
+ private void copyIfPresent(Map<String, String> source, String key,
Map<String, String> target) {
+ String value = source.get(key);
+ if (StringUtils.isNotBlank(value)) {
+ target.put(key, value);
+ }
+ }
+
@Override
public Map<String, String> toGravitinoTableProperties(Map<String, String>
properties) {
return new HashMap<>(properties);
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
new file mode 100644
index 0000000000..6d419c3c4b
--- /dev/null
+++
b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/iceberg/IcebergRestOAuthConfig.java
@@ -0,0 +1,135 @@
+/*
+ * 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.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;
+import org.apache.gravitino.auth.AuthProperties;
+import org.apache.gravitino.spark.connector.GravitinoSparkConfig;
+import org.apache.spark.SparkConf;
+
+/**
+ * Derives the Iceberg REST client's OAuth2 configuration, reusing Gravitino's
own OAuth2 client
+ * settings by default. When auth is explicitly set to a non-OAuth2 type, a
bearer {@code token} is
+ * supplied, or any other namespaced {@code rest.auth.*} property is present,
the caller-supplied
+ * config is treated as already complete and passed through unvalidated; {@link
+ * #validateLegacyOAuth2} only runs on the reuse-from-Gravitino /
explicit-legacy-property path.
+ */
+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";
+
+ // Properties whose presence alone implies the user is explicitly
configuring legacy OAuth2
+ // (triggering validation), as opposed to optional extras that ride along
with it. TOKEN is not
+ // included here: it already short-circuits to a plain pass-through above.
+ private static final Set<String> OAUTH2_TRIGGER_PROPERTIES =
ImmutableSet.of(CREDENTIAL);
+
+ private IcebergRestOAuthConfig() {}
+
+ static Map<String, String> resolve(SparkConf sparkConf, Map<String, String>
explicitRestConfig) {
+ Map<String, String> result = new HashMap<>(explicitRestConfig);
+ 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);
+ 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(OAUTH2_TRIGGER_PROPERTIES::contains);
+ if (!reuseOAuth2 && !hasExplicitLegacyOAuth2) {
+ return result;
+ }
+
+ 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 void putIfConfigured(Map<String, String> config, String key,
String value) {
+ if (!config.containsKey(key) && StringUtils.isNotBlank(value)) {
+ config.put(key, value);
+ }
+ }
+
+ // scope is intentionally not required: Iceberg itself defaults it to
"catalog". The server URI
+ // is required rather than left to Iceberg's own default, so the client
credential is never
+ // posted to an unintended token endpoint.
+ private static void validateLegacyOAuth2(Map<String, String> config) {
+ List<String> missing = new ArrayList<>();
+ for (String key : ImmutableSet.of(AUTH_TYPE, CREDENTIAL,
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) {
+ return StringUtils.removeEnd(serverUri, "/") + "/" +
StringUtils.removeStart(tokenPath, "/");
+ }
+}
diff --git
a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/plugin/GravitinoDriverPlugin.java
b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/plugin/GravitinoDriverPlugin.java
index ea0e03aa9d..59636ec0ef 100644
---
a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/plugin/GravitinoDriverPlugin.java
+++
b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/plugin/GravitinoDriverPlugin.java
@@ -112,6 +112,7 @@ public class GravitinoDriverPlugin implements DriverPlugin {
this.catalogManager =
GravitinoCatalogManager.create(
+ metalake,
() ->
createGravitinoClient(
gravitinoUri, metalake, conf, sc.sparkUser(),
gravitinoClientConfig));
diff --git
a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/catalog/TestGravitinoCatalogManager.java
b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/catalog/TestGravitinoCatalogManager.java
new file mode 100644
index 0000000000..d91034066a
--- /dev/null
+++
b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/catalog/TestGravitinoCatalogManager.java
@@ -0,0 +1,107 @@
+/*
+ * 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.catalog;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.gravitino.client.GravitinoClient;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+/** Verifies GravitinoCatalogManager's lazy caching of the Iceberg REST
endpoint discovery. */
+public class TestGravitinoCatalogManager {
+
+ private static final String METALAKE_NAME = "test_metalake";
+
+ @AfterEach
+ void closeManager() {
+ try {
+ GravitinoCatalogManager.get().close();
+ } catch (IllegalStateException e) {
+ // The test closed the manager itself.
+ }
+ }
+
+ @Test
+ void testIcebergRestUriIsCachedAfterSuccessfulDiscovery() {
+ AtomicInteger discoveryCalls = new AtomicInteger();
+ GravitinoCatalogManager manager =
+ createManager(
+ invocation -> {
+ discoveryCalls.incrementAndGet();
+ return Optional.of("http://irc:9001/iceberg";);
+ });
+
+ Optional<String> first = manager.getIcebergRestUri();
+ Optional<String> second = manager.getIcebergRestUri();
+
+ assertEquals(Optional.of("http://irc:9001/iceberg";), first);
+ assertSame(first, second, "A cached discovery result must not be
recomputed");
+ assertEquals(1, discoveryCalls.get());
+ }
+
+ @Test
+ void testNoIcebergRestUriDiscoveredIsAlsoCached() {
+ AtomicInteger discoveryCalls = new AtomicInteger();
+ GravitinoCatalogManager manager =
+ createManager(
+ invocation -> {
+ discoveryCalls.incrementAndGet();
+ return Optional.empty();
+ });
+
+ manager.getIcebergRestUri();
+ manager.getIcebergRestUri();
+
+ assertEquals(1, discoveryCalls.get(), "A negative discovery result must be
cached too");
+ }
+
+ @Test
+ void testIcebergRestUriDiscoveryFailurePropagatesAndIsNotCached() {
+ AtomicInteger discoveryCalls = new AtomicInteger();
+ GravitinoCatalogManager manager =
+ createManager(
+ invocation -> {
+ if (discoveryCalls.incrementAndGet() == 1) {
+ throw new RuntimeException("Gravitino server unreachable");
+ }
+ return Optional.of("http://irc:9001/iceberg";);
+ });
+
+ assertThrows(RuntimeException.class, manager::getIcebergRestUri);
+ Optional<String> second = manager.getIcebergRestUri();
+
+ assertEquals(Optional.of("http://irc:9001/iceberg";), second);
+ assertEquals(2, discoveryCalls.get(), "A failed discovery must be retried
on the next call");
+ }
+
+ private GravitinoCatalogManager createManager(
+ org.mockito.stubbing.Answer<Optional<String>> answer) {
+ GravitinoClient client = mock(GravitinoClient.class);
+ when(client.icebergRestServiceUri(anyString())).thenAnswer(answer);
+ return GravitinoCatalogManager.create(METALAKE_NAME, () -> client);
+ }
+}
diff --git
a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/glue/TestGravitinoGlueCatalog.java
b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/glue/TestGravitinoGlueCatalog.java
index af980f7309..c30265f089 100644
---
a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/glue/TestGravitinoGlueCatalog.java
+++
b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/glue/TestGravitinoGlueCatalog.java
@@ -66,7 +66,7 @@ public class TestGravitinoGlueCatalog {
// GravitinoGlueCatalog extends BaseCatalog which calls
GravitinoCatalogManager.get()
// in its constructor, so we must initialize the manager first.
GravitinoClient mockClient = mock(GravitinoClient.class);
- GravitinoCatalogManager.create(() -> mockClient);
+ GravitinoCatalogManager.create("test_metalake", () -> mockClient);
}
@AfterAll
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..5261ab73b9
--- /dev/null
+++
b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/iceberg/TestGravitinoIcebergCatalogRestRouting.java
@@ -0,0 +1,246 @@
+/*
+ * 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.credential.CredentialConstants;
+import org.apache.gravitino.spark.connector.GravitinoSparkConfig;
+import org.apache.spark.SparkConf;
+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();
+
+ // Routing disabled must skip even the credential-providers check: a
catalog with static
+ // credentials is exactly what stays on the legacy path here.
+ 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 testEligibleCatalogWithNativeFileIoAndNoCredentialProvidersFails() {
+ IllegalStateException exception =
+ Assertions.assertThrows(
+ IllegalStateException.class,
+ () ->
+ GravitinoIcebergCatalog.resolveIcebergRestUri(
+ s3HiveProperties(), key -> null, Optional::empty));
+
+ Assertions.assertTrue(
+
exception.getMessage().contains(CredentialConstants.CREDENTIAL_PROVIDERS));
+ }
+
+ @Test
+ void testEligibleCatalogWithoutNativeFileIoSkipsCredentialProvidersCheck() {
+ // hdfs:// has no native Iceberg FileIO, so routing carries no risk of
dropping static
+ // credentials even without credential-providers configured.
+ Optional<String> result =
+ GravitinoIcebergCatalog.resolveIcebergRestUri(
+ hiveProperties(), key -> null, () ->
Optional.of("http://discovered/iceberg";));
+
+ Assertions.assertEquals("http://discovered/iceberg";, result.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(
+ hivePropertiesWithCredentialProviders(),
+ sessionConfig::get,
+ () -> {
+ discoveryCalled.set(true);
+ return Optional.of("http://discovered/iceberg";);
+ });
+
+ Assertions.assertEquals("http://configured/iceberg";, result.get());
+ Assertions.assertFalse(discoveryCalled.get());
+ }
+
+ @Test
+ void testNoEndpointFailsWhenRoutingNotExplicitlySet() {
+ IllegalStateException exception =
+ Assertions.assertThrows(
+ IllegalStateException.class,
+ () ->
+ GravitinoIcebergCatalog.resolveIcebergRestUri(
+ hivePropertiesWithCredentialProviders(), key -> null,
Optional::empty));
+
+ Assertions.assertTrue(
+ exception
+ .getMessage()
+
.contains(GravitinoSparkConfig.GRAVITINO_ICEBERG_REST_ROUTING_ENABLED));
+ }
+
+ @Test
+ void testNoEndpointFailsWhenRoutingExplicitlyEnabled() {
+ Map<String, String> sessionConfig =
+
ImmutableMap.of(GravitinoSparkConfig.GRAVITINO_ICEBERG_REST_ROUTING_ENABLED,
"true");
+
+ IllegalStateException exception =
+ Assertions.assertThrows(
+ IllegalStateException.class,
+ () ->
+ GravitinoIcebergCatalog.resolveIcebergRestUri(
+ hivePropertiesWithCredentialProviders(),
sessionConfig::get, 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 testDiscoveryFailureFailsWhenRoutingNotExplicitlySet() {
+ RuntimeException discoveryFailure = new RuntimeException("connection
refused");
+
+ IllegalStateException exception =
+ Assertions.assertThrows(
+ IllegalStateException.class,
+ () ->
+ GravitinoIcebergCatalog.resolveIcebergRestUri(
+ hivePropertiesWithCredentialProviders(),
+ key -> null,
+ () -> {
+ throw discoveryFailure;
+ }));
+
+ Assertions.assertSame(discoveryFailure, exception.getCause());
+ }
+
+ @Test
+ void testDiscoveryFailureFailsWhenRoutingExplicitlyEnabled() {
+ Map<String, String> sessionConfig =
+
ImmutableMap.of(GravitinoSparkConfig.GRAVITINO_ICEBERG_REST_ROUTING_ENABLED,
"true");
+ RuntimeException discoveryFailure = new RuntimeException("connection
refused");
+
+ IllegalStateException exception =
+ Assertions.assertThrows(
+ IllegalStateException.class,
+ () ->
+ GravitinoIcebergCatalog.resolveIcebergRestUri(
+ hivePropertiesWithCredentialProviders(),
+ sessionConfig::get,
+ () -> {
+ 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());
+ }
+
+ @Test
+ void testAutoRoutedRestClientConfigStripsPrefixFromRealSparkConf() {
+ SparkConf sparkConf = new SparkConf(false);
+ sparkConf.set(
+ GravitinoSparkConfig.GRAVITINO_ICEBERG_REST_CONFIG_PREFIX +
"rest.auth.type", "basic");
+ sparkConf.set(
+ GravitinoSparkConfig.GRAVITINO_ICEBERG_REST_CONFIG_PREFIX +
"rest.auth.basic.username",
+ "admin");
+
+ Map<String, String> result =
+
GravitinoIcebergCatalog.getAutoRoutedIcebergRestClientConfig(sparkConf);
+
+ Assertions.assertEquals("basic", result.get("rest.auth.type"));
+ Assertions.assertEquals("admin", result.get("rest.auth.basic.username"));
+ }
+
+ private static ImmutableMap<String, String> hiveProperties() {
+ return ImmutableMap.of(IcebergConstants.CATALOG_BACKEND, "hive");
+ }
+
+ private static ImmutableMap<String, String>
hivePropertiesWithCredentialProviders() {
+ return ImmutableMap.of(
+ IcebergConstants.CATALOG_BACKEND,
+ "hive",
+ CredentialConstants.CREDENTIAL_PROVIDERS,
+ "s3-token");
+ }
+
+ private static ImmutableMap<String, String> s3HiveProperties() {
+ return ImmutableMap.of(
+ IcebergConstants.CATALOG_BACKEND, "hive",
+ IcebergConstants.WAREHOUSE, "s3://bucket/path");
+ }
+}
diff --git
a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/iceberg/TestIcebergPropertiesConverter.java
b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/iceberg/TestIcebergPropertiesConverter.java
index 643a977cdb..d314a1ed99 100644
---
a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/iceberg/TestIcebergPropertiesConverter.java
+++
b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/iceberg/TestIcebergPropertiesConverter.java
@@ -20,6 +20,7 @@
package org.apache.gravitino.spark.connector.iceberg;
import com.google.common.collect.ImmutableMap;
+import java.util.HashMap;
import java.util.Map;
import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergCatalogBackend;
import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants;
@@ -140,4 +141,193 @@ public class TestIcebergPropertiesConverter {
"custom-warehouse"),
properties);
}
+
+ @Test
+ void testIcebergRestPropertiesFromHiveBackendExcludesStaticCredentials() {
+ Map<String, String> gravitinoProperties =
+ ImmutableMap.of(
+ IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_BACKEND,
+ IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_BACKEND_HIVE,
+ IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_URI,
+ "thrift://hive-metastore:9083",
+ IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_WAREHOUSE,
+ "s3://bucket/warehouse",
+ "s3-access-key-id",
+ "AKIDEXAMPLE",
+ "s3-secret-access-key",
+ "secret");
+
+ Map<String, String> properties =
+ icebergPropertiesConverter.buildIcebergRestProperties(
+ "my_catalog", "http://gravitino:9001/iceberg";,
gravitinoProperties, ImmutableMap.of());
+
+ Assertions.assertEquals(
+ "rest",
properties.get(IcebergPropertiesConstants.ICEBERG_CATALOG_TYPE));
+ Assertions.assertEquals(
+ "http://gravitino:9001/iceberg";,
+ properties.get(IcebergPropertiesConstants.ICEBERG_CATALOG_URI));
+ Assertions.assertEquals(
+ "my_catalog",
properties.get(IcebergPropertiesConstants.ICEBERG_CATALOG_WAREHOUSE));
+ Assertions.assertEquals("my_catalog", properties.get("prefix"));
+ Assertions.assertEquals(
+ "vended-credentials",
properties.get("header.X-Iceberg-Access-Delegation"));
+ Assertions.assertEquals(
+ "org.apache.iceberg.aws.s3.S3FileIO",
properties.get(IcebergConstants.IO_IMPL));
+
+ // The REST protocol vends its own credentials, so no static secret should
be present.
+
Assertions.assertFalse(properties.containsKey(IcebergConstants.ICEBERG_S3_ACCESS_KEY_ID));
+
Assertions.assertFalse(properties.containsKey(IcebergConstants.ICEBERG_S3_SECRET_ACCESS_KEY));
+
Assertions.assertFalse(properties.containsKey(IcebergConstants.ICEBERG_JDBC_USER));
+
Assertions.assertFalse(properties.containsKey(IcebergConstants.ICEBERG_JDBC_PASSWORD));
+ }
+
+ @Test
+ void testIcebergRestPropertiesFromJdbcBackendExcludesStaticCredentials() {
+ Map<String, String> gravitinoProperties =
+ ImmutableMap.of(
+ IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_BACKEND,
+ IcebergPropertiesConstants.ICEBERG_CATALOG_BACKEND_JDBC,
+ IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_URI,
+ "jdbc:postgresql://db:5432/iceberg",
+ IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_JDBC_USER,
+ "user",
+ IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_JDBC_PASSWORD,
+ "passwd");
+
+ Map<String, String> properties =
+ icebergPropertiesConverter.buildIcebergRestProperties(
+ "my_catalog", "http://gravitino:9001/iceberg";,
gravitinoProperties, ImmutableMap.of());
+
+ Assertions.assertEquals(
+ "rest",
properties.get(IcebergPropertiesConstants.ICEBERG_CATALOG_TYPE));
+
Assertions.assertFalse(properties.containsKey(IcebergConstants.ICEBERG_JDBC_USER));
+
Assertions.assertFalse(properties.containsKey(IcebergConstants.ICEBERG_JDBC_PASSWORD));
+ }
+
+ @Test
+ void testIcebergRestPropertiesRespectsExplicitIoImpl() {
+ Map<String, String> gravitinoProperties =
+ ImmutableMap.of(
+ IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_BACKEND,
+ IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_BACKEND_HIVE,
+ IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_WAREHOUSE,
+ "s3://bucket/warehouse",
+ IcebergConstants.IO_IMPL,
+ "com.example.CustomFileIO");
+
+ Map<String, String> properties =
+ icebergPropertiesConverter.buildIcebergRestProperties(
+ "my_catalog", "http://gravitino:9001/iceberg";,
gravitinoProperties, ImmutableMap.of());
+
+ Assertions.assertEquals("com.example.CustomFileIO",
properties.get(IcebergConstants.IO_IMPL));
+ }
+
+ @Test
+ void testIcebergRestPropertiesDerivesNativeFileIoPerScheme() {
+ Assertions.assertEquals(
+ "org.apache.iceberg.gcp.gcs.GCSFileIO",
fileIoForWarehouse("gs://bucket/warehouse"));
+ Assertions.assertEquals(
+ "org.apache.iceberg.azure.adlsv2.ADLSFileIO",
+
fileIoForWarehouse("abfss://[email protected]/warehouse"));
+ Assertions.assertEquals(
+ "org.apache.iceberg.aliyun.oss.OSSFileIO",
fileIoForWarehouse("oss://bucket/warehouse"));
+ }
+
+ private String fileIoForWarehouse(String warehouse) {
+ Map<String, String> gravitinoProperties =
+ ImmutableMap.of(
+ IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_BACKEND,
+ IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_BACKEND_HIVE,
+ IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_WAREHOUSE,
+ warehouse);
+ Map<String, String> properties =
+ icebergPropertiesConverter.buildIcebergRestProperties(
+ "my_catalog", "http://gravitino:9001/iceberg";,
gravitinoProperties, ImmutableMap.of());
+ return properties.get(IcebergConstants.IO_IMPL);
+ }
+
+ @Test
+ void testIcebergRestPropertiesWarehouseSchemeWithoutNativeFileIo() {
+ Map<String, String> gravitinoProperties =
+ ImmutableMap.of(
+ IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_BACKEND,
+ IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_BACKEND_HIVE,
+ IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_WAREHOUSE,
+ "hdfs://namenode/warehouse");
+
+ Map<String, String> properties =
+ icebergPropertiesConverter.buildIcebergRestProperties(
+ "my_catalog", "http://gravitino:9001/iceberg";,
gravitinoProperties, ImmutableMap.of());
+
+ Assertions.assertFalse(properties.containsKey(IcebergConstants.IO_IMPL));
+ }
+
+ @Test
+ void testIcebergRestPropertiesBypassAndReservedKeyOverride() {
+ Map<String, String> gravitinoProperties =
+ ImmutableMap.of(
+ IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_BACKEND,
+ IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_BACKEND_HIVE,
+ "spark.bypass.some-custom-property",
+ "custom-value",
+ // Attempting to override a reserved routing key via spark.bypass
must be ignored.
+ "spark.bypass." + IcebergPropertiesConstants.ICEBERG_CATALOG_URI,
+ "http://attacker-controlled/iceberg";);
+
+ Map<String, String> properties =
+ icebergPropertiesConverter.buildIcebergRestProperties(
+ "my_catalog", "http://gravitino:9001/iceberg";,
gravitinoProperties, ImmutableMap.of());
+
+ Assertions.assertEquals("custom-value",
properties.get("some-custom-property"));
+ Assertions.assertEquals(
+ "http://gravitino:9001/iceberg";,
+ properties.get(IcebergPropertiesConstants.ICEBERG_CATALOG_URI));
+ }
+
+ @Test
+ void testReapplyReservedRestPropertiesUndoesSparkOptionsOverride() {
+ Map<String, String> gravitinoProperties =
+ ImmutableMap.of(
+ IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_BACKEND,
+ IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_BACKEND_HIVE);
+ Map<String, String> properties =
+ new HashMap<>(
+ icebergPropertiesConverter.buildIcebergRestProperties(
+ "my_catalog",
+ "http://gravitino:9001/iceberg";,
+ gravitinoProperties,
+ ImmutableMap.of()));
+
+ // Simulate a Spark-level catalog option redirecting the reserved routing
keys, the way
+ // buildAutoRoutedIcebergRestProperties merges `options` in after
buildIcebergRestProperties.
+ properties.put(IcebergPropertiesConstants.ICEBERG_CATALOG_URI,
"http://attacker-controlled";);
+ properties.put(IcebergPropertiesConstants.ICEBERG_REST_CATALOG_PREFIX,
"other_catalog");
+
+ icebergPropertiesConverter.reapplyReservedRestProperties(
+ "my_catalog", "http://gravitino:9001/iceberg";, properties);
+
+ Assertions.assertEquals(
+ "http://gravitino:9001/iceberg";,
+ properties.get(IcebergPropertiesConstants.ICEBERG_CATALOG_URI));
+ Assertions.assertEquals(
+ "my_catalog",
properties.get(IcebergPropertiesConstants.ICEBERG_REST_CATALOG_PREFIX));
+ }
+
+ @Test
+ void testIcebergRestPropertiesRestClientConfigPassthrough() {
+ Map<String, String> gravitinoProperties =
+ ImmutableMap.of(
+ IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_BACKEND,
+ IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_BACKEND_HIVE);
+
+ Map<String, String> properties =
+ icebergPropertiesConverter.buildIcebergRestProperties(
+ "my_catalog",
+ "http://gravitino:9001/iceberg";,
+ gravitinoProperties,
+ ImmutableMap.of("rest.auth.type", "basic",
"rest.auth.basic.username", "admin"));
+
+ Assertions.assertEquals("basic", properties.get("rest.auth.type"));
+ Assertions.assertEquals("admin",
properties.get("rest.auth.basic.username"));
+ }
}
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
new file mode 100644
index 0000000000..e4fae252e3
--- /dev/null
+++
b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/iceberg/TestIcebergRestOAuthConfig.java
@@ -0,0 +1,205 @@
+/*
+ * 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.Collections;
+import java.util.Map;
+import org.apache.gravitino.spark.connector.GravitinoSparkConfig;
+import org.apache.spark.SparkConf;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+/** Tests automatic reuse of Gravitino OAuth2 client settings for Iceberg REST
catalogs. */
+public class TestIcebergRestOAuthConfig {
+
+ @Test
+ void testDerivesIcebergOAuthConfig() {
+ SparkConf sparkConf = oauthSparkConf("https://identity.example.com/";,
"/oauth/token");
+
+ Map<String, String> result = IcebergRestOAuthConfig.resolve(sparkConf,
Collections.emptyMap());
+
+ Assertions.assertEquals("oauth2",
result.get(IcebergRestOAuthConfig.AUTH_TYPE));
+ Assertions.assertEquals("alice: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 testExplicitRestAuthenticationTakesPrecedence() {
+ SparkConf sparkConf = oauthSparkConf("https://identity.example.com";,
"oauth/token");
+ Map<String, String> explicit =
+ ImmutableMap.of("rest.auth.type", "basic", "rest.auth.basic.username",
"admin");
+
+ Map<String, String> result = IcebergRestOAuthConfig.resolve(sparkConf,
explicit);
+
+ Assertions.assertEquals(explicit, result);
+ }
+
+ @Test
+ void testExplicitOAuthCredentialOverridesReusedCredential() {
+ SparkConf sparkConf = oauthSparkConf("https://identity.example.com";,
"oauth/token");
+ Map<String, String> explicit = ImmutableMap.of("credential", "irc:secret");
+
+ 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));
+
+ // scope is not required: Iceberg itself defaults it. oauth2-server-uri is
still required.
+
Assertions.assertTrue(exception.getMessage().contains("oauth2-server-uri"));
+ }
+
+ @Test
+ void testOAuthConfigWithoutScopeSucceeds() {
+ SparkConf sparkConf = new SparkConf(false);
+ Map<String, String> explicit =
+ ImmutableMap.of(
+ "credential",
+ "irc:secret",
+ "oauth2-server-uri",
+ "https://irc-identity.example.com/token";);
+
+ 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(
+ "https://irc-identity.example.com/token";,
+ result.get(IcebergRestOAuthConfig.OAUTH2_SERVER_URI));
+ Assertions.assertFalse(result.containsKey(IcebergRestOAuthConfig.SCOPE));
+ }
+
+ @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
+ void testNonAuthenticationRestPropertiesDoNotDisableAutomaticReuse() {
+ SparkConf sparkConf = oauthSparkConf("https://identity.example.com";,
"oauth/token");
+
+ Map<String, String> result =
+ IcebergRestOAuthConfig.resolve(
+ sparkConf, ImmutableMap.of("header.X-Iceberg-Custom", "value"));
+
+ Assertions.assertEquals("value", result.get("header.X-Iceberg-Custom"));
+ Assertions.assertEquals("oauth2",
result.get(IcebergRestOAuthConfig.AUTH_TYPE));
+ Assertions.assertEquals("alice:secret",
result.get(IcebergRestOAuthConfig.CREDENTIAL));
+ }
+
+ @Test
+ void testCanDisableOAuthReuse() {
+ SparkConf sparkConf = oauthSparkConf("https://identity.example.com";,
"oauth/token");
+ sparkConf.set(GravitinoSparkConfig.GRAVITINO_ICEBERG_REUSE_OAUTH2,
"false");
+
+ Map<String, String> result = IcebergRestOAuthConfig.resolve(sparkConf,
Collections.emptyMap());
+
+ Assertions.assertTrue(result.isEmpty());
+ }
+
+ @Test
+ void testDoesNotDeriveConfigForOtherAuthenticationTypes() {
+ SparkConf sparkConf = new SparkConf(false);
+ sparkConf.set(GravitinoSparkConfig.GRAVITINO_AUTH_TYPE, "simple");
+
+ Map<String, String> result = IcebergRestOAuthConfig.resolve(sparkConf,
Collections.emptyMap());
+
+ Assertions.assertTrue(result.isEmpty());
+ }
+
+ @Test
+ void testReuseFailsWhenGravitinoOAuthConfigIsIncomplete() {
+ SparkConf sparkConf = new SparkConf(false);
+ sparkConf.set(GravitinoSparkConfig.GRAVITINO_AUTH_TYPE, "oauth2");
+ sparkConf.set(GravitinoSparkConfig.GRAVITINO_OAUTH2_CREDENTIAL,
"alice:secret");
+ // GRAVITINO_OAUTH2_URI/PATH are intentionally left unset, so
oauth2-server-uri can't be
+ // derived: this is the default-reuse path, not an explicit override, and
must still validate.
+
+ IllegalArgumentException exception =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> IcebergRestOAuthConfig.resolve(sparkConf,
Collections.emptyMap()));
+
+
Assertions.assertTrue(exception.getMessage().contains("oauth2-server-uri"));
+ }
+
+ private SparkConf oauthSparkConf(String serverUri, String tokenPath) {
+ SparkConf sparkConf = new SparkConf(false);
+ sparkConf.set(GravitinoSparkConfig.GRAVITINO_AUTH_TYPE, "oauth2");
+ sparkConf.set(GravitinoSparkConfig.GRAVITINO_OAUTH2_URI, serverUri);
+ sparkConf.set(GravitinoSparkConfig.GRAVITINO_OAUTH2_PATH, tokenPath);
+ sparkConf.set(GravitinoSparkConfig.GRAVITINO_OAUTH2_CREDENTIAL,
"alice:secret");
+ sparkConf.set(GravitinoSparkConfig.GRAVITINO_OAUTH2_SCOPE, "openid");
+ return sparkConf;
+ }
+}
diff --git
a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/SparkEnvIT.java
b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/SparkEnvIT.java
index fcd2c2d819..e5191a7d2b 100644
---
a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/SparkEnvIT.java
+++
b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/SparkEnvIT.java
@@ -28,6 +28,7 @@ import java.util.HashMap;
import java.util.Map;
import org.apache.gravitino.Catalog;
import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants;
import org.apache.gravitino.client.GravitinoMetalake;
import org.apache.gravitino.function.FunctionDefinition;
import org.apache.gravitino.function.FunctionDefinitions;
@@ -84,6 +85,27 @@ public abstract class SparkEnvIT extends SparkUtilIT {
return true;
}
+ /**
+ * Whether the Iceberg REST auxiliary service should run with {@code
dynamic-config-provider}
+ * (routing requests to whichever Gravitino catalog the request's {@code
prefix} names) instead of
+ * the default single-catalog {@code static-config-provider}. Tests that
verify the Spark
+ * connector's Iceberg REST auto-discovery/routing need this.
+ */
+ protected boolean useDynamicIcebergRestConfigProvider() {
+ return false;
+ }
+
+ /**
+ * Extra Spark session configs applied on top of the base set in {@link
#initSparkEnv()}. Tests
+ * that intentionally exercise a non-REST Iceberg catalog backend without a
discoverable Iceberg
+ * REST endpoint (e.g. {@code useDynamicIcebergRestConfigProvider() ==
false}) must disable
+ * Iceberg REST routing here, since routing is otherwise required and fails
catalog initialization
+ * when no endpoint is discoverable.
+ */
+ protected Map<String, String> getExtraSparkConfigs() {
+ return Collections.emptyMap();
+ }
+
/** Returns the Gravitino {@link Catalog} for the catalog under test. */
protected Catalog getGravitinoCatalog() {
return client.loadMetalake(metalakeName).loadCatalog(getCatalogName());
@@ -232,24 +254,36 @@ public abstract class SparkEnvIT extends SparkUtilIT {
private void initIcebergRestServiceEnv() {
super.ignoreIcebergAuxRestService = false;
Map<String, String> icebergRestServiceConfigs = new HashMap<>();
- icebergRestServiceConfigs.put(
- "gravitino."
- + icebergRestServiceName
- + "."
- + IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_BACKEND,
- IcebergPropertiesConstants.ICEBERG_CATALOG_BACKEND_HIVE);
- icebergRestServiceConfigs.put(
- "gravitino."
- + icebergRestServiceName
- + "."
- + IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_URI,
- hiveMetastoreUri);
- icebergRestServiceConfigs.put(
- "gravitino."
- + icebergRestServiceName
- + "."
- + IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_WAREHOUSE,
- warehouse);
+ if (useDynamicIcebergRestConfigProvider()) {
+ icebergRestServiceConfigs.put(
+ "gravitino."
+ + icebergRestServiceName
+ + "."
+ + IcebergConstants.ICEBERG_REST_CATALOG_CONFIG_PROVIDER,
+ IcebergConstants.DYNAMIC_ICEBERG_CATALOG_CONFIG_PROVIDER_NAME);
+ icebergRestServiceConfigs.put(
+ "gravitino." + icebergRestServiceName + "." +
IcebergConstants.GRAVITINO_METALAKE,
+ metalakeName);
+ } else {
+ icebergRestServiceConfigs.put(
+ "gravitino."
+ + icebergRestServiceName
+ + "."
+ + IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_BACKEND,
+ IcebergPropertiesConstants.ICEBERG_CATALOG_BACKEND_HIVE);
+ icebergRestServiceConfigs.put(
+ "gravitino."
+ + icebergRestServiceName
+ + "."
+ + IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_URI,
+ hiveMetastoreUri);
+ icebergRestServiceConfigs.put(
+ "gravitino."
+ + icebergRestServiceName
+ + "."
+ + IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_WAREHOUSE,
+ warehouse);
+ }
registerCustomConfigs(icebergRestServiceConfigs);
}
@@ -280,6 +314,7 @@ public abstract class SparkEnvIT extends SparkUtilIT {
.set("hive.exec.dynamic.partition.mode", "nonstrict")
.set("spark.sql.warehouse.dir", warehouse)
.set("spark.sql.session.timeZone", TIME_ZONE_UTC);
+ getExtraSparkConfigs().forEach(sparkConf::set);
sparkSession =
SparkSession.builder()
.master("local[1]")
diff --git
a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/iceberg/SparkIcebergCatalogHiveBackendIT.java
b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/iceberg/SparkIcebergCatalogHiveBackendIT.java
index c160f97103..fd1edd6784 100644
---
a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/iceberg/SparkIcebergCatalogHiveBackendIT.java
+++
b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/iceberg/SparkIcebergCatalogHiveBackendIT.java
@@ -18,9 +18,11 @@
*/
package org.apache.gravitino.spark.connector.integration.test.iceberg;
+import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Maps;
import java.util.Map;
import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants;
+import org.apache.gravitino.spark.connector.GravitinoSparkConfig;
import org.apache.gravitino.spark.connector.iceberg.IcebergPropertiesConstants;
import org.junit.jupiter.api.Tag;
@@ -28,6 +30,14 @@ import org.junit.jupiter.api.Tag;
@Tag("gravitino-docker-test")
public abstract class SparkIcebergCatalogHiveBackendIT extends
SparkIcebergCatalogIT {
+ @Override
+ protected Map<String, String> getExtraSparkConfigs() {
+ // This class deliberately exercises the legacy Hive backend without a
discoverable Iceberg
+ // REST endpoint (useDynamicIcebergRestConfigProvider() is false), so
routing must be disabled
+ // explicitly.
+ return
ImmutableMap.of(GravitinoSparkConfig.GRAVITINO_ICEBERG_REST_ROUTING_ENABLED,
"false");
+ }
+
@Override
protected Map<String, String> getCatalogConfigs() {
Map<String, String> catalogProperties = Maps.newHashMap();
diff --git
a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/iceberg/SparkIcebergCatalogRestRoutingIT.java
b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/iceberg/SparkIcebergCatalogRestRoutingIT.java
new file mode 100644
index 0000000000..80b726429b
--- /dev/null
+++
b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/iceberg/SparkIcebergCatalogRestRoutingIT.java
@@ -0,0 +1,76 @@
+/*
+ * 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.integration.test.iceberg;
+
+import com.google.common.collect.Maps;
+import java.util.Map;
+import org.apache.gravitino.spark.connector.iceberg.GravitinoIcebergCatalog;
+import org.apache.gravitino.spark.connector.iceberg.IcebergPropertiesConstants;
+import org.apache.iceberg.catalog.Catalog;
+import org.apache.spark.sql.connector.catalog.CatalogPlugin;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.DisabledIf;
+
+/**
+ * This class configures the Gravitino Iceberg catalog with a Hive backend,
but runs the Iceberg
+ * REST auxiliary service in {@code dynamic-config-provider} mode so the Spark
connector
+ * auto-discovers it and routes the catalog through the Iceberg REST protocol
instead of talking to
+ * the Hive metastore directly. All test cases inherited from {@link
SparkIcebergCatalogIT} exercise
+ * this REST-routed path end to end.
+ */
+@Tag("gravitino-docker-test")
+// Spark connector uses a low Iceberg version, couldn't work with Iceberg REST
server with high
+// Iceberg version in embedded mode.
+@DisabledIf("org.apache.gravitino.integration.test.util.ITUtils#isEmbedded")
+public abstract class SparkIcebergCatalogRestRoutingIT extends
SparkIcebergCatalogIT {
+
+ @Override
+ protected boolean useDynamicIcebergRestConfigProvider() {
+ return true;
+ }
+
+ @Override
+ protected Map<String, String> getCatalogConfigs() {
+ Map<String, String> catalogProperties = Maps.newHashMap();
+ catalogProperties.put(
+ IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_BACKEND,
+ IcebergPropertiesConstants.ICEBERG_CATALOG_BACKEND_HIVE);
+ catalogProperties.put(
+ IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_WAREHOUSE,
warehouse);
+ catalogProperties.put(
+ IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_URI,
hiveMetastoreUri);
+ return catalogProperties;
+ }
+
+ @Test
+ void testCatalogIsRoutedThroughIcebergRestServer() {
+ CatalogPlugin catalogPlugin =
+
getSparkSession().sessionState().catalogManager().catalog(getCatalogName());
+ Assertions.assertInstanceOf(GravitinoIcebergCatalog.class, catalogPlugin);
+
+ Catalog icebergCatalog = ((GravitinoIcebergCatalog)
catalogPlugin).icebergCatalog();
+ Assertions.assertEquals(
+ "org.apache.iceberg.rest.RESTCatalog",
+ icebergCatalog.getClass().getName(),
+ "The hive-backed catalog should have been routed through the
discovered Iceberg REST "
+ + "server, instead of Iceberg's native HiveCatalog.");
+ }
+}
diff --git
a/spark-connector/v3.3/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/iceberg/SparkIcebergCatalogRestRoutingIT33.java
b/spark-connector/v3.3/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/iceberg/SparkIcebergCatalogRestRoutingIT33.java
new file mode 100644
index 0000000000..181cf6a0bc
--- /dev/null
+++
b/spark-connector/v3.3/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/iceberg/SparkIcebergCatalogRestRoutingIT33.java
@@ -0,0 +1,62 @@
+/*
+ * 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.integration.test.iceberg;
+
+import
org.apache.gravitino.spark.connector.integration.test.util.SparkMetadataColumnInfo;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.StructField;
+import org.junit.jupiter.api.condition.DisabledIf;
+
+/**
+ * Spark 3.3 integration tests for Iceberg REST routing.
+ *
+ * <p>Spark 3.3 pins Iceberg to 1.8.x ({@code iceberg4spark33}); Gravitino's
lakehouse Iceberg stack
+ * is 1.11. {@link DisabledIf} skips embedded mode to avoid classpath
conflicts. Deploy-mode tests
+ * run a 1.8.x client against Gravitino's 1.11 REST server and cover basic
catalog CRUD only; they
+ * do not exercise Iceberg 1.11 REST changes such as {@code file-scan-tasks}
scan planning or the
+ * configurable multi-level namespace separator contract.
+ */
+@DisabledIf("org.apache.gravitino.integration.test.util.ITUtils#isEmbedded")
+public class SparkIcebergCatalogRestRoutingIT33 extends
SparkIcebergCatalogRestRoutingIT {
+ @Override
+ protected boolean supportsFunction() {
+ // Spark 3.3 does not support function operations
+ return false;
+ }
+
+ /**
+ * Spark 3.3 uses Iceberg 1.8.x; its {@code SparkTable#metadataColumns()}
exposes five columns,
+ * not the seven row-lineage columns added in Iceberg 1.11.
+ */
+ @Override
+ protected SparkMetadataColumnInfo[] getIcebergMetadataColumns() {
+ return new SparkMetadataColumnInfo[] {
+ new SparkMetadataColumnInfo("_spec_id", DataTypes.IntegerType, false),
+ new SparkMetadataColumnInfo(
+ "_partition",
+ DataTypes.createStructType(
+ new StructField[] {DataTypes.createStructField("name",
DataTypes.StringType, true)}),
+ true),
+ new SparkMetadataColumnInfo("_file", DataTypes.StringType, false),
+ new SparkMetadataColumnInfo("_pos", DataTypes.LongType, false),
+ new SparkMetadataColumnInfo("_deleted", DataTypes.BooleanType, false)
+ };
+ }
+}
diff --git
a/spark-connector/v3.4/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/iceberg/SparkIcebergCatalogRestRoutingIT34.java
b/spark-connector/v3.4/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/iceberg/SparkIcebergCatalogRestRoutingIT34.java
new file mode 100644
index 0000000000..ae23429b09
--- /dev/null
+++
b/spark-connector/v3.4/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/iceberg/SparkIcebergCatalogRestRoutingIT34.java
@@ -0,0 +1,25 @@
+/*
+ * 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.integration.test.iceberg;
+
+import org.junit.jupiter.api.condition.DisabledIf;
+
+@DisabledIf("org.apache.gravitino.integration.test.util.ITUtils#isEmbedded")
+public class SparkIcebergCatalogRestRoutingIT34 extends
SparkIcebergCatalogRestRoutingIT {}
diff --git a/spark-connector/v3.5/spark/build.gradle.kts
b/spark-connector/v3.5/spark/build.gradle.kts
index 4b2a7b7402..d59088d5bf 100644
--- a/spark-connector/v3.5/spark/build.gradle.kts
+++ b/spark-connector/v3.5/spark/build.gradle.kts
@@ -151,6 +151,8 @@ dependencies {
testImplementation(libs.mysql.driver)
testImplementation(libs.postgresql.driver)
testImplementation(libs.testcontainers)
+ testImplementation(libs.aws.policy)
+ testImplementation(project(":iceberg:iceberg-common"))
// org.apache.iceberg.rest.RESTSerializers#registerAll(ObjectMapper) has
different method signature for iceberg-core and iceberg-spark-runtime package,
we must make sure iceberg-core is in front to start up MiniGravitino server.
testImplementation("org.apache.iceberg:iceberg-core:$icebergVersion")
diff --git
a/spark-connector/v3.5/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/iceberg/SparkIcebergCatalogRestRoutingIT35.java
b/spark-connector/v3.5/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/iceberg/SparkIcebergCatalogRestRoutingIT35.java
new file mode 100644
index 0000000000..7a091bc0c5
--- /dev/null
+++
b/spark-connector/v3.5/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/iceberg/SparkIcebergCatalogRestRoutingIT35.java
@@ -0,0 +1,25 @@
+/*
+ * 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.integration.test.iceberg;
+
+import org.junit.jupiter.api.condition.DisabledIf;
+
+@DisabledIf("org.apache.gravitino.integration.test.util.ITUtils#isEmbedded")
+public class SparkIcebergCatalogRestRoutingIT35 extends
SparkIcebergCatalogRestRoutingIT {}
diff --git
a/spark-connector/v3.5/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/iceberg/SparkIcebergCatalogRestS3CredentialVendingIT35.java
b/spark-connector/v3.5/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/iceberg/SparkIcebergCatalogRestS3CredentialVendingIT35.java
new file mode 100644
index 0000000000..606131a4a9
--- /dev/null
+++
b/spark-connector/v3.5/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/iceberg/SparkIcebergCatalogRestS3CredentialVendingIT35.java
@@ -0,0 +1,239 @@
+/*
+ * 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.integration.test.iceberg;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Maps;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.sql.SQLException;
+import java.time.Instant;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import org.apache.gravitino.Catalog;
+import org.apache.gravitino.Configs;
+import org.apache.gravitino.MetadataObject;
+import org.apache.gravitino.MetadataObjects;
+import org.apache.gravitino.auth.AuthenticatorType;
+import org.apache.gravitino.authorization.Owner;
+import org.apache.gravitino.authorization.Privileges;
+import org.apache.gravitino.authorization.SecurableObject;
+import org.apache.gravitino.authorization.SecurableObjects;
+import org.apache.gravitino.client.GravitinoMetalake;
+import org.apache.gravitino.integration.test.container.ContainerSuite;
+import org.apache.gravitino.integration.test.container.PostgreSQLContainer;
+import org.apache.gravitino.integration.test.util.BaseIT;
+import org.apache.gravitino.integration.test.util.JwksMockServerHelper;
+import org.apache.gravitino.integration.test.util.OAuthMockDataProvider;
+import org.apache.gravitino.integration.test.util.TestDatabaseName;
+import org.apache.gravitino.server.authentication.OAuthConfig;
+import org.apache.gravitino.spark.connector.GravitinoSparkConfig;
+import org.apache.gravitino.spark.connector.iceberg.GravitinoIcebergCatalog;
+import org.apache.gravitino.spark.connector.plugin.GravitinoSparkPlugin;
+import org.apache.spark.SparkConf;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.SparkSession;
+import org.apache.spark.sql.connector.catalog.CatalogPlugin;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.EnabledIfEnvironmentVariable;
+
+/** Spark 3.5 integration test for OAuth2 IRC routing with AWS credential
vending. */
+@Tag("gravitino-docker-test")
+@EnabledIfEnvironmentVariable(named = "GRAVITINO_TEST_CLOUD_IT", matches =
"true")
+public class SparkIcebergCatalogRestS3CredentialVendingIT35 extends BaseIT {
+
+ private static final String ADMIN = "gravitino";
+ private static final String ALICE = "alice";
+ private static final String ALICE_CREDENTIAL = "alice:alice-secret";
+ private static final String AUDIENCE = "service1";
+ private static final String METALAKE = "spark35_irc_oauth";
+ private static final String CATALOG = "iceberg_pg";
+ private static final String SCHEMA = "aws_vending";
+ private static final String TABLE = "spark35_test";
+ private static final String ROLE = "spark35-irc-role";
+ private static final String AWS_ROLE_ARN = System.getenv("AWS_ROLE_ARN");
+
+ private final ContainerSuite containerSuite = ContainerSuite.getInstance();
+
+ private JwksMockServerHelper mockServer;
+ private SparkSession spark;
+
+ @BeforeAll
+ @Override
+ public void startIntegrationTest() throws Exception {
+ initializeOAuthServer();
+ initializeGravitinoConfig();
+ initializePostgreSQL();
+ super.startIntegrationTest();
+ initializeMetadata();
+ spark = createSparkSession();
+ }
+
+ @AfterAll
+ @Override
+ public void stopIntegrationTest() throws IOException, InterruptedException {
+ if (spark != null) {
+ spark.stop();
+ }
+ if (mockServer != null) {
+ mockServer.close();
+ }
+ super.stopIntegrationTest();
+ }
+
+ @Test
+ void testPostgreSQLCatalogRoutesThroughOAuthIrcAndVendsS3Credentials() {
+ CatalogPlugin catalogPlugin =
spark.sessionState().catalogManager().catalog(CATALOG);
+ Assertions.assertInstanceOf(GravitinoIcebergCatalog.class, catalogPlugin);
+ org.apache.iceberg.catalog.Catalog icebergCatalog =
+ ((GravitinoIcebergCatalog) catalogPlugin).icebergCatalog();
+ Assertions.assertEquals(
+ "org.apache.iceberg.rest.RESTCatalog",
icebergCatalog.getClass().getName());
+
+ spark.sql(String.format("DROP TABLE IF EXISTS %s.%s.%s", CATALOG, SCHEMA,
TABLE));
+ spark.sql(
+ String.format(
+ "CREATE TABLE %s.%s.%s (id BIGINT, data STRING) USING iceberg",
+ CATALOG, SCHEMA, TABLE));
+ spark.sql(
+ String.format(
+ "INSERT INTO %s.%s.%s VALUES (1, 'one'), (2, 'two')", CATALOG,
SCHEMA, TABLE));
+
+ List<Row> rows =
+ spark
+ .sql(String.format("SELECT id, data FROM %s.%s.%s ORDER BY id",
CATALOG, SCHEMA, TABLE))
+ .collectAsList();
+ Assertions.assertEquals(2, rows.size());
+ Assertions.assertEquals(1L, rows.get(0).getLong(0));
+ Assertions.assertEquals("one", rows.get(0).getString(1));
+ Assertions.assertEquals(2L, rows.get(1).getLong(0));
+ Assertions.assertEquals("two", rows.get(1).getString(1));
+
+ Optional<Owner> owner =
+ client
+ .loadMetalake(METALAKE)
+ .getOwner(
+ MetadataObjects.of(
+ ImmutableList.of(CATALOG, SCHEMA, TABLE),
MetadataObject.Type.TABLE));
+ Assertions.assertTrue(owner.isPresent());
+ Assertions.assertEquals(ALICE, owner.get().name());
+
+ spark.sql(String.format("DROP TABLE %s.%s.%s PURGE", CATALOG, SCHEMA,
TABLE));
+ }
+
+ private void initializeOAuthServer() throws Exception {
+ mockServer = JwksMockServerHelper.create("spark35-irc-kid");
+ Instant expiration = Instant.now().plusSeconds(3600);
+ String adminToken = mockServer.mintToken(ADMIN, AUDIENCE, expiration);
+ String aliceToken = mockServer.mintToken(ALICE, AUDIENCE, expiration);
+ mockServer.registerUserToken(ADMIN, adminToken);
+ mockServer.registerUserToken(ALICE, aliceToken);
+ mockServer.setFallbackToken(adminToken);
+
OAuthMockDataProvider.getInstance().setTokenData(adminToken.getBytes(StandardCharsets.UTF_8));
+ }
+
+ private void initializeGravitinoConfig() {
+ ignoreIcebergAuxRestService = false;
+ Map<String, String> configs = Maps.newHashMap();
+ configs.put(Configs.AUTHENTICATORS.getKey(),
AuthenticatorType.OAUTH.name().toLowerCase());
+ configs.put(OAuthConfig.SERVICE_AUDIENCE.getKey(), AUDIENCE);
+ configs.put(
+ OAuthConfig.TOKEN_VALIDATOR_CLASS.getKey(),
+ "org.apache.gravitino.server.authentication.JwksTokenValidator");
+ configs.put(OAuthConfig.JWKS_URI.getKey(), mockServer.jwksUri());
+ configs.put(OAuthConfig.PRINCIPAL_FIELDS.getKey(), "sub");
+ configs.put(Configs.ENABLE_AUTHORIZATION.getKey(), "true");
+ configs.put(Configs.SERVICE_ADMINS.getKey(), ADMIN);
+ configs.put("gravitino.iceberg-rest.catalog-config-provider",
"dynamic-config-provider");
+ configs.put("gravitino.iceberg-rest.gravitino-metalake", METALAKE);
+ registerCustomConfigs(configs);
+ }
+
+ private void initializePostgreSQL() {
+ containerSuite.startPostgreSQLContainer(
+ TestDatabaseName.PG_TEST_ICEBERG_CATALOG_MULTIPLE_JDBC_LOAD);
+ }
+
+ private void initializeMetadata() throws SQLException {
+ PostgreSQLContainer postgres = containerSuite.getPostgreSQLContainer();
+ TestDatabaseName database =
TestDatabaseName.PG_TEST_ICEBERG_CATALOG_MULTIPLE_JDBC_LOAD;
+ client.createMetalake(METALAKE, "", new HashMap<>());
+ GravitinoMetalake metalake = client.loadMetalake(METALAKE);
+ metalake.addUser(ALICE);
+
+ Map<String, String> properties = Maps.newHashMap();
+ properties.put("catalog-backend", "jdbc");
+ properties.put("uri", postgres.getJdbcUrl(database));
+ properties.put("jdbc-driver", postgres.getDriverClassName(database));
+ properties.put("jdbc-user", postgres.getUsername());
+ properties.put("jdbc-password", postgres.getPassword());
+ properties.put(
+ "warehouse",
+ String.format(
+ "s3://%s/gravitino-irc-oauth-demo/spark-3.5-pg",
System.getenv("AWS_S3_TEST_BUCKET")));
+ properties.put("credential-providers", "s3-token");
+ properties.put("s3-access-key-id", System.getenv("AWS_ACCESS_KEY_ID"));
+ properties.put("s3-secret-access-key",
System.getenv("AWS_SECRET_ACCESS_KEY"));
+ properties.put("s3-region", System.getenv("AWS_DEFAULT_REGION"));
+ properties.put("s3-role-arn", AWS_ROLE_ARN);
+ properties.put("io-impl", "org.apache.iceberg.aws.s3.S3FileIO");
+ Catalog catalog =
+ metalake.createCatalog(
+ CATALOG, Catalog.Type.RELATIONAL, "lakehouse-iceberg", "",
properties);
+ catalog.asSchemas().createSchema(SCHEMA, "", new HashMap<>());
+
+ SecurableObject access =
+ SecurableObjects.ofCatalog(
+ CATALOG,
+ ImmutableList.of(
+ Privileges.UseCatalog.allow(),
+ Privileges.UseSchema.allow(),
+ Privileges.CreateTable.allow(),
+ Privileges.ModifyTable.allow(),
+ Privileges.SelectTable.allow()));
+ metalake.createRole(ROLE, new HashMap<>(), ImmutableList.of(access));
+ metalake.grantRolesToUser(ImmutableList.of(ROLE), ALICE);
+ }
+
+ private SparkSession createSparkSession() {
+ SparkConf conf =
+ new SparkConf()
+ .set("spark.plugins", GravitinoSparkPlugin.class.getName())
+ .set(GravitinoSparkConfig.GRAVITINO_URI, serverUri)
+ .set(GravitinoSparkConfig.GRAVITINO_METALAKE, METALAKE)
+ .set(GravitinoSparkConfig.GRAVITINO_ENABLE_ICEBERG_SUPPORT, "true")
+ .set(GravitinoSparkConfig.GRAVITINO_AUTH_TYPE, "oauth2")
+ .set(GravitinoSparkConfig.GRAVITINO_OAUTH2_URI,
mockServer.baseUri())
+ .set(GravitinoSparkConfig.GRAVITINO_OAUTH2_PATH, "token")
+ .set(GravitinoSparkConfig.GRAVITINO_OAUTH2_CREDENTIAL,
ALICE_CREDENTIAL)
+ .set(GravitinoSparkConfig.GRAVITINO_OAUTH2_SCOPE, "openid");
+ return SparkSession.builder()
+ .master("local[1]")
+ .appName("SparkIcebergCatalogRestS3CredentialVendingIT35")
+ .config(conf)
+ .getOrCreate();
+ }
+}
