This is an automated email from the ASF dual-hosted git repository.
mchades 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 612932ce68 [Cherry-pick to branch-1.3] [#11601] fix(flink-connector):
Propagate REST auth on catalog-store load path (#11694) (#11724)
612932ce68 is described below
commit 612932ce684b0391ee80e6ed7dc8cab76c7a2b26
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Thu Jun 18 09:36:47 2026 +0800
[Cherry-pick to branch-1.3] [#11601] fix(flink-connector): Propagate REST
auth on catalog-store load path (#11694) (#11724)
**Cherry-pick Information:**
- Original commit: 1014f2677c8f10e78f7e8c140103df65c63b197b
- Target branch: `branch-1.3`
- Status: ✅ Clean cherry-pick (no conflicts)
Co-authored-by: Yuhui <[email protected]>
---
.../iceberg/GravitinoIcebergCatalogFactory.java | 55 ++++++++++++++++------
.../TestGravitinoIcebergCatalogFactory.java | 30 +++++++++++-
.../test/iceberg/FlinkIcebergJdbcCatalogIT.java | 36 +++-----------
3 files changed, 77 insertions(+), 44 deletions(-)
diff --git
a/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/iceberg/GravitinoIcebergCatalogFactory.java
b/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/iceberg/GravitinoIcebergCatalogFactory.java
index b39ef2e53c..022ff34613 100644
---
a/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/iceberg/GravitinoIcebergCatalogFactory.java
+++
b/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/iceberg/GravitinoIcebergCatalogFactory.java
@@ -122,35 +122,62 @@ public class GravitinoIcebergCatalogFactory implements
BaseCatalogFactory {
Map<String, String> icebergCatalogOptions =
Maps.newHashMap(catalogOptions);
String catalogBackend =
catalogOptions.get(IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_BACKEND);
- // Only infer `catalog-type` from the backend when neither `catalog-type`
nor `catalog-impl` is
- // already set, otherwise an explicitly provided `catalog-impl` would
conflict with it.
+ // catalogBackend is only present here on the CREATE CATALOG path (raw
user SQL options). On
+ // the USE CATALOG path (loading a catalog already persisted in Gravitino),
+ // IcebergPropertiesConverter has already renamed catalog-backend ->
catalog-type during
+ // property conversion, so catalogBackend is null even for a REST/JDBC
catalog. This
+ // normalizes the CREATE CATALOG case so both paths converge on the same
catalog-type key.
+ String catalogType = normalizeCatalogType(icebergCatalogOptions,
catalogBackend);
+ // catalogType (not catalogBackend, which is unreliable as explained
above) must be used for
+ // every backend check from here on, or it silently no-ops on the USE
CATALOG path -- this was
+ // the root cause of #11601 (REST auth never propagated when loading a
persisted catalog).
+ propagateRestAuthIfNeeded(icebergCatalogOptions, catalogType);
+ translateJdbcBackendToCatalogImpl(icebergCatalogOptions, catalogType);
+ // The outer Flink factory is `gravitino-iceberg`, but the nested Iceberg
factory still expects
+ // `catalog-type=iceberg` when building the native Iceberg catalog
instance.
+ icebergCatalogOptions.put(CommonCatalogOptions.CATALOG_TYPE.key(),
"iceberg");
+ return icebergCatalogOptions;
+ }
+
+ // Copies catalog-backend into catalog-type so the CREATE CATALOG path lines
up with the
+ // USE CATALOG path, where the rename already happened upstream. Skipped
when catalog-type or
+ // catalog-impl is already set, otherwise an explicitly provided
catalog-impl would conflict
+ // with it. Returns the effective backend type for downstream gating: the
normalized catalog-type
+ // when present, or catalog-backend as a fallback (e.g. when catalog-impl
was already set).
+ private static String normalizeCatalogType(
+ Map<String, String> icebergCatalogOptions, String catalogBackend) {
if (catalogBackend != null
&&
!icebergCatalogOptions.containsKey(IcebergPropertiesConstants.ICEBERG_CATALOG_TYPE)
&&
!icebergCatalogOptions.containsKey(IcebergPropertiesConstants.ICEBERG_CATALOG_IMPL))
{
icebergCatalogOptions.put(IcebergPropertiesConstants.ICEBERG_CATALOG_TYPE,
catalogBackend);
}
- // A REST backend connects directly to the Iceberg REST service, bypassing
the Gravitino
- // server's auth proxy, so propagate the Gravitino client's authentication
to the REST client,
- // unless the user has already configured REST auth explicitly.
- if
(IcebergPropertiesConstants.ICEBERG_CATALOG_BACKEND_REST.equalsIgnoreCase(catalogBackend)
+ return icebergCatalogOptions.getOrDefault(
+ IcebergPropertiesConstants.ICEBERG_CATALOG_TYPE, catalogBackend);
+ }
+
+ // A REST backend connects directly to the Iceberg REST service, bypassing
the Gravitino
+ // server's auth proxy, so the Gravitino client's authentication must be
propagated to the REST
+ // client, unless the user has already configured REST auth explicitly.
+ private static void propagateRestAuthIfNeeded(
+ Map<String, String> icebergCatalogOptions, String catalogType) {
+ if
(IcebergPropertiesConstants.ICEBERG_CATALOG_BACKEND_REST.equalsIgnoreCase(catalogType)
&& !icebergCatalogOptions.containsKey(AuthProperties.AUTH_TYPE)) {
icebergCatalogOptions.putAll(
IcebergPropertiesConverter.INSTANCE.toRestAuthProperties(
GravitinoCatalogManager.get().getGravitinoClientConfig()));
}
- // Iceberg's FlinkCatalogFactory only accepts hive/hadoop/rest as
`catalog-type`; a JDBC backend
- // must be loaded through `catalog-impl` instead. The two keys are
mutually exclusive, so drop
- // `catalog-type` and use `putIfAbsent` to respect an explicitly provided
`catalog-impl`.
- String catalogType =
icebergCatalogOptions.get(IcebergPropertiesConstants.ICEBERG_CATALOG_TYPE);
+ }
+
+ // Iceberg's FlinkCatalogFactory only accepts hive/hadoop/rest as
catalog-type; a JDBC backend
+ // must be loaded through catalog-impl instead. The two keys are mutually
exclusive, so
+ // catalog-type is dropped, and putIfAbsent respects an explicitly provided
catalog-impl.
+ private static void translateJdbcBackendToCatalogImpl(
+ Map<String, String> icebergCatalogOptions, String catalogType) {
if
(IcebergPropertiesConstants.ICEBERG_CATALOG_BACKEND_JDBC.equalsIgnoreCase(catalogType))
{
icebergCatalogOptions.remove(IcebergPropertiesConstants.ICEBERG_CATALOG_TYPE);
icebergCatalogOptions.putIfAbsent(
IcebergPropertiesConstants.ICEBERG_CATALOG_IMPL,
IcebergPropertiesConstants.ICEBERG_JDBC_CATALOG_IMPL);
}
- // The outer Flink factory is `gravitino-iceberg`, but the nested Iceberg
factory still expects
- // `catalog-type=iceberg` when building the native Iceberg catalog
instance.
- icebergCatalogOptions.put(CommonCatalogOptions.CATALOG_TYPE.key(),
"iceberg");
- return icebergCatalogOptions;
}
}
diff --git
a/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/iceberg/TestGravitinoIcebergCatalogFactory.java
b/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/iceberg/TestGravitinoIcebergCatalogFactory.java
index 6b89862ab3..b65dfb2191 100644
---
a/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/iceberg/TestGravitinoIcebergCatalogFactory.java
+++
b/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/iceberg/TestGravitinoIcebergCatalogFactory.java
@@ -22,6 +22,7 @@ package org.apache.gravitino.flink.connector.iceberg;
import com.google.common.collect.ImmutableMap;
import java.util.Map;
import org.apache.flink.table.catalog.CommonCatalogOptions;
+import org.apache.iceberg.rest.auth.AuthProperties;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -105,7 +106,11 @@ class TestGravitinoIcebergCatalogFactory {
IcebergPropertiesConstants.ICEBERG_CATALOG_TYPE,
IcebergPropertiesConstants.ICEBERG_CATALOG_BACKEND_REST,
IcebergPropertiesConstants.ICEBERG_CATALOG_URI,
- "http://localhost:9001/iceberg/";);
+ "http://localhost:9001/iceberg/";,
+ // Pre-set so the REST auth-propagation branch (which needs a live
+ // GravitinoCatalogManager) is skipped; that branch is covered
separately below.
+ AuthProperties.AUTH_TYPE,
+ "none");
Map<String, String> result = factory.toIcebergCatalogOptions(options);
@@ -115,4 +120,27 @@ class TestGravitinoIcebergCatalogFactory {
Assertions.assertFalse(result.containsKey(IcebergPropertiesConstants.ICEBERG_CATALOG_IMPL));
Assertions.assertEquals("iceberg",
result.get(CommonCatalogOptions.CATALOG_TYPE.key()));
}
+
+ @Test
+ void testRestAuthPropagatedWhenLoadedFromCatalogStore() {
+ // Simulates the catalog-store load path (`USE CATALOG ...`): by the time
options reach this
+ // factory, IcebergPropertiesConverter has already renamed
`catalog-backend` -> `catalog-type`
+ // during property conversion, so only `catalog-type=rest` is present, no
`catalog-backend`.
+ Map<String, String> options =
+ ImmutableMap.of(
+ IcebergPropertiesConstants.ICEBERG_CATALOG_TYPE,
+ IcebergPropertiesConstants.ICEBERG_CATALOG_BACKEND_REST,
+ IcebergPropertiesConstants.ICEBERG_CATALOG_URI,
+ "http://localhost:9001/iceberg/";);
+
+ // Regression check for #11601: the REST auth-propagation branch must be
gated on the
+ // normalized `catalog-type`, not the renamed-away `catalog-backend` key,
otherwise it would
+ // silently skip auth propagation on this path. GravitinoCatalogManager
isn't initialized in
+ // this unit test, so reaching it (and failing there) proves the gate now
fires correctly.
+ IllegalStateException exception =
+ Assertions.assertThrows(
+ IllegalStateException.class, () ->
factory.toIcebergCatalogOptions(options));
+ String msg = exception.getMessage();
+ Assertions.assertTrue(msg != null &&
msg.contains("GravitinoCatalogManager"));
+ }
}
diff --git
a/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/integration/test/iceberg/FlinkIcebergJdbcCatalogIT.java
b/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/integration/test/iceberg/FlinkIcebergJdbcCatalogIT.java
index cdd01e6240..2be4e715ca 100644
---
a/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/integration/test/iceberg/FlinkIcebergJdbcCatalogIT.java
+++
b/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/integration/test/iceberg/FlinkIcebergJdbcCatalogIT.java
@@ -23,8 +23,6 @@ import com.google.common.collect.Maps;
import java.sql.SQLException;
import java.util.Map;
import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants;
-import org.apache.gravitino.credential.CredentialConstants;
-import org.apache.gravitino.credential.JdbcCredential;
import
org.apache.gravitino.flink.connector.iceberg.GravitinoIcebergCatalogFactoryOptions;
import org.apache.gravitino.flink.connector.iceberg.IcebergPropertiesConstants;
import org.apache.gravitino.integration.test.container.ContainerSuite;
@@ -61,36 +59,19 @@ public abstract class FlinkIcebergJdbcCatalogIT extends
FlinkIcebergCatalogIT {
catalogProperties.put(IcebergConstants.GRAVITINO_JDBC_PASSWORD,
mySQLContainer.getPassword());
catalogProperties.put(IcebergConstants.GRAVITINO_JDBC_DRIVER,
getDriverClassName());
catalogProperties.put(IcebergConstants.IO_IMPL,
"org.apache.iceberg.hadoop.HadoopFileIO");
- // Align the catalog name that the server-side and Flink-side native
Iceberg JdbcCatalog use for
- // the `catalog_name` column in the JDBC backend tables. The Flink-side
native catalog is named
- // after the Flink catalog (the Gravitino catalog name), while the
server-side defaults to the
- // `catalog-backend` value (`jdbc`). Without this, tables created through
Gravitino are
- // invisible
- // to the Flink-side reads/writes. The table read/write tests run on the
default catalog.
+ // Align catalog_name between server-side and Flink-side JdbcCatalog so
tables created through
+ // Gravitino are visible to the Flink-side reads/writes.
catalogProperties.put(IcebergConstants.CATALOG_BACKEND_NAME,
DEFAULT_ICEBERG_CATALOG);
- // Gravitino hides `jdbc-user`/`jdbc-password`, so they never reach the
Flink-side native
- // Iceberg
- // catalog through the loaded properties. Enable credential vending so the
server hands the JDBC
- // user/password to the client, which GravitinoIcebergCatalog.open()
injects into the native
- // catalog via CredentialPropertyUtils.applyIcebergCredentials. This
covers all access to an
- // already-created catalog.
- catalogProperties.put(
- CredentialConstants.CREDENTIAL_PROVIDERS,
JdbcCredential.JDBC_CREDENTIAL_TYPE);
- // Vending cannot help at CREATE time: Flink opens the catalog (and
Iceberg's JdbcCatalog
- // eagerly
- // connects to the database) before the catalog is persisted in Gravitino,
so getCredentials()
- // finds nothing to vend. The descriptor-based create path therefore needs
the native
- // `jdbc.user`/`jdbc.password` directly. See
testCreateGravitinoIcebergCatalog.
+ // `credential-providers` is intentionally omitted: the lakehouse-iceberg
catalog auto-enables
+ // `jdbc-user-password` for JDBC backends, so
GravitinoIcebergCatalog.open() injects the
+ // credentials via credential vending. This exercises the real
auto-vending path.
+ // `jdbc.user`/`jdbc.password` are only needed for CREATE CATALOG, where
Flink opens the
+ // catalog before it is persisted in Gravitino and vending is not yet
available.
catalogProperties.put("jdbc.user", mySQLContainer.getUsername());
catalogProperties.put("jdbc.password", mySQLContainer.getPassword());
return catalogProperties;
}
- /**
- * The base WITH clause only carries backend/uri/warehouse, but a JDBC
backend also needs the jdbc
- * driver, credential vending, and native jdbc credentials (the latter for
the create-time eager
- * connect, since vending is not available until the catalog is persisted in
Gravitino).
- */
@Override
protected String buildCreateCatalogSql(String catalogName) {
return String.format(
@@ -102,7 +83,6 @@ public abstract class FlinkIcebergJdbcCatalogIT extends
FlinkIcebergCatalogIT {
+ "'jdbc-user'='%s',"
+ "'jdbc-password'='%s',"
+ "'jdbc-driver'='%s',"
- + "'%s'='%s',"
+ "'jdbc.user'='%s',"
+ "'jdbc.password'='%s'"
+ ")",
@@ -114,8 +94,6 @@ public abstract class FlinkIcebergJdbcCatalogIT extends
FlinkIcebergCatalogIT {
mySQLContainer.getUsername(),
mySQLContainer.getPassword(),
getDriverClassName(),
- CredentialConstants.CREDENTIAL_PROVIDERS,
- JdbcCredential.JDBC_CREDENTIAL_TYPE,
mySQLContainer.getUsername(),
mySQLContainer.getPassword());
}
