This is an automated email from the ASF dual-hosted git repository.
diqiu50 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new 5a45beb1c7 [#13967] fix(spark-connector): Reuse token and reject
Kerberos for routed Iceberg REST catalogs (#13968)
5a45beb1c7 is described below
commit 5a45beb1c743fe9b34be7a8d0d5a3596e7ec84af
Author: Yuhui <[email protected]>
AuthorDate: Tue Oct 20 22:21:37 2026 +0800
[#13967] fix(spark-connector): Reuse token and reject Kerberos for routed
Iceberg REST catalogs (#13968)
### What changes were proposed in this pull request?
- With `spark.sql.gravitino.authType=token`, pass the bearer token to
the routed Iceberg REST client as the `token` property.
- With `kerberos`, fail at catalog initialization with an error naming
the auth mode.
- Explicit `rest.auth.*`, `token` or legacy OAuth2 properties keep
priority and disable both.
### Why are the changes needed?
Routed Iceberg catalogs reach IRC without credentials under token auth
and fail with 401, and Kerberos fails silently.
Fix: #13967
### Does this PR introduce _any_ user-facing change?
Yes. Token is reused for IRC, and a Kerberos session with a routed
Iceberg catalog and no explicit `rest.*` auth now fails at
initialization. Docs are updated. No new property keys.
### How was this patch tested?
Added unit tests in `TestIcebergRestOAuthConfig`.
---
docs/spark-connector/spark-catalog-iceberg.md | 4 ++
.../connector/iceberg/IcebergRestOAuthConfig.java | 27 +++++++-
.../iceberg/TestIcebergRestOAuthConfig.java | 77 ++++++++++++++++++++++
3 files changed, 105 insertions(+), 3 deletions(-)
diff --git a/docs/spark-connector/spark-catalog-iceberg.md
b/docs/spark-connector/spark-catalog-iceberg.md
index 22769b20dc..a08b993fb1 100644
--- a/docs/spark-connector/spark-catalog-iceberg.md
+++ b/docs/spark-connector/spark-catalog-iceberg.md
@@ -199,6 +199,10 @@ See [Connect Spark to Iceberg
REST](../iceberg-rest-engine/spark.md) for the ful
When the Gravitino client uses Basic authentication
(`spark.sql.gravitino.authType=basic`), the connector reuses its user name and
password for IRC. Any `rest.auth.*` or `token` property set through
`spark.sql.gravitino.iceberg.rest.*` disables the reuse.
+When the Gravitino client uses token authentication
(`spark.sql.gravitino.authType=token`), the connector passes its bearer token
to IRC as the Iceberg `token` property. The token is read when the catalog is
initialized and isn't refreshed afterwards. Any `rest.auth.*` or `token`
property set through `spark.sql.gravitino.iceberg.rest.*` disables the reuse.
+
+Kerberos can't be carried to IRC because the Iceberg REST client doesn't
support SPNEGO. With `spark.sql.gravitino.authType=kerberos`, a routed Iceberg
catalog fails at initialization unless its authentication is configured through
`spark.sql.gravitino.iceberg.rest.*`.
+
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:
diff --git
a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/iceberg/IcebergRestOAuthConfig.java
b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/iceberg/IcebergRestOAuthConfig.java
index 1a65415aef..540ba982b2 100644
---
a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/iceberg/IcebergRestOAuthConfig.java
+++
b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/iceberg/IcebergRestOAuthConfig.java
@@ -29,6 +29,7 @@ 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.gravitino.spark.connector.plugin.GravitinoDriverPlugin;
import org.apache.spark.SparkConf;
/**
@@ -83,13 +84,25 @@ class IcebergRestOAuthConfig {
if (AuthProperties.isBasic(authType)) {
return applyBasicCredentials(sparkConf, result);
}
+ boolean hasExplicitLegacyOAuth2 =
+ StringUtils.equalsIgnoreCase(explicitAuthType, AUTH_TYPE_OAUTH2)
+ ||
result.keySet().stream().anyMatch(OAUTH2_TRIGGER_PROPERTIES::contains);
+ // An explicit legacy OAuth2 configuration already decides how IRC is
authenticated.
+ if (!hasExplicitLegacyOAuth2) {
+ if (AuthProperties.isToken(authType)) {
+ return applyBearerToken(sparkConf, result);
+ }
+ // The Iceberg REST client has no SPNEGO support, so Kerberos can't be
carried over.
+ Preconditions.checkArgument(
+ !AuthProperties.isKerberos(authType),
+ "Iceberg REST service can't be accessed with %s=kerberos; set"
+ + " spark.sql.gravitino.iceberg.rest.* to configure its
authentication explicitly",
+ GravitinoSparkConfig.GRAVITINO_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;
}
@@ -135,6 +148,14 @@ class IcebergRestOAuthConfig {
return config;
}
+ // Reuses the bearer token of the Gravitino client. The token is captured
when the catalog is
+ // initialized, so it isn't refreshed when the token source changes
afterwards.
+ private static Map<String, String> applyBearerToken(
+ SparkConf sparkConf, Map<String, String> config) {
+ config.put(TOKEN, GravitinoDriverPlugin.resolveToken(sparkConf));
+ return config;
+ }
+
private static void putIfConfigured(Map<String, String> config, String key,
String value) {
if (!config.containsKey(key) && StringUtils.isNotBlank(value)) {
config.put(key, value);
diff --git
a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/iceberg/TestIcebergRestOAuthConfig.java
b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/iceberg/TestIcebergRestOAuthConfig.java
index 4abcd2fed7..b615ef3358 100644
---
a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/iceberg/TestIcebergRestOAuthConfig.java
+++
b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/iceberg/TestIcebergRestOAuthConfig.java
@@ -239,6 +239,83 @@ public class TestIcebergRestOAuthConfig {
Assertions.assertTrue(exception.getMessage().contains("basic.password"));
}
+ @Test
+ void testDerivesIcebergTokenConfig() {
+ SparkConf sparkConf = new SparkConf(false);
+ sparkConf.set(GravitinoSparkConfig.GRAVITINO_AUTH_TYPE, "token");
+ sparkConf.set(GravitinoSparkConfig.GRAVITINO_TOKEN_VALUE, " bearer-token
");
+
+ Map<String, String> result = IcebergRestOAuthConfig.resolve(sparkConf,
Collections.emptyMap());
+
+ Assertions.assertEquals(ImmutableMap.of("token", "bearer-token"), result);
+ }
+
+ @Test
+ void testExplicitConfigTakesPrecedenceOverReusedToken() {
+ SparkConf sparkConf = new SparkConf(false);
+ sparkConf.set(GravitinoSparkConfig.GRAVITINO_AUTH_TYPE, "token");
+ sparkConf.set(GravitinoSparkConfig.GRAVITINO_TOKEN_VALUE, "bearer-token");
+ Map<String, String> explicit = ImmutableMap.of("token", "irc-token");
+
+ Map<String, String> result = IcebergRestOAuthConfig.resolve(sparkConf,
explicit);
+
+ Assertions.assertEquals(explicit, result);
+ }
+
+ @Test
+ void testTokenReuseFailsWhenTokenIsMissing() {
+ SparkConf sparkConf = new SparkConf(false);
+ sparkConf.set(GravitinoSparkConfig.GRAVITINO_AUTH_TYPE, "token");
+
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> IcebergRestOAuthConfig.resolve(sparkConf,
Collections.emptyMap()));
+ }
+
+ @Test
+ void testKerberosFailsFast() {
+ SparkConf sparkConf = new SparkConf(false);
+ sparkConf.set(GravitinoSparkConfig.GRAVITINO_AUTH_TYPE, "kerberos");
+
+ IllegalArgumentException exception =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> IcebergRestOAuthConfig.resolve(sparkConf,
Collections.emptyMap()));
+
+ Assertions.assertTrue(exception.getMessage().contains("kerberos"));
+ }
+
+ @Test
+ void testExplicitConfigSkipsKerberosFailure() {
+ SparkConf sparkConf = new SparkConf(false);
+ sparkConf.set(GravitinoSparkConfig.GRAVITINO_AUTH_TYPE, "kerberos");
+ Map<String, String> explicit = ImmutableMap.of("token", "irc-token");
+
+ Map<String, String> result = IcebergRestOAuthConfig.resolve(sparkConf,
explicit);
+
+ Assertions.assertEquals(explicit, result);
+ }
+
+ @Test
+ void testExplicitLegacyOAuth2SkipsTokenAndKerberosHandling() {
+ Map<String, String> explicit =
+ ImmutableMap.of(
+ "credential",
+ "irc:secret",
+ "oauth2-server-uri",
+ "https://identity.example.com/oauth/token";);
+ for (String authType : new String[] {"token", "kerberos"}) {
+ SparkConf sparkConf = new SparkConf(false);
+ sparkConf.set(GravitinoSparkConfig.GRAVITINO_AUTH_TYPE, authType);
+ sparkConf.set(GravitinoSparkConfig.GRAVITINO_TOKEN_VALUE,
"bearer-token");
+
+ Map<String, String> result = IcebergRestOAuthConfig.resolve(sparkConf,
explicit);
+
+ Assertions.assertFalse(result.containsKey(IcebergRestOAuthConfig.TOKEN));
+ Assertions.assertEquals("irc:secret",
result.get(IcebergRestOAuthConfig.CREDENTIAL));
+ }
+ }
+
private SparkConf basicSparkConf(String username, String password) {
SparkConf sparkConf = new SparkConf(false);
sparkConf.set(GravitinoSparkConfig.GRAVITINO_AUTH_TYPE, "basic");
