This is an automated email from the ASF dual-hosted git repository.
jerryshao 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 77b164ebbe [Cherry-pick to branch-1.3] [#14031] fix(flink-connector):
Validate credential providers for Iceberg non-REST backends (#14032) (#14132)
77b164ebbe is described below
commit 77b164ebbe88cdcb3ccd11a74f05bd75084e0ac9
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Fri Oct 23 22:55:27 2026 +0800
[Cherry-pick to branch-1.3] [#14031] fix(flink-connector): Validate
credential providers for Iceberg non-REST backends (#14032) (#14132)
**Cherry-pick Information:**
- Original commit: f559909447f33aca3abbfd15547dfad25e22ef36
- Target branch: `branch-1.3`
- Status: ✅ Conflicts resolved manually
---------
Co-authored-by: Yuhui <[email protected]>
---
docs/flink-connector/flink-catalog-iceberg.md | 9 ++
docs/security/credential-vending.md | 4 +-
.../connector/iceberg/GravitinoIcebergCatalog.java | 91 +++++++++++-
.../iceberg/IcebergPropertiesConstants.java | 9 ++
.../iceberg/TestGravitinoIcebergCatalog.java | 159 +++++++++++++++++++++
.../iceberg/TestGravitinoIcebergCatalogOpen.java | 101 +++++++++++++
6 files changed, 370 insertions(+), 3 deletions(-)
diff --git a/docs/flink-connector/flink-catalog-iceberg.md
b/docs/flink-connector/flink-catalog-iceberg.md
index b6c19a8de2..dea6de1936 100644
--- a/docs/flink-connector/flink-catalog-iceberg.md
+++ b/docs/flink-connector/flink-catalog-iceberg.md
@@ -119,6 +119,15 @@ Gravitino catalog property names with the prefix
`flink.bypass.` are passed to F
## Storage
+### Credential Vending
+
+Unless the catalog uses a `REST` backend with `data-access` set to
`vended-credentials`, the Flink connector fetches catalog-level credentials
from Gravitino once, when the catalog is opened, and never refreshes them. Only
static credential providers, such as `s3-secret-key`, `oss-secret-key` and
`azure-account-key`, are vended to Flink:
+
+- `s3-token`, `gcs-token`, `adls-token` and `oss-token` vend no credential at
catalog level. Flink logs a warning and Iceberg uses the credentials available
on the Flink nodes, such as an instance profile.
+- `aws-irsa` is rejected when the catalog is loaded, because its credential
expires and is never refreshed.
+
+For temporary credentials vended per table and refreshed automatically, use a
`REST` backend with `data-access` set to `vended-credentials`.
+
### OSS
Additionally, you need download the [Aliyun OSS
SDK](https://gosspublic.alicdn.com/sdks/java/aliyun_java_sdk_3.10.2.zip), and
copy `aliyun-sdk-oss-3.10.2.jar`, `hamcrest-core-1.1.jar`, `jdom2-2.0.6.jar` to
the Flink classpath.
diff --git a/docs/security/credential-vending.md
b/docs/security/credential-vending.md
index 7604f51548..dcae62aa75 100644
--- a/docs/security/credential-vending.md
+++ b/docs/security/credential-vending.md
@@ -27,7 +27,7 @@ Gravitino credential vending is used to generate temporary or
static credentials
- GCS: `GCSTokenCredential`
- ADLS: `ADLSTokenCredential`, `AzureAccountKeyCredential`
- OSS: `OSSTokenCredential`, `OSSSecretKeyCredential`
-- Spark, Flink, and Trino connectors automatically consume vended credentials
for Hive, Iceberg, Glue, JDBC, and Paimon catalogs.
+- Spark, Flink, and Trino connectors automatically consume vended credentials
for Hive, Iceberg, Glue, JDBC, and Paimon catalogs. For Iceberg catalogs
without a `REST` backend that vends credentials, the Flink connector receives
only static credentials, see [Flink Iceberg credential
vending](../flink-connector/flink-catalog-iceberg.md#credential-vending).
## General Configurations
@@ -225,7 +225,7 @@ Clients retrieve vended credentials from:
GET /api/metalakes/{metalake}/objects/catalog/{catalog}/credentials
```
-The server returns short-lived or static credentials that the client can use
to access the underlying storage directly.
+The server returns short-lived or static credentials that the client can use
to access the underlying storage directly. These credentials are catalog-level
and are not refreshed: token providers such as `s3-token` vend nothing at
catalog level, and `aws-irsa` credentials expire after
`s3-token-expire-in-secs`.
### Spark and Flink Connector Integration
diff --git
a/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/iceberg/GravitinoIcebergCatalog.java
b/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/iceberg/GravitinoIcebergCatalog.java
index 7bf33b5e1c..6edf8b255b 100644
---
a/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/iceberg/GravitinoIcebergCatalog.java
+++
b/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/iceberg/GravitinoIcebergCatalog.java
@@ -18,18 +18,37 @@
*/
package org.apache.gravitino.flink.connector.iceberg;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Splitter;
+import com.google.common.collect.ImmutableSet;
+import java.util.Collections;
import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
import java.util.Map;
import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+import javax.annotation.Nullable;
+import org.apache.commons.lang3.StringUtils;
import org.apache.flink.table.catalog.AbstractCatalog;
import org.apache.flink.table.catalog.exceptions.CatalogException;
import org.apache.flink.table.factories.Factory;
import org.apache.flink.util.Preconditions;
+import org.apache.gravitino.Catalog;
+import org.apache.gravitino.credential.ADLSTokenCredential;
+import org.apache.gravitino.credential.AwsIrsaCredential;
+import org.apache.gravitino.credential.CredentialConstants;
import org.apache.gravitino.credential.CredentialPropertyUtils;
+import org.apache.gravitino.credential.GCSTokenCredential;
+import org.apache.gravitino.credential.OSSTokenCredential;
+import org.apache.gravitino.credential.S3SecretKeyCredential;
+import org.apache.gravitino.credential.S3TokenCredential;
import org.apache.gravitino.exceptions.NoSuchCatalogException;
import org.apache.gravitino.flink.connector.PartitionConverter;
import org.apache.gravitino.flink.connector.SchemaAndTablePropertiesConverter;
import org.apache.gravitino.flink.connector.catalog.BaseCatalog;
+import org.apache.gravitino.storage.GCSProperties;
import org.apache.iceberg.flink.FlinkCatalogFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -39,6 +58,13 @@ public class GravitinoIcebergCatalog extends BaseCatalog {
private static final Logger LOG =
LoggerFactory.getLogger(GravitinoIcebergCatalog.class);
+ private static final Set<String> PATH_SCOPED_TOKEN_PROVIDERS =
+ ImmutableSet.of(
+ S3TokenCredential.S3_TOKEN_CREDENTIAL_TYPE,
+ GCSTokenCredential.GCS_TOKEN_CREDENTIAL_TYPE,
+ ADLSTokenCredential.ADLS_TOKEN_CREDENTIAL_TYPE,
+ OSSTokenCredential.OSS_TOKEN_CREDENTIAL_TYPE);
+
private final String icebergCatalogName;
// Mutable copy so credential injection in open() propagates to the inner
Iceberg catalog.
private final Map<String, String> mutableIcebergProperties;
@@ -83,8 +109,20 @@ public class GravitinoIcebergCatalog extends BaseCatalog {
public void open() throws CatalogException {
if (icebergCatalog == null) {
try {
+ Catalog gravitinoCatalog = catalog();
+ List<String> tokenProviders =
+ checkCredentialProviders(
+ catalogName(), mutableIcebergProperties,
gravitinoCatalog.properties());
+ if (!tokenProviders.isEmpty()) {
+ LOG.warn(
+ "Catalog '{}' uses credential providers {}, which vend no
credential at catalog"
+ + " level. The Flink connector injects no storage credential
for them, so"
+ + " Iceberg FileIO uses the credentials available on the
Flink nodes.",
+ catalogName(),
+ tokenProviders);
+ }
CredentialPropertyUtils.applyIcebergCredentials(
- CredentialPropertyUtils.getCredentials(catalog()),
mutableIcebergProperties);
+ CredentialPropertyUtils.getCredentials(gravitinoCatalog),
mutableIcebergProperties);
} catch (NoSuchCatalogException e) {
LOG.warn(
"Catalog '{}' not found in Gravitino during open(); credential
injection skipped."
@@ -119,4 +157,55 @@ public class GravitinoIcebergCatalog extends BaseCatalog {
catalog.getClass().getName());
return (AbstractCatalog) catalog;
}
+
+ // Without per-table vended credentials, the inner catalog only receives the
catalog-level
+ // credentials fetched once in open(). A REST backend with vended
credentials enabled is exempt
+ // because the inner REST catalog obtains per-table, refreshable credentials
itself. aws-irsa is
+ // rejected: its credential is injected but never refreshed, so it overrides
the node's own
+ // credentials and then expires. Path-scoped token providers vend nothing at
catalog level, which
+ // leaves the node's own credentials in effect, so they are only returned
for the caller to warn
+ // about. The REST check uses the effective inner catalog-type, since an
explicit catalog-type may
+ // differ from the persisted catalog-backend. When no provider list is set,
the server infers
+ // gcs-token from gcs-service-account-file, which is not reflected in the
catalog properties seen
+ // by the client, so the same inference is applied here.
+ @VisibleForTesting
+ static List<String> checkCredentialProviders(
+ String catalogName,
+ Map<String, String> icebergCatalogProperties,
+ @Nullable Map<String, String> gravitinoCatalogProperties) {
+ boolean restWithVendedCredentials =
+
IcebergPropertiesConstants.ICEBERG_CATALOG_BACKEND_REST.equalsIgnoreCase(
+
icebergCatalogProperties.get(IcebergPropertiesConstants.ICEBERG_CATALOG_TYPE))
+ &&
IcebergPropertiesConstants.ICEBERG_DATA_ACCESS_VENDED_CREDENTIALS.equalsIgnoreCase(
+
icebergCatalogProperties.get(IcebergPropertiesConstants.ICEBERG_ACCESS_DELEGATION));
+ if (restWithVendedCredentials || gravitinoCatalogProperties == null) {
+ return Collections.emptyList();
+ }
+ String providers =
gravitinoCatalogProperties.get(CredentialConstants.CREDENTIAL_PROVIDERS);
+ if (StringUtils.isBlank(providers)) {
+ return StringUtils.isNotBlank(
+
gravitinoCatalogProperties.get(GCSProperties.GRAVITINO_GCS_SERVICE_ACCOUNT_FILE))
+ ?
Collections.singletonList(GCSTokenCredential.GCS_TOKEN_CREDENTIAL_TYPE)
+ : Collections.emptyList();
+ }
+ List<String> providerList =
+
Splitter.on(',').trimResults().omitEmptyStrings().splitToList(providers);
+ if (providerList.stream()
+
.anyMatch(AwsIrsaCredential.AWS_IRSA_CREDENTIAL_TYPE::equalsIgnoreCase)) {
+ throw new CatalogException(
+ String.format(
+ "Catalog '%s' uses credential provider %s, which the Flink
connector does not"
+ + " support for Iceberg catalogs with a non-REST backend:
its credential is"
+ + " never refreshed and expires. Use a static provider such
as %s, or use a"
+ + " REST backend with %s=%s.",
+ catalogName,
+ AwsIrsaCredential.AWS_IRSA_CREDENTIAL_TYPE,
+ S3SecretKeyCredential.S3_SECRET_KEY_CREDENTIAL_TYPE,
+ IcebergPropertiesConstants.GRAVITINO_ICEBERG_DATA_ACCESS,
+
IcebergPropertiesConstants.ICEBERG_DATA_ACCESS_VENDED_CREDENTIALS));
+ }
+ return providerList.stream()
+ .filter(provider ->
PATH_SCOPED_TOKEN_PROVIDERS.contains(provider.toLowerCase(Locale.ROOT)))
+ .collect(Collectors.toList());
+ }
}
diff --git
a/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/iceberg/IcebergPropertiesConstants.java
b/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/iceberg/IcebergPropertiesConstants.java
index d879bb10e0..0d286c5d0c 100644
---
a/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/iceberg/IcebergPropertiesConstants.java
+++
b/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/iceberg/IcebergPropertiesConstants.java
@@ -52,4 +52,13 @@ public class IcebergPropertiesConstants {
@VisibleForTesting
public static final String ICEBERG_CATALOG_BACKEND_REST =
CatalogUtil.ICEBERG_CATALOG_TYPE_REST;
+
+ /** Gravitino Iceberg catalog property that selects how clients access table
data. */
+ public static final String GRAVITINO_ICEBERG_DATA_ACCESS =
IcebergConstants.DATA_ACCESS;
+
+ /** Iceberg catalog property that the data access mode is passed to the REST
backend as. */
+ public static final String ICEBERG_ACCESS_DELEGATION =
IcebergConstants.ICEBERG_ACCESS_DELEGATION;
+
+ /** Data access mode in which the REST backend vends per-table storage
credentials. */
+ public static final String ICEBERG_DATA_ACCESS_VENDED_CREDENTIALS =
"vended-credentials";
}
diff --git
a/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/iceberg/TestGravitinoIcebergCatalog.java
b/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/iceberg/TestGravitinoIcebergCatalog.java
index d801841b53..8cb4972175 100644
---
a/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/iceberg/TestGravitinoIcebergCatalog.java
+++
b/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/iceberg/TestGravitinoIcebergCatalog.java
@@ -19,16 +19,31 @@
package org.apache.gravitino.flink.connector.iceberg;
+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.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.Mockito.mock;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import java.util.Locale;
+import java.util.Map;
import org.apache.flink.table.catalog.AbstractCatalog;
import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.gravitino.credential.CredentialConstants;
+import org.apache.gravitino.credential.GCSTokenCredential;
+import org.apache.gravitino.storage.GCSProperties;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
class TestGravitinoIcebergCatalog {
+ private static final Map<String, String> HIVE_CATALOG =
+ ImmutableMap.of(IcebergPropertiesConstants.ICEBERG_CATALOG_TYPE, "hive");
+
@Test
void testRequireAbstractCatalogAcceptsAbstractCatalog() {
AbstractCatalog catalog = mock(AbstractCatalog.class);
@@ -43,4 +58,148 @@ class TestGravitinoIcebergCatalog {
assertThrows(
IllegalStateException.class, () ->
GravitinoIcebergCatalog.asAbstractCatalog(catalog));
}
+
+ @Test
+ void testCheckCredentialProvidersRejectsAwsIrsaForNonRestBackend() {
+ Map<String, String> properties =
+ ImmutableMap.of(
+ IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_BACKEND,
+ "hive",
+ CredentialConstants.CREDENTIAL_PROVIDERS,
+ "s3-secret-key, AWS-IRSA");
+
+ CatalogException exception =
+ assertThrows(
+ CatalogException.class,
+ () -> GravitinoIcebergCatalog.checkCredentialProviders("c1",
HIVE_CATALOG, properties));
+ assertTrue(exception.getMessage().contains("aws-irsa"));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"s3-token", "gcs-token", "adls-token", "oss-token"})
+ void
testCheckCredentialProvidersReturnsTokenProvidersForNonRestBackend(String
provider) {
+ String upperCaseProvider = provider.toUpperCase(Locale.ROOT);
+ Map<String, String> properties =
+ ImmutableMap.of(
+ IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_BACKEND,
+ "hive",
+ CredentialConstants.CREDENTIAL_PROVIDERS,
+ "s3-secret-key, " + upperCaseProvider);
+
+ assertEquals(
+ ImmutableList.of(upperCaseProvider),
+ GravitinoIcebergCatalog.checkCredentialProviders("c1", HIVE_CATALOG,
properties));
+ }
+
+ @Test
+ void testCheckCredentialProvidersIgnoresStaticProviders() {
+ Map<String, String> properties =
+ ImmutableMap.of(
+ IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_BACKEND,
+ "jdbc",
+ CredentialConstants.CREDENTIAL_PROVIDERS,
+ "s3-secret-key,jdbc-user-password");
+
+ assertTrue(
+ GravitinoIcebergCatalog.checkCredentialProviders("c1",
ImmutableMap.of(), properties)
+ .isEmpty());
+ }
+
+ @Test
+ void testCheckCredentialProvidersIgnoresMissingProviders() {
+ Map<String, String> properties =
+
ImmutableMap.of(IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_BACKEND,
"hive");
+
+ assertTrue(
+ GravitinoIcebergCatalog.checkCredentialProviders("c1", HIVE_CATALOG,
properties).isEmpty());
+ }
+
+ @Test
+ void testCheckCredentialProvidersInfersGcsTokenFromServiceAccountFile() {
+ Map<String, String> properties =
+ ImmutableMap.of(
+ IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_BACKEND,
+ "hive",
+ GCSProperties.GRAVITINO_GCS_SERVICE_ACCOUNT_FILE,
+ "/path/to/sa.json");
+
+ assertEquals(
+ ImmutableList.of(GCSTokenCredential.GCS_TOKEN_CREDENTIAL_TYPE),
+ GravitinoIcebergCatalog.checkCredentialProviders("c1", HIVE_CATALOG,
properties));
+ }
+
+ @Test
+ void testCheckCredentialProvidersUsesExplicitProvidersOverGcsInference() {
+ Map<String, String> properties =
+ ImmutableMap.of(
+ IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_BACKEND,
+ "hive",
+ GCSProperties.GRAVITINO_GCS_SERVICE_ACCOUNT_FILE,
+ "/path/to/sa.json",
+ CredentialConstants.CREDENTIAL_PROVIDERS,
+ "s3-secret-key");
+
+ assertTrue(
+ GravitinoIcebergCatalog.checkCredentialProviders("c1", HIVE_CATALOG,
properties).isEmpty());
+ }
+
+ @Test
+ void testCheckCredentialProvidersSkipsRestBackendWithVendedCredentials() {
+ Map<String, String> properties =
+ ImmutableMap.of(
+ IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_BACKEND,
+ "REST",
+ CredentialConstants.CREDENTIAL_PROVIDERS,
+ "s3-token,aws-irsa");
+
+ assertTrue(
+ GravitinoIcebergCatalog.checkCredentialProviders(
+ "c1", restCatalog("REST", true), properties)
+ .isEmpty());
+ }
+
+ @Test
+ void testCheckCredentialProvidersUsesEffectiveCatalogType() {
+ Map<String, String> properties =
+ ImmutableMap.of(
+ IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_BACKEND,
+ "hive",
+ CredentialConstants.CREDENTIAL_PROVIDERS,
+ "aws-irsa");
+
+ assertTrue(
+ GravitinoIcebergCatalog.checkCredentialProviders(
+ "c1", restCatalog("rest", true), properties)
+ .isEmpty());
+ }
+
+ @Test
+ void
testCheckCredentialProvidersRejectsAwsIrsaForRestBackendWithoutVendedCredentials()
{
+ Map<String, String> properties =
+ ImmutableMap.of(CredentialConstants.CREDENTIAL_PROVIDERS, "aws-irsa");
+
+ assertThrows(
+ CatalogException.class,
+ () ->
+ GravitinoIcebergCatalog.checkCredentialProviders(
+ "c1", restCatalog("rest", false), properties));
+ }
+
+ @Test
+ void testCheckCredentialProvidersIgnoresNullProperties() {
+ assertTrue(
+ GravitinoIcebergCatalog.checkCredentialProviders("c1", HIVE_CATALOG,
null).isEmpty());
+ }
+
+ private static Map<String, String> restCatalog(String catalogType, boolean
vendedCredentials) {
+ ImmutableMap.Builder<String, String> builder =
+ ImmutableMap.<String, String>builder()
+ .put(IcebergPropertiesConstants.ICEBERG_CATALOG_TYPE, catalogType);
+ if (vendedCredentials) {
+ builder.put(
+ IcebergPropertiesConstants.ICEBERG_ACCESS_DELEGATION,
+ IcebergPropertiesConstants.ICEBERG_DATA_ACCESS_VENDED_CREDENTIALS);
+ }
+ return builder.build();
+ }
}
diff --git
a/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/iceberg/TestGravitinoIcebergCatalogOpen.java
b/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/iceberg/TestGravitinoIcebergCatalogOpen.java
new file mode 100644
index 0000000000..453f1d6fc0
--- /dev/null
+++
b/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/iceberg/TestGravitinoIcebergCatalogOpen.java
@@ -0,0 +1,101 @@
+/*
+ * 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.flink.connector.iceberg;
+
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import com.google.common.collect.ImmutableMap;
+import java.util.Map;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.gravitino.Catalog;
+import org.apache.gravitino.credential.Credential;
+import org.apache.gravitino.credential.CredentialConstants;
+import org.apache.gravitino.credential.SupportsCredentials;
+import org.apache.gravitino.flink.connector.PartitionConverter;
+import org.apache.gravitino.flink.connector.SchemaAndTablePropertiesConverter;
+import org.apache.gravitino.flink.connector.utils.CatalogCompat;
+import org.junit.jupiter.api.Test;
+
+class TestGravitinoIcebergCatalogOpen {
+
+ private static final Map<String, String> HIVE_BACKEND_WITH_AWS_IRSA =
+ ImmutableMap.of(
+ IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_BACKEND,
+ "hive",
+ CredentialConstants.CREDENTIAL_PROVIDERS,
+ "aws-irsa");
+
+ @Test
+ void testOpenRejectsAwsIrsaWhenEffectiveCatalogTypeIsNotRest() {
+ TestableIcebergCatalog catalog =
+ new TestableIcebergCatalog(
+ ImmutableMap.of(IcebergPropertiesConstants.ICEBERG_CATALOG_TYPE,
"hive"),
+ mockGravitinoCatalog(HIVE_BACKEND_WITH_AWS_IRSA));
+
+ assertThrows(CatalogException.class, catalog::open);
+ }
+
+ @Test
+ void testOpenRejectsAwsIrsaWhenRestBackendHasNoVendedCredentials() {
+ TestableIcebergCatalog catalog =
+ new TestableIcebergCatalog(
+ ImmutableMap.of(IcebergPropertiesConstants.ICEBERG_CATALOG_TYPE,
"rest"),
+ mockGravitinoCatalog(HIVE_BACKEND_WITH_AWS_IRSA));
+
+ assertThrows(CatalogException.class, catalog::open);
+ }
+
+ private static Catalog mockGravitinoCatalog(Map<String, String> properties) {
+ Catalog gravitinoCatalog = mock(Catalog.class);
+ SupportsCredentials supportsCredentials = mock(SupportsCredentials.class);
+ when(gravitinoCatalog.properties()).thenReturn(properties);
+
when(gravitinoCatalog.supportsCredentials()).thenReturn(supportsCredentials);
+ when(supportsCredentials.getCredentials()).thenReturn(new Credential[0]);
+ return gravitinoCatalog;
+ }
+
+ private static class TestableIcebergCatalog extends GravitinoIcebergCatalog {
+
+ private final Catalog gravitinoCatalog;
+
+ TestableIcebergCatalog(Map<String, String> icebergProperties, Catalog
gravitinoCatalog) {
+ super(
+ "c1",
+ "default",
+ mock(SchemaAndTablePropertiesConverter.class),
+ mock(PartitionConverter.class),
+ icebergProperties,
+ icebergProperties);
+ this.gravitinoCatalog = gravitinoCatalog;
+ }
+
+ @Override
+ protected Catalog catalog() {
+ return gravitinoCatalog;
+ }
+
+ @Override
+ protected CatalogCompat catalogCompat() {
+ return mock(CatalogCompat.class);
+ }
+ }
+}
