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 f559909447 [#14031] fix(flink-connector): Validate credential
providers for Iceberg non-REST backends (#14032)
f559909447 is described below
commit f559909447f33aca3abbfd15547dfad25e22ef36
Author: Yuhui <[email protected]>
AuthorDate: Thu Oct 22 09:49:33 2026 +0800
[#14031] fix(flink-connector): Validate credential providers for Iceberg
non-REST backends (#14032)
### What changes were proposed in this pull request?
For Iceberg catalogs with a non-REST backend, the Flink connector now
checks `credential-providers` when it opens the catalog:
- `aws-irsa` is rejected, because its catalog-level credential is
injected once and expires after `s3-token-expire-in-secs`.
- `s3-token`, `gcs-token`, `adls-token`, `oss-token` and `cos-token` log
a warning, because they vend nothing at catalog level and Iceberg falls
back to the credentials on the Flink nodes.
- REST backends are not affected.
The docs now state that Flink only receives static credentials for
non-REST backends, and that per-table refreshable credentials require a
REST backend with `data-access=vended-credentials`.
### Why are the changes needed?
The Flink connector fetches catalog-level credentials once and never
refreshes them. With `aws-irsa`, jobs fail with `ExpiredToken` after
about an hour; with the other token providers, Flink silently gets no
credential, while the docs claim credentials are consumed automatically.
Fix: #14031
### Does this PR introduce _any_ user-facing change?
Yes. Loading an Iceberg catalog with a Hive or JDBC backend and
`aws-irsa` now fails in Flink. Token providers only log a warning.
### How was this patch tested?
Added unit tests in `TestGravitinoIcebergCatalog`.
---
docs/flink-connector/flink-catalog-iceberg.md | 9 ++
docs/security/credential-vending.md | 4 +-
.../connector/iceberg/GravitinoIcebergCatalog.java | 93 +++++++++++-
.../iceberg/IcebergPropertiesConstants.java | 9 ++
.../iceberg/TestGravitinoIcebergCatalog.java | 159 +++++++++++++++++++++
.../iceberg/TestGravitinoIcebergCatalogOpen.java | 133 +++++++++++++++++
6 files changed, 404 insertions(+), 3 deletions(-)
diff --git a/docs/flink-connector/flink-catalog-iceberg.md
b/docs/flink-connector/flink-catalog-iceberg.md
index db88e7a2b4..f11491abbb 100644
--- a/docs/flink-connector/flink-catalog-iceberg.md
+++ b/docs/flink-connector/flink-catalog-iceberg.md
@@ -120,6 +120,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`, `oss-token` and `cos-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 d58dcb78ae..46eccd0306 100755
--- a/docs/security/credential-vending.md
+++ b/docs/security/credential-vending.md
@@ -20,7 +20,7 @@ Gravitino credential vending is used to generate temporary or
static credentials
| JDBC | JDBC user and password |
| Paimon | S3, OSS, JDBC user and password |
-S3 is Amazon S3, OSS is Alibaba Cloud OSS, GCS is Google Cloud Storage, ADLS
is Azure Data Lake Storage, and COS is Tencent Cloud COS. The Gravitino Spark,
Flink, and Trino connectors consume vended credentials automatically for these
catalogs.
+S3 is Amazon S3, OSS is Alibaba Cloud OSS, GCS is Google Cloud Storage, ADLS
is Azure Data Lake Storage, and COS is Tencent Cloud COS. The Gravitino Spark,
Flink, and Trino connectors consume vended credentials automatically for these
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).
## Quick Start
@@ -475,7 +475,7 @@ For a catalog, `{type}` is `catalog` and `{full_name}` is
the catalog name:
GET /api/metalakes/{metalake}/objects/catalog/{catalog}/credentials
```
-The Gravitino Spark and Flink connectors call this for you and inject the
returned credentials, so no client configuration is needed.
+The Gravitino Spark and Flink connectors call this for you and inject the
returned credentials, so no client configuration is needed. 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`.
### Read or Write Scope
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 9e3b19d14d..3ec62cfa15 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,38 @@
*/
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.COSTokenCredential;
+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.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -38,6 +58,14 @@ public abstract 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,
+ COSTokenCredential.COS_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;
@@ -82,8 +110,20 @@ public abstract 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."
@@ -132,4 +172,55 @@ public abstract 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..b5c241ae94 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",
"cos-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..dc3c606db3
--- /dev/null
+++
b/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/iceberg/TestGravitinoIcebergCatalogOpen.java
@@ -0,0 +1,133 @@
+/*
+ * 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.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import com.google.common.collect.ImmutableMap;
+import java.util.Map;
+import org.apache.flink.table.catalog.AbstractCatalog;
+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
testOpenAllowsAwsIrsaWhenEffectiveCatalogTypeIsRestWithVendedCredentials() {
+ AbstractCatalog innerCatalog = mock(AbstractCatalog.class);
+ TestableIcebergCatalog catalog =
+ new TestableIcebergCatalog(
+ ImmutableMap.of(
+ IcebergPropertiesConstants.ICEBERG_CATALOG_TYPE,
+ "rest",
+ IcebergPropertiesConstants.ICEBERG_ACCESS_DELEGATION,
+
IcebergPropertiesConstants.ICEBERG_DATA_ACCESS_VENDED_CREDENTIALS),
+ mockGravitinoCatalog(HIVE_BACKEND_WITH_AWS_IRSA),
+ innerCatalog);
+
+ assertDoesNotThrow(catalog::open);
+ verify(innerCatalog).open();
+ }
+
+ @Test
+ void testOpenRejectsAwsIrsaWhenEffectiveCatalogTypeIsNotRest() {
+ TestableIcebergCatalog catalog =
+ new TestableIcebergCatalog(
+ ImmutableMap.of(IcebergPropertiesConstants.ICEBERG_CATALOG_TYPE,
"hive"),
+ mockGravitinoCatalog(HIVE_BACKEND_WITH_AWS_IRSA),
+ mock(AbstractCatalog.class));
+
+ 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),
+ mock(AbstractCatalog.class));
+
+ 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;
+ private final AbstractCatalog innerCatalog;
+
+ TestableIcebergCatalog(
+ Map<String, String> icebergProperties,
+ Catalog gravitinoCatalog,
+ AbstractCatalog innerCatalog) {
+ super(
+ "c1",
+ "default",
+ mock(SchemaAndTablePropertiesConverter.class),
+ mock(PartitionConverter.class),
+ icebergProperties,
+ icebergProperties);
+ this.gravitinoCatalog = gravitinoCatalog;
+ this.innerCatalog = innerCatalog;
+ }
+
+ @Override
+ protected Catalog catalog() {
+ return gravitinoCatalog;
+ }
+
+ @Override
+ protected Object createInnerIcebergCatalog(String catalogName, Map<String,
String> properties) {
+ return innerCatalog;
+ }
+
+ @Override
+ protected CatalogCompat catalogCompat() {
+ return mock(CatalogCompat.class);
+ }
+ }
+}
