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 e0ff8a90f0 [Cherry-pick to branch-1.3] [#11596] fix(flink-connector):
Support Iceberg catalog with JDBC backend (#11636) (#11656)
e0ff8a90f0 is described below
commit e0ff8a90f03472b74fcb2be4ec2a184a217bb2d3
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Tue Jun 16 10:26:51 2026 +0800
[Cherry-pick to branch-1.3] [#11596] fix(flink-connector): Support Iceberg
catalog with JDBC backend (#11636) (#11656)
**Cherry-pick Information:**
- Original commit: ba92cf1d60e1286efb3fe59c7eeadd9fab81629e
- Target branch: `branch-1.3`
- Status: ✅ Clean cherry-pick (no conflicts)
Co-authored-by: Yuhui <[email protected]>
Co-authored-by: Qi Yu <[email protected]>
---
.../iceberg/GravitinoIcebergCatalogFactory.java | 19 ++-
.../iceberg/IcebergPropertiesConstants.java | 6 +
.../TestGravitinoIcebergCatalogFactory.java | 118 +++++++++++++++++
.../test/iceberg/FlinkIcebergCatalogIT.java | 38 +++---
.../test/iceberg/FlinkIcebergJdbcCatalogIT.java | 140 +++++++++++++++++++++
.../test/iceberg/FlinkIcebergJdbcCatalogIT118.java | 27 ++++
.../test/iceberg/FlinkIcebergJdbcCatalogIT119.java | 27 ++++
.../test/iceberg/FlinkIcebergJdbcCatalogIT120.java | 29 +++++
.../integration/test/util/TestDatabaseName.java | 3 +
9 files changed, 391 insertions(+), 16 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 58644081ec..3337b469d2 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
@@ -18,6 +18,7 @@
*/
package org.apache.gravitino.flink.connector.iceberg;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Maps;
import java.util.Collections;
import java.util.Map;
@@ -114,14 +115,28 @@ public class GravitinoIcebergCatalogFactory implements
BaseCatalogFactory {
return DefaultPartitionConverter.INSTANCE;
}
- private Map<String, String> toIcebergCatalogOptions(Map<String, String>
catalogOptions) {
+ @VisibleForTesting
+ Map<String, String> toIcebergCatalogOptions(Map<String, String>
catalogOptions) {
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.
if (catalogBackend != null
- &&
!icebergCatalogOptions.containsKey(IcebergPropertiesConstants.ICEBERG_CATALOG_TYPE))
{
+ &&
!icebergCatalogOptions.containsKey(IcebergPropertiesConstants.ICEBERG_CATALOG_TYPE)
+ &&
!icebergCatalogOptions.containsKey(IcebergPropertiesConstants.ICEBERG_CATALOG_IMPL))
{
icebergCatalogOptions.put(IcebergPropertiesConstants.ICEBERG_CATALOG_TYPE,
catalogBackend);
}
+ // 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);
+ 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");
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 163cfac882..d879bb10e0 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
@@ -31,6 +31,12 @@ public class IcebergPropertiesConstants {
public static final String ICEBERG_CATALOG_TYPE =
FlinkCatalogFactory.ICEBERG_CATALOG_TYPE;
+ public static final String ICEBERG_CATALOG_IMPL =
CatalogProperties.CATALOG_IMPL;
+
+ public static final String ICEBERG_CATALOG_BACKEND_JDBC = "jdbc";
+
+ public static final String ICEBERG_JDBC_CATALOG_IMPL =
"org.apache.iceberg.jdbc.JdbcCatalog";
+
public static final String GRAVITINO_ICEBERG_CATALOG_WAREHOUSE =
IcebergConstants.WAREHOUSE;
public static final String ICEBERG_CATALOG_WAREHOUSE =
CatalogProperties.WAREHOUSE_LOCATION;
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
new file mode 100644
index 0000000000..6b89862ab3
--- /dev/null
+++
b/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/iceberg/TestGravitinoIcebergCatalogFactory.java
@@ -0,0 +1,118 @@
+/*
+ * 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 com.google.common.collect.ImmutableMap;
+import java.util.Map;
+import org.apache.flink.table.catalog.CommonCatalogOptions;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+class TestGravitinoIcebergCatalogFactory {
+
+ private final GravitinoIcebergCatalogFactory factory = new
GravitinoIcebergCatalogFactory();
+
+ @Test
+ void testJdbcBackendTranslatedToCatalogImpl() {
+ Map<String, String> options =
+ ImmutableMap.of(
+ IcebergPropertiesConstants.ICEBERG_CATALOG_TYPE,
+ IcebergPropertiesConstants.ICEBERG_CATALOG_BACKEND_JDBC,
+ IcebergPropertiesConstants.ICEBERG_CATALOG_URI,
+ "jdbc:mysql://localhost:3306/gravitino",
+ IcebergPropertiesConstants.ICEBERG_CATALOG_WAREHOUSE,
+ "hdfs://localhost:9000/user/hive/warehouse",
+ "jdbc.user",
+ "iceberg",
+ "jdbc.password",
+ "iceberg");
+
+ Map<String, String> result = factory.toIcebergCatalogOptions(options);
+
+ // JDBC backend must be loaded through catalog-impl, not catalog-type.
+ Assertions.assertEquals(
+ IcebergPropertiesConstants.ICEBERG_JDBC_CATALOG_IMPL,
+ result.get(IcebergPropertiesConstants.ICEBERG_CATALOG_IMPL));
+ Assertions.assertFalse(
+ result.containsKey(IcebergPropertiesConstants.ICEBERG_CATALOG_TYPE),
+ "catalog-type and catalog-impl are mutually exclusive");
+ // JDBC connection properties are preserved.
+ Assertions.assertEquals("iceberg", result.get("jdbc.user"));
+ Assertions.assertEquals("iceberg", result.get("jdbc.password"));
+ Assertions.assertEquals(
+ "jdbc:mysql://localhost:3306/gravitino",
+ result.get(IcebergPropertiesConstants.ICEBERG_CATALOG_URI));
+ Assertions.assertEquals("iceberg",
result.get(CommonCatalogOptions.CATALOG_TYPE.key()));
+ }
+
+ @Test
+ void testExplicitCatalogImplIsRespected() {
+ Map<String, String> options =
+ ImmutableMap.of(
+ IcebergPropertiesConstants.ICEBERG_CATALOG_TYPE,
+ IcebergPropertiesConstants.ICEBERG_CATALOG_BACKEND_JDBC,
+ IcebergPropertiesConstants.ICEBERG_CATALOG_IMPL,
+ "com.example.CustomJdbcCatalog");
+
+ Map<String, String> result = factory.toIcebergCatalogOptions(options);
+
+ // An explicitly provided catalog-impl must not be overwritten, and
catalog-type is dropped.
+ Assertions.assertEquals(
+ "com.example.CustomJdbcCatalog",
+ result.get(IcebergPropertiesConstants.ICEBERG_CATALOG_IMPL));
+
Assertions.assertFalse(result.containsKey(IcebergPropertiesConstants.ICEBERG_CATALOG_TYPE));
+ }
+
+ @Test
+ void testHiveBackendKeepsCatalogType() {
+ Map<String, String> options =
+ ImmutableMap.of(
+ IcebergPropertiesConstants.ICEBERG_CATALOG_TYPE,
+ IcebergPropertiesConstants.ICEBERG_CATALOG_BACKEND_HIVE,
+ IcebergPropertiesConstants.ICEBERG_CATALOG_URI,
+ "thrift://localhost:9083");
+
+ Map<String, String> result = factory.toIcebergCatalogOptions(options);
+
+ Assertions.assertEquals(
+ IcebergPropertiesConstants.ICEBERG_CATALOG_BACKEND_HIVE,
+ result.get(IcebergPropertiesConstants.ICEBERG_CATALOG_TYPE));
+
Assertions.assertFalse(result.containsKey(IcebergPropertiesConstants.ICEBERG_CATALOG_IMPL));
+ Assertions.assertEquals("iceberg",
result.get(CommonCatalogOptions.CATALOG_TYPE.key()));
+ }
+
+ @Test
+ void testRestBackendKeepsCatalogType() {
+ Map<String, String> options =
+ ImmutableMap.of(
+ IcebergPropertiesConstants.ICEBERG_CATALOG_TYPE,
+ IcebergPropertiesConstants.ICEBERG_CATALOG_BACKEND_REST,
+ IcebergPropertiesConstants.ICEBERG_CATALOG_URI,
+ "http://localhost:9001/iceberg/";);
+
+ Map<String, String> result = factory.toIcebergCatalogOptions(options);
+
+ Assertions.assertEquals(
+ IcebergPropertiesConstants.ICEBERG_CATALOG_BACKEND_REST,
+ result.get(IcebergPropertiesConstants.ICEBERG_CATALOG_TYPE));
+
Assertions.assertFalse(result.containsKey(IcebergPropertiesConstants.ICEBERG_CATALOG_IMPL));
+ Assertions.assertEquals("iceberg",
result.get(CommonCatalogOptions.CATALOG_TYPE.key()));
+ }
+}
diff --git
a/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/integration/test/iceberg/FlinkIcebergCatalogIT.java
b/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/integration/test/iceberg/FlinkIcebergCatalogIT.java
index 8bbbe32e56..fb1d8a925a 100644
---
a/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/integration/test/iceberg/FlinkIcebergCatalogIT.java
+++
b/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/integration/test/iceberg/FlinkIcebergCatalogIT.java
@@ -58,7 +58,7 @@ import org.junit.jupiter.api.Test;
public abstract class FlinkIcebergCatalogIT extends FlinkCommonIT {
- private static final String DEFAULT_ICEBERG_CATALOG =
"flink_iceberg_catalog";
+ protected static final String DEFAULT_ICEBERG_CATALOG =
"flink_iceberg_catalog";
private static org.apache.gravitino.Catalog icebergCatalog;
@@ -146,19 +146,7 @@ public abstract class FlinkIcebergCatalogIT extends
FlinkCommonIT {
// Create a new catalog.
String catalogName = "gravitino_iceberg_using_sql";
- tableEnv.executeSql(
- String.format(
- "create catalog %s with ("
- + "'type'='%s', "
- + "'catalog-backend'='%s',"
- + "'uri'='%s',"
- + "'warehouse'='%s'"
- + ")",
- catalogName,
- GravitinoIcebergCatalogFactoryOptions.IDENTIFIER,
- getCatalogBackend(),
- getUri(),
- warehouse));
+ tableEnv.executeSql(buildCreateCatalogSql(catalogName));
Assertions.assertTrue(metalake.catalogExists(catalogName));
// Check the properties of the created catalog.
@@ -505,4 +493,26 @@ public abstract class FlinkIcebergCatalogIT extends
FlinkCommonIT {
protected abstract String getCatalogBackend();
protected abstract String getUri();
+
+ /**
+ * Builds the {@code CREATE CATALOG} SQL used by {@link
#testCreateGravitinoIcebergUsingSQL()}.
+ * Subclasses override this to add backend-specific options (e.g. JDBC
credentials).
+ *
+ * @param catalogName the catalog name to create.
+ * @return the {@code CREATE CATALOG} SQL statement.
+ */
+ protected String buildCreateCatalogSql(String catalogName) {
+ return String.format(
+ "create catalog %s with ("
+ + "'type'='%s', "
+ + "'catalog-backend'='%s',"
+ + "'uri'='%s',"
+ + "'warehouse'='%s'"
+ + ")",
+ catalogName,
+ GravitinoIcebergCatalogFactoryOptions.IDENTIFIER,
+ getCatalogBackend(),
+ getUri(),
+ warehouse);
+ }
}
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
new file mode 100644
index 0000000000..cdd01e6240
--- /dev/null
+++
b/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/integration/test/iceberg/FlinkIcebergJdbcCatalogIT.java
@@ -0,0 +1,140 @@
+/*
+ * 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.integration.test.iceberg;
+
+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;
+import org.apache.gravitino.integration.test.container.MySQLContainer;
+import org.apache.gravitino.integration.test.util.TestDatabaseName;
+import org.junit.jupiter.api.Tag;
+
+/** Iceberg catalog with a JDBC (MySQL) metadata backend. */
+@Tag("gravitino-docker-test")
+public abstract class FlinkIcebergJdbcCatalogIT extends FlinkIcebergCatalogIT {
+
+ private static final TestDatabaseName TEST_DB_NAME =
+ TestDatabaseName.FLINK_ICEBERG_JDBC_CATALOG_IT;
+
+ private static MySQLContainer mySQLContainer;
+
+ @Override
+ protected void initCatalogEnv() throws Exception {
+ ContainerSuite containerSuite = ContainerSuite.getInstance();
+ containerSuite.startMySQLContainer(TEST_DB_NAME);
+ mySQLContainer = containerSuite.getMySQLContainer();
+ }
+
+ @Override
+ protected Map<String, String> getCatalogConfigs() {
+ Map<String, String> catalogProperties = Maps.newHashMap();
+ catalogProperties.put(
+ IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_BACKEND,
+ IcebergPropertiesConstants.ICEBERG_CATALOG_BACKEND_JDBC);
+
catalogProperties.put(IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_URI,
getUri());
+ catalogProperties.put(
+ IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_WAREHOUSE,
warehouse);
+ catalogProperties.put(IcebergConstants.GRAVITINO_JDBC_USER,
mySQLContainer.getUsername());
+ 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.
+ 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.
+ 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(
+ "create catalog %s with ("
+ + "'type'='%s', "
+ + "'catalog-backend'='%s',"
+ + "'uri'='%s',"
+ + "'warehouse'='%s',"
+ + "'jdbc-user'='%s',"
+ + "'jdbc-password'='%s',"
+ + "'jdbc-driver'='%s',"
+ + "'%s'='%s',"
+ + "'jdbc.user'='%s',"
+ + "'jdbc.password'='%s'"
+ + ")",
+ catalogName,
+ GravitinoIcebergCatalogFactoryOptions.IDENTIFIER,
+ getCatalogBackend(),
+ getUri(),
+ warehouse,
+ mySQLContainer.getUsername(),
+ mySQLContainer.getPassword(),
+ getDriverClassName(),
+ CredentialConstants.CREDENTIAL_PROVIDERS,
+ JdbcCredential.JDBC_CREDENTIAL_TYPE,
+ mySQLContainer.getUsername(),
+ mySQLContainer.getPassword());
+ }
+
+ @Override
+ protected String getCatalogBackend() {
+ return IcebergPropertiesConstants.ICEBERG_CATALOG_BACKEND_JDBC;
+ }
+
+ @Override
+ protected String getUri() {
+ return mySQLContainer.getJdbcUrl(TEST_DB_NAME);
+ }
+
+ private String getDriverClassName() {
+ try {
+ return mySQLContainer.getDriverClassName(TEST_DB_NAME);
+ } catch (SQLException e) {
+ throw new RuntimeException("Failed to get MySQL driver class name", e);
+ }
+ }
+}
diff --git
a/flink-connector/v1.18/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/iceberg/FlinkIcebergJdbcCatalogIT118.java
b/flink-connector/v1.18/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/iceberg/FlinkIcebergJdbcCatalogIT118.java
new file mode 100644
index 0000000000..10d082f5f6
--- /dev/null
+++
b/flink-connector/v1.18/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/iceberg/FlinkIcebergJdbcCatalogIT118.java
@@ -0,0 +1,27 @@
+/*
+ * 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.integration.test.iceberg;
+
+import org.junit.jupiter.api.condition.DisabledIf;
+
+// Flink 1.18 uses Iceberg 1.9.x; lakehouse-iceberg (1.11) runs in embedded
MiniGravitino in the
+// same JVM. Run JDBC-backend Iceberg IT in deploy mode only.
+@DisabledIf("org.apache.gravitino.integration.test.util.ITUtils#isEmbedded")
+public class FlinkIcebergJdbcCatalogIT118 extends FlinkIcebergJdbcCatalogIT {}
diff --git
a/flink-connector/v1.19/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/iceberg/FlinkIcebergJdbcCatalogIT119.java
b/flink-connector/v1.19/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/iceberg/FlinkIcebergJdbcCatalogIT119.java
new file mode 100644
index 0000000000..9ec9d6e9ff
--- /dev/null
+++
b/flink-connector/v1.19/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/iceberg/FlinkIcebergJdbcCatalogIT119.java
@@ -0,0 +1,27 @@
+/*
+ * 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.integration.test.iceberg;
+
+import org.junit.jupiter.api.condition.DisabledIf;
+
+// Flink 1.19 uses Iceberg 1.10.x; lakehouse-iceberg (1.11) runs in embedded
MiniGravitino in the
+// same JVM. Run JDBC-backend Iceberg IT in deploy mode only.
+@DisabledIf("org.apache.gravitino.integration.test.util.ITUtils#isEmbedded")
+public class FlinkIcebergJdbcCatalogIT119 extends FlinkIcebergJdbcCatalogIT {}
diff --git
a/flink-connector/v1.20/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/iceberg/FlinkIcebergJdbcCatalogIT120.java
b/flink-connector/v1.20/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/iceberg/FlinkIcebergJdbcCatalogIT120.java
new file mode 100644
index 0000000000..cef41b29ac
--- /dev/null
+++
b/flink-connector/v1.20/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/iceberg/FlinkIcebergJdbcCatalogIT120.java
@@ -0,0 +1,29 @@
+/*
+ * 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.integration.test.iceberg;
+
+import org.junit.jupiter.api.condition.DisabledIf;
+
+// Even though Flink 1.20 and lakehouse-iceberg both use Iceberg 1.11, a JDBC
backend exercises the
+// server-side Iceberg JdbcCatalog, which hits a cross-classloader
IllegalAccessError when the
+// embedded MiniGravitino server shares the JVM with the Flink Iceberg
runtime. Run in deploy mode
+// only. @DisabledIf is not @Inherited, so each concrete subclass must declare
it explicitly.
+@DisabledIf("org.apache.gravitino.integration.test.util.ITUtils#isEmbedded")
+public class FlinkIcebergJdbcCatalogIT120 extends FlinkIcebergJdbcCatalogIT {}
diff --git
a/integration-test-common/src/test/java/org/apache/gravitino/integration/test/util/TestDatabaseName.java
b/integration-test-common/src/test/java/org/apache/gravitino/integration/test/util/TestDatabaseName.java
index 39b644eb65..8e7db655b6 100644
---
a/integration-test-common/src/test/java/org/apache/gravitino/integration/test/util/TestDatabaseName.java
+++
b/integration-test-common/src/test/java/org/apache/gravitino/integration/test/util/TestDatabaseName.java
@@ -136,4 +136,7 @@ public enum TestDatabaseName {
}
},
FLINK_HIVE_CATALOG_IT,
+
+ /** Represents the MySQL database for the Flink Iceberg JDBC-backend catalog
integration test. */
+ FLINK_ICEBERG_JDBC_CATALOG_IT,
}
