This is an automated email from the ASF dual-hosted git repository.
yuqi1129 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 2fc3388049 [#9710] feat(flink-connector): add Flink 1.19 and 1.20
support (#11071)
2fc3388049 is described below
commit 2fc3388049b3349352a9fab112b333d50a037cae
Author: FANNG <[email protected]>
AuthorDate: Fri May 22 13:23:53 2026 +0800
[#9710] feat(flink-connector): add Flink 1.19 and 1.20 support (#11071)
### What changes were proposed in this pull request?
This PR adds Gravitino Flink connector support for Flink 1.19 and Flink
1.20
on top of the versioned-layout baseline established for Flink 1.18
(#10517).
Main changes:
- Add `flink-connector/v1.19/flink` +
`flink-connector/v1.19/flink-runtime`
modules.
- Add `flink-connector/v1.20/flink` +
`flink-connector/v1.20/flink-runtime`
modules.
- Add version-specific catalog/factory entry classes (`*Flink119` /
`*Flink120`) for Hive, Iceberg, JDBC-MySQL, JDBC-Postgres, and Paimon.
- Reuse `DefaultCatalogCompat` for Flink 1.19 and add
`CatalogCompatFlink120`
that uses `CatalogTable.newBuilder(...)` because `CatalogTable.of(...)`
is
deprecated in Flink 1.20.
- The JDBC catalog for 1.19 / 1.20 uses the relocated
`org.apache.flink.connector.jdbc.core.*` factory packages shipped in
`flink-connector-jdbc 3.3.0-1.19` / `3.3.0-1.20`.
- Add provider-specific integration-test entry classes for each new
minor
version.
- Add Flink 1.20-specific overrides for missing required catalog options
where
Flink 1.20 reports `IllegalArgumentException` rather than the
`ValidationException` thrown by older minor versions.
- Extend the Flink CI job to run embedded and deploy mode integration
tests across 1.18, 1.19, and 1.20.
- Exclude the new Flink modules from unrelated Backend IT and Trino CI
jobs.
No `flink-common` production changes are required; the 1.18 baseline
already
exposes the necessary extension hooks (`BaseCatalog#catalogCompat()`,
the
protected `GravitinoJdbcCatalog(Context, ..., AbstractCatalog)`
constructor,
etc.). The shared Iceberg REST SQL integration test now lives in the
common
abstract IT base so the 1.18 / 1.19 / 1.20 entry classes can remain thin
stubs.
This PR supersedes #11061 (closed) and the now-stale #10501.
### Why are the changes needed?
The Flink 1.18 baseline (#10517) was intentionally limited to one minor
and
deferred 1.19 / 1.20 to follow-up work under parent issue #9710. This PR
is that follow-up, so users can target the two currently-maintained
Flink
minor versions.
Fix: #9710
### Does this PR introduce _any_ user-facing change?
Yes. Two new connector artifacts are produced:
- `gravitino-flink-connector-runtime-1.19_2.12`
- `gravitino-flink-connector-runtime-1.20_2.12`
The user-facing SQL syntax is unchanged.
### How was this patch tested?
Local validations (all green):
- `./gradlew spotlessApply`
- `./gradlew :flink-connector:flink-1.19:test -PskipITs`
- `./gradlew :flink-connector:flink-1.20:test -PskipITs`
- `./gradlew :flink-connector:flink-runtime-1.19:test`
- `./gradlew :flink-connector:flink-runtime-1.20:test`
- `./gradlew :flink-connector:flink-runtime-1.19:shadowJar
:flink-connector:flink-runtime-1.20:shadowJar`
- `./gradlew :spotlessKotlinGradleCheck :flink-connector:flink-1.20:test
-PskipITs`
- `./gradlew :flink-connector:flink-common:compileTestJava
:flink-connector:flink-1.19:compileJava
:flink-connector:flink-1.19:compileTestJava
:flink-connector:flink-1.20:compileTestJava
:flink-connector:flink-common:spotlessJavaCheck
:flink-connector:flink-1.19:spotlessJavaCheck
:flink-connector:flink-1.20:spotlessJavaCheck -PskipITs`
The runtime-jar tests verify that the version-specific SPI descriptors
(`META-INF/services/org.apache.flink.table.factories.Factory`) are
merged
into the shaded uber-jar.
The full Hive / Iceberg / JDBC / Paimon integration tests for both new
minor versions will run in CI via the updated
`.github/workflows/flink-integration-test-action.yml`.
---------
Co-authored-by: fanng <“[email protected]”>
---
.../workflows/backend-integration-test-action.yml | 5 +-
.../workflows/flink-integration-test-action.yml | 18 +-
.github/workflows/trino-integration-test.yml | 4 +
.github/workflows/trino-multi-version-test.yml | 4 +
.../test/iceberg/FlinkIcebergRestCatalogIT.java | 66 +++++++
.../test/iceberg/FlinkIcebergRestCatalogIT118.java | 75 +-------
.../v1.19/flink-runtime/build.gradle.kts | 90 +++++++++
.../flink/runtime/TestRuntimeJarDependencies.java | 80 ++++++++
flink-connector/v1.19/flink/build.gradle.kts | 207 +++++++++++++++++++++
.../hive/GravitinoHiveCatalogFactoryFlink119.java | 50 +++++
.../hive/GravitinoHiveCatalogFlink119.java | 55 ++++++
.../GravitinoIcebergCatalogFactoryFlink119.java | 46 +++++
.../iceberg/GravitinoIcebergCatalogFlink119.java | 59 ++++++
.../jdbc/GravitinoJdbcCatalogFlink119.java | 62 ++++++
.../GravitinoMysqlJdbcCatalogFactoryFlink119.java | 40 ++++
...ravitinoPostgresJdbcCatalogFactoryFlink119.java | 41 ++++
.../GravitinoPaimonCatalogFactoryFlink119.java | 38 ++++
.../paimon/GravitinoPaimonCatalogFlink119.java | 43 +++++
.../org.apache.flink.table.factories.Factory | 24 +++
.../test/catalog/GravitinoCatalogManagerIT119.java | 22 +++
.../test/hive/FlinkHiveCatalogIT119.java | 22 +++
.../test/hive/FlinkHiveKerberosClientIT119.java | 22 +++
.../test/iceberg/FlinkIcebergHiveCatalogIT119.java | 22 +++
.../test/iceberg/FlinkIcebergRestCatalogIT119.java | 22 +++
.../test/jdbc/FlinkJdbcMysqlCatalogIT119.java | 22 +++
.../test/paimon/FlinkPaimonHiveBackendIT119.java | 22 +++
.../test/paimon/FlinkPaimonJdbcBackendIT119.java | 22 +++
.../FlinkPaimonLocalFileSystemBackendIT119.java | 22 +++
.../v1.20/flink-runtime/build.gradle.kts | 90 +++++++++
.../flink/runtime/TestRuntimeJarDependencies.java | 80 ++++++++
flink-connector/v1.20/flink/build.gradle.kts | 207 +++++++++++++++++++++
.../hive/GravitinoHiveCatalogFactoryFlink120.java | 50 +++++
.../hive/GravitinoHiveCatalogFlink120.java | 55 ++++++
.../GravitinoIcebergCatalogFactoryFlink120.java | 46 +++++
.../iceberg/GravitinoIcebergCatalogFlink120.java | 59 ++++++
.../jdbc/GravitinoJdbcCatalogFlink120.java | 62 ++++++
.../GravitinoMysqlJdbcCatalogFactoryFlink120.java | 40 ++++
...ravitinoPostgresJdbcCatalogFactoryFlink120.java | 41 ++++
.../GravitinoPaimonCatalogFactoryFlink120.java | 38 ++++
.../paimon/GravitinoPaimonCatalogFlink120.java | 43 +++++
.../connector/utils/CatalogCompatFlink120.java | 49 +++++
.../org.apache.flink.table.factories.Factory | 24 +++
.../test/catalog/GravitinoCatalogManagerIT120.java | 22 +++
.../test/hive/FlinkHiveCatalogIT120.java | 78 ++++++++
.../test/hive/FlinkHiveKerberosClientIT120.java | 22 +++
.../test/iceberg/FlinkIcebergHiveCatalogIT120.java | 22 +++
.../test/iceberg/FlinkIcebergRestCatalogIT120.java | 22 +++
.../test/jdbc/FlinkJdbcMysqlCatalogIT120.java | 51 +++++
.../test/paimon/FlinkPaimonHiveBackendIT120.java | 22 +++
.../test/paimon/FlinkPaimonJdbcBackendIT120.java | 22 +++
.../FlinkPaimonLocalFileSystemBackendIT120.java | 22 +++
gradle/libs.versions.toml | 10 +
settings.gradle.kts | 15 +-
53 files changed, 2317 insertions(+), 80 deletions(-)
diff --git a/.github/workflows/backend-integration-test-action.yml
b/.github/workflows/backend-integration-test-action.yml
index 9e2239db13..09992506b1 100644
--- a/.github/workflows/backend-integration-test-action.yml
+++ b/.github/workflows/backend-integration-test-action.yml
@@ -80,7 +80,10 @@ jobs:
./gradlew test -PskipTests -PtestMode=${{ inputs.test-mode }}
-PjdbcBackend=${{ inputs.backend }} -PskipDockerTests=false -PskipWeb=true \
-x :web:web:test -x :web:integration-test:test -x :web-v2:web:test
-x :web-v2:integration-test:test -x :clients:client-python:test \
- -x :flink-connector:flink-common:test -x
:flink-connector:flink-1.18:test -x :flink-connector:flink-runtime-1.18:test \
+ -x :flink-connector:flink-common:test \
+ -x :flink-connector:flink-1.18:test -x
:flink-connector:flink-runtime-1.18:test \
+ -x :flink-connector:flink-1.19:test -x
:flink-connector:flink-runtime-1.19:test \
+ -x :flink-connector:flink-1.20:test -x
:flink-connector:flink-runtime-1.20:test \
-x :spark-connector:spark-common:test \
-x :spark-connector:spark-3.3:test -x
:spark-connector:spark-3.4:test -x :spark-connector:spark-3.5:test \
-x :spark-connector:spark-runtime-3.3:test -x
:spark-connector:spark-runtime-3.4:test -x
:spark-connector:spark-runtime-3.5:test \
diff --git a/.github/workflows/flink-integration-test-action.yml
b/.github/workflows/flink-integration-test-action.yml
index b4cc9416a4..257d6784f1 100644
--- a/.github/workflows/flink-integration-test-action.yml
+++ b/.github/workflows/flink-integration-test-action.yml
@@ -17,7 +17,7 @@ jobs:
start-runner:
name: JDK${{ inputs.java-version }}
runs-on: ubuntu-latest
- timeout-minutes: 60
+ timeout-minutes: 120
env:
PLATFORM: ${{ inputs.architecture }}
steps:
@@ -46,10 +46,16 @@ jobs:
- name: Flink Integration Test
id: integrationTest
- # run embedded mode and deploy mode integration tests
+ # run embedded mode and deploy mode integration tests for every
supported Flink minor
run: |
- ./gradlew -PskipTests -PtestMode=embedded -PskipDockerTests=false
-PskipWeb=true :flink-connector:flink-1.18:test --tests
"org.apache.gravitino.flink.connector.integration.test.**"
- ./gradlew -PskipTests -PtestMode=deploy -PskipDockerTests=false
-PskipWeb=true :flink-connector:flink-1.18:test --tests
"org.apache.gravitino.flink.connector.integration.test.**"
+ for v in 1.18 1.19 1.20; do
+ ./gradlew -PskipTests -PtestMode=embedded -PskipDockerTests=false
-PskipWeb=true \
+ :flink-connector:flink-$v:test \
+ --tests
"org.apache.gravitino.flink.connector.integration.test.**"
+ ./gradlew -PskipTests -PtestMode=deploy -PskipDockerTests=false
-PskipWeb=true \
+ :flink-connector:flink-$v:test \
+ --tests
"org.apache.gravitino.flink.connector.integration.test.**"
+ done
- name: Upload integrate tests reports
uses: actions/upload-artifact@v7
@@ -60,6 +66,10 @@ jobs:
build/reports
flink-connector/v1.18/flink/build/*.log
flink-connector/v1.18/flink/build/*.tar
+ flink-connector/v1.19/flink/build/*.log
+ flink-connector/v1.19/flink/build/*.tar
+ flink-connector/v1.20/flink/build/*.log
+ flink-connector/v1.20/flink/build/*.tar
distribution/package/logs/gravitino-server.out
distribution/package/logs/gravitino-server.log
catalogs/**/*.log
diff --git a/.github/workflows/trino-integration-test.yml
b/.github/workflows/trino-integration-test.yml
index 5911ac2e3f..7dd6213727 100644
--- a/.github/workflows/trino-integration-test.yml
+++ b/.github/workflows/trino-integration-test.yml
@@ -81,6 +81,10 @@ jobs:
-x :flink-connector:flink-common:build \
-x :flink-connector:flink-1.18:build \
-x :flink-connector:flink-runtime-1.18:build \
+ -x :flink-connector:flink-1.19:build \
+ -x :flink-connector:flink-runtime-1.19:build \
+ -x :flink-connector:flink-1.20:build \
+ -x :flink-connector:flink-runtime-1.20:build \
-x :spark-connector:spark-common:build \
-x :spark-connector:spark-3.3:build -x
:spark-connector:spark-3.4:build -x :spark-connector:spark-3.5:build \
-x :spark-connector:spark-runtime-3.3:build -x
:spark-connector:spark-runtime-3.4:build -x
:spark-connector:spark-runtime-3.5:build \
diff --git a/.github/workflows/trino-multi-version-test.yml
b/.github/workflows/trino-multi-version-test.yml
index f5a620d6c0..f9c55137f0 100644
--- a/.github/workflows/trino-multi-version-test.yml
+++ b/.github/workflows/trino-multi-version-test.yml
@@ -29,6 +29,10 @@ jobs:
-x :flink-connector:flink-common:build \
-x :flink-connector:flink-1.18:build \
-x :flink-connector:flink-runtime-1.18:build \
+ -x :flink-connector:flink-1.19:build \
+ -x :flink-connector:flink-runtime-1.19:build \
+ -x :flink-connector:flink-1.20:build \
+ -x :flink-connector:flink-runtime-1.20:build \
-x :spark-connector:spark-common:build \
-x :spark-connector:spark-3.3:build -x
:spark-connector:spark-3.4:build -x :spark-connector:spark-3.5:build \
-x :spark-connector:spark-runtime-3.3:build -x
:spark-connector:spark-runtime-3.4:build -x
:spark-connector:spark-runtime-3.5:build \
diff --git
a/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/integration/test/iceberg/FlinkIcebergRestCatalogIT.java
b/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/integration/test/iceberg/FlinkIcebergRestCatalogIT.java
index 4d77b6167b..197f75f288 100644
---
a/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/integration/test/iceberg/FlinkIcebergRestCatalogIT.java
+++
b/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/integration/test/iceberg/FlinkIcebergRestCatalogIT.java
@@ -19,11 +19,17 @@
package org.apache.gravitino.flink.connector.integration.test.iceberg;
+import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import java.util.Arrays;
import java.util.Map;
+import java.util.Optional;
import org.apache.flink.table.api.ResultKind;
+import org.apache.flink.table.api.TableResult;
import org.apache.flink.types.Row;
+import org.apache.gravitino.Catalog;
+import org.apache.gravitino.flink.connector.iceberg.GravitinoIcebergCatalog;
+import
org.apache.gravitino.flink.connector.iceberg.GravitinoIcebergCatalogFactoryOptions;
import org.apache.gravitino.flink.connector.iceberg.IcebergPropertiesConstants;
import org.apache.gravitino.flink.connector.integration.test.utils.TestUtils;
import org.junit.jupiter.api.Assertions;
@@ -48,6 +54,66 @@ public abstract class FlinkIcebergRestCatalogIT extends
FlinkIcebergCatalogIT {
return catalogProperties;
}
+ @Override
+ @Test
+ public void testCreateGravitinoIcebergUsingSQL() {
+ tableEnv.useCatalog(DEFAULT_CATALOG);
+ int numCatalogs = tableEnv.listCatalogs().length;
+
+ String catalogName = "gravitino_iceberg_using_sql";
+ tableEnv.executeSql(
+ String.format(
+ "create catalog %s with ("
+ + "'type'='%s', "
+ + "'catalog-backend'='%s',"
+ + "'uri'='%s'"
+ + ")",
+ catalogName,
+ GravitinoIcebergCatalogFactoryOptions.IDENTIFIER,
+ getCatalogBackend(),
+ getUri()));
+ Assertions.assertTrue(metalake.catalogExists(catalogName));
+
+ Catalog gravitinoCatalog = metalake.loadCatalog(catalogName);
+ Map<String, String> properties = gravitinoCatalog.properties();
+ Assertions.assertEquals(
+ getUri(),
properties.get(IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_URI));
+ Assertions.assertFalse(
+
properties.containsKey(IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_WAREHOUSE));
+
+ Optional<org.apache.flink.table.catalog.Catalog> catalog =
tableEnv.getCatalog(catalogName);
+ Assertions.assertTrue(catalog.isPresent());
+ Assertions.assertInstanceOf(GravitinoIcebergCatalog.class, catalog.get());
+
+ String[] catalogs = tableEnv.listCatalogs();
+ Assertions.assertEquals(numCatalogs + 1, catalogs.length, "Should create a
new catalog");
+ Assertions.assertTrue(
+ Arrays.asList(catalogs).contains(catalogName), "Should create the
correct catalog.");
+
+ TableResult result = tableEnv.executeSql("show catalogs");
+ Assertions.assertEquals(
+ numCatalogs + 1, Lists.newArrayList(result.collect()).size(), "Should
have 2 catalogs");
+
+ Assertions.assertEquals(
+ DEFAULT_CATALOG,
+ tableEnv.getCurrentCatalog(),
+ "Current catalog should be default_catalog in flink");
+
+ tableEnv.useCatalog(catalogName);
+ Assertions.assertEquals(
+ catalogName,
+ tableEnv.getCurrentCatalog(),
+ "Current catalog should be the one that is created just now.");
+
+ tableEnv.useCatalog(DEFAULT_CATALOG);
+ tableEnv.executeSql("drop catalog " + catalogName);
+ Assertions.assertFalse(metalake.catalogExists(catalogName));
+
+ Optional<org.apache.flink.table.catalog.Catalog> droppedCatalog =
+ tableEnv.getCatalog(catalogName);
+ Assertions.assertFalse(droppedCatalog.isPresent(), "Catalog should be
dropped");
+ }
+
@Override
@Test
public void testListSchema() {
diff --git
a/flink-connector/v1.18/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/iceberg/FlinkIcebergRestCatalogIT118.java
b/flink-connector/v1.18/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/iceberg/FlinkIcebergRestCatalogIT118.java
index c383100ca6..94c0803035 100644
---
a/flink-connector/v1.18/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/iceberg/FlinkIcebergRestCatalogIT118.java
+++
b/flink-connector/v1.18/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/iceberg/FlinkIcebergRestCatalogIT118.java
@@ -19,77 +19,4 @@
package org.apache.gravitino.flink.connector.integration.test.iceberg;
-import com.google.common.collect.Lists;
-import java.util.Arrays;
-import java.util.Map;
-import java.util.Optional;
-import org.apache.flink.table.api.TableResult;
-import org.apache.gravitino.Catalog;
-import org.apache.gravitino.flink.connector.iceberg.GravitinoIcebergCatalog;
-import
org.apache.gravitino.flink.connector.iceberg.GravitinoIcebergCatalogFactoryOptions;
-import org.apache.gravitino.flink.connector.iceberg.IcebergPropertiesConstants;
-import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.Test;
-
-public class FlinkIcebergRestCatalogIT118 extends FlinkIcebergRestCatalogIT {
-
- @Override
- @Test
- public void testCreateGravitinoIcebergUsingSQL() {
- tableEnv.useCatalog(DEFAULT_CATALOG);
- int numCatalogs = tableEnv.listCatalogs().length;
-
- String catalogName = "gravitino_iceberg_using_sql";
- tableEnv.executeSql(
- String.format(
- "create catalog %s with ("
- + "'type'='%s', "
- + "'catalog-backend'='%s',"
- + "'uri'='%s'"
- + ")",
- catalogName,
- GravitinoIcebergCatalogFactoryOptions.IDENTIFIER,
- getCatalogBackend(),
- getUri()));
- Assertions.assertTrue(metalake.catalogExists(catalogName));
-
- Catalog gravitinoCatalog = metalake.loadCatalog(catalogName);
- Map<String, String> properties = gravitinoCatalog.properties();
- Assertions.assertEquals(
- getUri(),
properties.get(IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_URI));
- Assertions.assertFalse(
-
properties.containsKey(IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_WAREHOUSE));
-
- Optional<org.apache.flink.table.catalog.Catalog> catalog =
tableEnv.getCatalog(catalogName);
- Assertions.assertTrue(catalog.isPresent());
- Assertions.assertInstanceOf(GravitinoIcebergCatalog.class, catalog.get());
-
- String[] catalogs = tableEnv.listCatalogs();
- Assertions.assertEquals(numCatalogs + 1, catalogs.length, "Should create a
new catalog");
- Assertions.assertTrue(
- Arrays.asList(catalogs).contains(catalogName), "Should create the
correct catalog.");
-
- TableResult result = tableEnv.executeSql("show catalogs");
- Assertions.assertEquals(
- numCatalogs + 1, Lists.newArrayList(result.collect()).size(), "Should
have 2 catalogs");
-
- Assertions.assertEquals(
- DEFAULT_CATALOG,
- tableEnv.getCurrentCatalog(),
- "Current catalog should be default_catalog in flink");
-
- tableEnv.useCatalog(catalogName);
- Assertions.assertEquals(
- catalogName,
- tableEnv.getCurrentCatalog(),
- "Current catalog should be the one that is created just now.");
-
- tableEnv.useCatalog(DEFAULT_CATALOG);
- tableEnv.executeSql("drop catalog " + catalogName);
- Assertions.assertFalse(metalake.catalogExists(catalogName));
-
- Optional<org.apache.flink.table.catalog.Catalog> droppedCatalog =
- tableEnv.getCatalog(catalogName);
- Assertions.assertFalse(droppedCatalog.isPresent(), "Catalog should be
dropped");
- }
-}
+public class FlinkIcebergRestCatalogIT118 extends FlinkIcebergRestCatalogIT {}
diff --git a/flink-connector/v1.19/flink-runtime/build.gradle.kts
b/flink-connector/v1.19/flink-runtime/build.gradle.kts
new file mode 100644
index 0000000000..f9792df561
--- /dev/null
+++ b/flink-connector/v1.19/flink-runtime/build.gradle.kts
@@ -0,0 +1,90 @@
+/*
+ * 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.
+ */
+import com.github.jengelman.gradle.plugins.shadow.tasks.ShadowJar
+
+plugins {
+ `maven-publish`
+ id("java")
+ id("idea")
+ alias(libs.plugins.shadow)
+}
+
+repositories {
+ mavenCentral()
+}
+
+val flinkVersion: String = libs.versions.flink119.get()
+val flinkMajorVersion: String = flinkVersion.substringBeforeLast(".")
+val scalaVersion: String = "2.12"
+val artifactName = "gravitino-${project.name}_$scalaVersion"
+val baseName =
"${rootProject.name}-flink-connector-runtime-${flinkMajorVersion}_$scalaVersion"
+
+configurations.all {
+ resolutionStrategy.eachDependency {
+ if (requested.group == "org.apache.logging.log4j") {
+ throw GradleException("Dependency 'org.apache.logging.log4j' is not
allowed.")
+ }
+ }
+}
+
+dependencies {
+ implementation(project(":clients:client-java-runtime", configuration =
"shadow"))
+ implementation(project(":flink-connector:flink-1.19"))
+
+ testImplementation(libs.junit.jupiter.api)
+ testRuntimeOnly(libs.junit.jupiter.engine)
+}
+
+val shadowJarTask = tasks.named<ShadowJar>("shadowJar")
+
+shadowJarTask.configure {
+ isZip64 = true
+ configurations = listOf(project.configurations.runtimeClasspath.get())
+ archiveFileName.set("$baseName-$version.jar")
+ archiveClassifier.set("")
+ mergeServiceFiles()
+
+ exclude("org/slf4j/**")
+ exclude("META-INF/maven/org.slf4j/**")
+
+ relocate("com.google", "org.apache.gravitino.shaded.com.google")
+ relocate("google", "org.apache.gravitino.shaded.google")
+ relocate("org.apache.hc", "org.apache.gravitino.shaded.org.apache.hc")
+}
+
+tasks.test {
+ useJUnitPlatform()
+ dependsOn(shadowJarTask)
+ doFirst {
+ systemProperty("shadowJarPath",
shadowJarTask.get().archiveFile.get().asFile.absolutePath)
+ }
+}
+
+publishing {
+ publications {
+ withType<MavenPublication>().configureEach {
+ artifactId = baseName
+ }
+ }
+}
+
+tasks.jar {
+ dependsOn(tasks.named("shadowJar"))
+ archiveClassifier.set("empty")
+}
diff --git
a/flink-connector/v1.19/flink-runtime/src/test/java/org/apache/gravitino/flink/runtime/TestRuntimeJarDependencies.java
b/flink-connector/v1.19/flink-runtime/src/test/java/org/apache/gravitino/flink/runtime/TestRuntimeJarDependencies.java
new file mode 100644
index 0000000000..014c6896ab
--- /dev/null
+++
b/flink-connector/v1.19/flink-runtime/src/test/java/org/apache/gravitino/flink/runtime/TestRuntimeJarDependencies.java
@@ -0,0 +1,80 @@
+/*
+ * 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.runtime;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.jar.JarEntry;
+import java.util.jar.JarFile;
+import java.util.stream.Collectors;
+import org.junit.jupiter.api.Test;
+
+class TestRuntimeJarDependencies {
+ private static final String FACTORY_SERVICE_ENTRY =
+ "META-INF/services/org.apache.flink.table.factories.Factory";
+
+ @Test
+ void shadowJarShouldNotBundleSlf4j() throws IOException {
+ String jarPath = System.getProperty("shadowJarPath");
+ assertNotNull(jarPath, "shadowJarPath system property should be provided
by the build");
+
+ File runtimeJar = new File(jarPath);
+ assertTrue(runtimeJar.exists(), "shadow jar does not exist: " +
runtimeJar);
+
+ try (JarFile jarFile = new JarFile(runtimeJar)) {
+ List<String> entries =
jarFile.stream().map(JarEntry::getName).collect(Collectors.toList());
+ boolean hasSlf4jClasses = entries.stream().anyMatch(entry ->
entry.startsWith("org/slf4j/"));
+ boolean hasSlf4jMetadata =
+ entries.stream().anyMatch(entry ->
entry.startsWith("META-INF/maven/org.slf4j/"));
+ assertFalse(
+ hasSlf4jClasses,
+ "Flink connector runtime jar should rely on Flink provided slf4j
instead of shading it");
+ assertFalse(hasSlf4jMetadata, "SLF4J metadata should not be packaged in
runtime jar");
+ }
+ }
+
+ @Test
+ void shadowJarShouldMergeFactoryServiceDescriptors() throws IOException {
+ String jarPath = System.getProperty("shadowJarPath");
+ assertNotNull(jarPath, "shadowJarPath system property should be provided
by the build");
+
+ try (JarFile jarFile = new JarFile(new File(jarPath))) {
+ JarEntry factoryServiceEntry =
jarFile.getJarEntry(FACTORY_SERVICE_ENTRY);
+ assertNotNull(factoryServiceEntry, "Factory service descriptor should
exist in runtime jar");
+
+ String factoryServices =
+ new String(
+ jarFile.getInputStream(factoryServiceEntry).readAllBytes(),
StandardCharsets.UTF_8);
+ assertTrue(
+ factoryServices.contains(
+
"org.apache.gravitino.flink.connector.store.GravitinoCatalogStoreFactory"),
+ "Runtime jar should include the shared catalog-store factory
descriptor");
+ assertTrue(
+ factoryServices.contains(
+
"org.apache.gravitino.flink.connector.hive.GravitinoHiveCatalogFactoryFlink119"),
+ "Runtime jar should include the version-specific catalog factory
descriptor");
+ }
+ }
+}
diff --git a/flink-connector/v1.19/flink/build.gradle.kts
b/flink-connector/v1.19/flink/build.gradle.kts
new file mode 100644
index 0000000000..9ca0201df6
--- /dev/null
+++ b/flink-connector/v1.19/flink/build.gradle.kts
@@ -0,0 +1,207 @@
+/*
+ * 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.
+ */
+import org.gradle.api.tasks.SourceSetContainer
+
+plugins {
+ `maven-publish`
+ id("java")
+ id("idea")
+}
+
+repositories {
+ mavenCentral()
+}
+
+val commonProject = project(":flink-connector:flink-common")
+val commonSourceSets = commonProject.extensions.getByType<SourceSetContainer>()
+val commonTestOutput = commonSourceSets.named("test").get().output
+val flinkVersion: String = libs.versions.flink119.get()
+val flinkMajorVersion: String = flinkVersion.substringBeforeLast(".")
+val icebergVersion: String = libs.versions.iceberg4flink119.get()
+val paimonVersion: String = libs.versions.paimon4flink119.get()
+val scalaVersion: String = "2.12"
+val artifactName =
"${rootProject.name}-flink-${flinkMajorVersion}_$scalaVersion"
+
+dependencies {
+ implementation(commonProject)
+
+ compileOnly(project(":clients:client-java-runtime", configuration =
"shadow"))
+
compileOnly("org.apache.iceberg:iceberg-flink-runtime-$flinkMajorVersion:$icebergVersion")
+
compileOnly("org.apache.flink:flink-connector-hive_$scalaVersion:$flinkVersion")
+ compileOnly("org.apache.flink:flink-table-common:$flinkVersion")
+ compileOnly("org.apache.flink:flink-table-api-java:$flinkVersion")
+
compileOnly("org.apache.paimon:paimon-flink-$flinkMajorVersion:$paimonVersion")
+ compileOnly(libs.flinkjdbc119)
+ compileOnly(libs.hive2.common) {
+ exclude("org.eclipse.jetty.aggregate", "jetty-all")
+ exclude("org.eclipse.jetty.orbit", "javax.servlet")
+ }
+
+ testImplementation(project(":api"))
+ testImplementation(project(":catalogs:catalog-jdbc-common")) {
+ exclude("org.apache.logging.log4j")
+ }
+ testImplementation(project(":clients:client-java"))
+ testImplementation(project(":common"))
+ testImplementation(project(":core"))
+ testImplementation(project(":flink-connector:flink-common", "testArtifacts"))
+ testImplementation(project(":integration-test-common", "testArtifacts"))
+ testImplementation(project(":server"))
+ testImplementation(project(":server-common"))
+ testImplementation(libs.awaitility)
+ testImplementation(libs.flinkjdbc119)
+ testImplementation(libs.junit.jupiter.api)
+ testImplementation(libs.junit.jupiter.params)
+ testImplementation(libs.metrics.core)
+ testImplementation(libs.minikdc)
+ testImplementation(libs.mockito.core)
+ testImplementation(libs.mysql.driver)
+ testImplementation(libs.postgresql.driver)
+ testImplementation(libs.sqlite.jdbc)
+ testImplementation(libs.testcontainers)
+ testImplementation(libs.testcontainers.junit.jupiter)
+ testImplementation(libs.testcontainers.mysql)
+
+ testImplementation("org.apache.iceberg:iceberg-core:$icebergVersion")
+
testImplementation("org.apache.iceberg:iceberg-hive-metastore:$icebergVersion")
+
testImplementation("org.apache.iceberg:iceberg-flink-runtime-$flinkMajorVersion:$icebergVersion")
+
testImplementation("org.apache.flink:flink-connector-hive_$scalaVersion:$flinkVersion")
+ testImplementation("org.apache.flink:flink-table-common:$flinkVersion")
+ testImplementation("org.apache.flink:flink-table-api-java:$flinkVersion")
+ testImplementation("org.apache.flink:flink-sql-gateway:$flinkVersion")
+
testImplementation("org.apache.paimon:paimon-flink-$flinkMajorVersion:$paimonVersion")
+
+ testImplementation(libs.hive2.exec) {
+ artifact {
+ classifier = "core"
+ }
+ exclude("com.fasterxml.jackson.core")
+ exclude("com.google.code.findbugs", "jsr305")
+ exclude("com.google.protobuf")
+ exclude("org.apache.avro")
+ exclude("org.apache.calcite")
+ exclude("org.apache.calcite.avatica")
+ exclude("org.apache.curator")
+ exclude("org.apache.hadoop", "hadoop-yarn-server-resourcemanager")
+ exclude("org.apache.logging.log4j")
+ exclude("org.apache.zookeeper")
+ exclude("org.eclipse.jetty.aggregate", "jetty-all")
+ exclude("org.eclipse.jetty.orbit", "javax.servlet")
+ exclude("org.openjdk.jol")
+ exclude("org.pentaho")
+ exclude("org.slf4j")
+ }
+
+ testImplementation(libs.hadoop3.common) {
+ exclude("*")
+ }
+ testImplementation(libs.hadoop3.hdfs) {
+ exclude("com.sun.jersey")
+ exclude("commons-cli", "commons-cli")
+ exclude("commons-io", "commons-io")
+ exclude("commons-codec", "commons-codec")
+ exclude("commons-logging", "commons-logging")
+ exclude("javax.servlet", "servlet-api")
+ exclude("org.mortbay.jetty")
+ }
+ testImplementation(libs.hadoop3.hdfs.client)
+ testImplementation(libs.hadoop3.mapreduce.client.core) {
+ exclude("*")
+ }
+ // Hadoop 3.x runtime requirements (stripped by exclude("*") above)
+ testImplementation(libs.hadoop3.shaded.guava)
+ testImplementation(libs.hadoop3.shaded.protobuf)
+ testImplementation(libs.commons.configuration2)
+ testImplementation(libs.re2j)
+ testImplementation(libs.hive2.common) {
+ exclude("org.eclipse.jetty.aggregate", "jetty-all")
+ exclude("org.eclipse.jetty.orbit", "javax.servlet")
+ }
+ testImplementation(libs.hive2.metastore) {
+ exclude("co.cask.tephra")
+ exclude("com.github.joshelser")
+ exclude("com.google.code.findbugs", "jsr305")
+ exclude("com.google.code.findbugs", "sr305")
+ exclude("com.tdunning", "json")
+ exclude("com.zaxxer", "HikariCP")
+ exclude("io.dropwizard.metrics")
+ exclude("javax.transaction", "transaction-api")
+ exclude("org.apache.avro")
+ exclude("org.apache.curator")
+ exclude("org.apache.hbase")
+ exclude("org.apache.hadoop", "hadoop-yarn-server-resourcemanager")
+ exclude("org.apache.logging.log4j")
+ exclude("org.apache.parquet", "parquet-hadoop-bundle")
+ exclude("org.apache.zookeeper")
+ exclude("org.eclipse.jetty.aggregate", "jetty-all")
+ exclude("org.eclipse.jetty.orbit", "javax.servlet")
+ exclude("org.slf4j")
+ }
+
testImplementation("org.apache.flink:flink-table-api-bridge-base:$flinkVersion")
{
+ exclude("commons-cli", "commons-cli")
+ exclude("commons-io", "commons-io")
+ exclude("com.google.code.findbugs", "jsr305")
+ }
+
testImplementation("org.apache.flink:flink-table-planner_$scalaVersion:$flinkVersion")
+ testImplementation("org.apache.flink:flink-test-utils:$flinkVersion")
+
+ testRuntimeOnly(libs.junit.jupiter.engine)
+}
+
+tasks.test {
+ dependsOn(commonProject.tasks.named("testClasses"))
+ // A test artifact dependency only adds common test classes to the
classpath. Add the common
+ // test output explicitly so Gradle discovers and executes those shared
tests in this variant.
+ testClassesDirs = files(commonTestOutput.classesDirs,
sourceSets["test"].output.classesDirs)
+ classpath = files(commonTestOutput, sourceSets["test"].runtimeClasspath)
+
+ val skipITs = project.hasProperty("skipITs")
+ if (skipITs) {
+ exclude("**/integration/test/**")
+ } else {
+ dependsOn(tasks.jar)
+ dependsOn(":catalogs:catalog-hive:jar")
+ dependsOn(":catalogs:catalog-hive:runtimeJars")
+ dependsOn(":catalogs:catalog-lakehouse-iceberg:jar")
+ dependsOn(":catalogs:catalog-lakehouse-iceberg:runtimeJars")
+ dependsOn(":iceberg:iceberg-rest-server:jar")
+ dependsOn(":catalogs:catalog-lakehouse-paimon:jar")
+ dependsOn(":catalogs:catalog-lakehouse-paimon:runtimeJars")
+ dependsOn(":catalogs:catalog-jdbc-mysql:jar")
+ dependsOn(":catalogs:catalog-jdbc-mysql:runtimeJars")
+ dependsOn(":catalogs:catalog-jdbc-postgresql:jar")
+ dependsOn(":catalogs:catalog-jdbc-postgresql:runtimeJars")
+ }
+}
+
+tasks.withType<Jar> {
+ archiveBaseName.set(artifactName)
+}
+
+publishing {
+ publications {
+ withType<MavenPublication>().configureEach {
+ artifactId = artifactName
+ }
+ }
+}
+
+tasks.named<Jar>("sourcesJar") {
+ duplicatesStrategy = DuplicatesStrategy.EXCLUDE
+}
diff --git
a/flink-connector/v1.19/flink/src/main/java/org/apache/gravitino/flink/connector/hive/GravitinoHiveCatalogFactoryFlink119.java
b/flink-connector/v1.19/flink/src/main/java/org/apache/gravitino/flink/connector/hive/GravitinoHiveCatalogFactoryFlink119.java
new file mode 100644
index 0000000000..c427e113b2
--- /dev/null
+++
b/flink-connector/v1.19/flink/src/main/java/org/apache/gravitino/flink/connector/hive/GravitinoHiveCatalogFactoryFlink119.java
@@ -0,0 +1,50 @@
+/*
+ * 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.hive;
+
+import java.util.Map;
+import javax.annotation.Nullable;
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.gravitino.flink.connector.PartitionConverter;
+import org.apache.gravitino.flink.connector.SchemaAndTablePropertiesConverter;
+import org.apache.hadoop.hive.conf.HiveConf;
+
+/** {@link GravitinoHiveCatalogFactory} implementation for Flink 1.19. */
+public class GravitinoHiveCatalogFactoryFlink119 extends
GravitinoHiveCatalogFactory {
+
+ @Override
+ protected Catalog newCatalog(
+ String catalogName,
+ String defaultDatabase,
+ Map<String, String> catalogOptions,
+ SchemaAndTablePropertiesConverter schemaAndTablePropertiesConverter,
+ PartitionConverter partitionConverter,
+ @Nullable HiveConf hiveConf,
+ @Nullable String hiveVersion) {
+ return new GravitinoHiveCatalogFlink119(
+ catalogName,
+ defaultDatabase,
+ catalogOptions,
+ schemaAndTablePropertiesConverter,
+ partitionConverter,
+ hiveConf,
+ hiveVersion);
+ }
+}
diff --git
a/flink-connector/v1.19/flink/src/main/java/org/apache/gravitino/flink/connector/hive/GravitinoHiveCatalogFlink119.java
b/flink-connector/v1.19/flink/src/main/java/org/apache/gravitino/flink/connector/hive/GravitinoHiveCatalogFlink119.java
new file mode 100644
index 0000000000..b133df9394
--- /dev/null
+++
b/flink-connector/v1.19/flink/src/main/java/org/apache/gravitino/flink/connector/hive/GravitinoHiveCatalogFlink119.java
@@ -0,0 +1,55 @@
+/*
+ * 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.hive;
+
+import java.util.Map;
+import javax.annotation.Nullable;
+import org.apache.gravitino.flink.connector.PartitionConverter;
+import org.apache.gravitino.flink.connector.SchemaAndTablePropertiesConverter;
+import org.apache.gravitino.flink.connector.utils.CatalogCompat;
+import org.apache.gravitino.flink.connector.utils.DefaultCatalogCompat;
+import org.apache.hadoop.hive.conf.HiveConf;
+
+/** {@link GravitinoHiveCatalog} implementation for Flink 1.19. */
+public class GravitinoHiveCatalogFlink119 extends GravitinoHiveCatalog {
+
+ GravitinoHiveCatalogFlink119(
+ String catalogName,
+ String defaultDatabase,
+ Map<String, String> catalogOptions,
+ SchemaAndTablePropertiesConverter schemaAndTablePropertiesConverter,
+ PartitionConverter partitionConverter,
+ @Nullable HiveConf hiveConf,
+ @Nullable String hiveVersion) {
+ super(
+ catalogName,
+ defaultDatabase,
+ catalogOptions,
+ schemaAndTablePropertiesConverter,
+ partitionConverter,
+ hiveConf,
+ hiveVersion);
+ }
+
+ @Override
+ protected CatalogCompat catalogCompat() {
+ return DefaultCatalogCompat.INSTANCE;
+ }
+}
diff --git
a/flink-connector/v1.19/flink/src/main/java/org/apache/gravitino/flink/connector/iceberg/GravitinoIcebergCatalogFactoryFlink119.java
b/flink-connector/v1.19/flink/src/main/java/org/apache/gravitino/flink/connector/iceberg/GravitinoIcebergCatalogFactoryFlink119.java
new file mode 100644
index 0000000000..61ca21b8c0
--- /dev/null
+++
b/flink-connector/v1.19/flink/src/main/java/org/apache/gravitino/flink/connector/iceberg/GravitinoIcebergCatalogFactoryFlink119.java
@@ -0,0 +1,46 @@
+/*
+ * 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 java.util.Map;
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.gravitino.flink.connector.PartitionConverter;
+import org.apache.gravitino.flink.connector.SchemaAndTablePropertiesConverter;
+
+/** {@link GravitinoIcebergCatalogFactory} implementation for Flink 1.19. */
+public class GravitinoIcebergCatalogFactoryFlink119 extends
GravitinoIcebergCatalogFactory {
+
+ @Override
+ protected Catalog newCatalog(
+ String catalogName,
+ String defaultDatabase,
+ SchemaAndTablePropertiesConverter schemaAndTablePropertiesConverter,
+ PartitionConverter partitionConverter,
+ Map<String, String> catalogOptions,
+ Map<String, String> icebergCatalogProperties) {
+ return new GravitinoIcebergCatalogFlink119(
+ catalogName,
+ defaultDatabase,
+ schemaAndTablePropertiesConverter,
+ partitionConverter,
+ catalogOptions,
+ icebergCatalogProperties);
+ }
+}
diff --git
a/flink-connector/v1.19/flink/src/main/java/org/apache/gravitino/flink/connector/iceberg/GravitinoIcebergCatalogFlink119.java
b/flink-connector/v1.19/flink/src/main/java/org/apache/gravitino/flink/connector/iceberg/GravitinoIcebergCatalogFlink119.java
new file mode 100644
index 0000000000..e4f56e3a43
--- /dev/null
+++
b/flink-connector/v1.19/flink/src/main/java/org/apache/gravitino/flink/connector/iceberg/GravitinoIcebergCatalogFlink119.java
@@ -0,0 +1,59 @@
+/*
+ * 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 java.util.Map;
+import org.apache.flink.table.catalog.AbstractCatalog;
+import org.apache.gravitino.flink.connector.PartitionConverter;
+import org.apache.gravitino.flink.connector.SchemaAndTablePropertiesConverter;
+import org.apache.gravitino.flink.connector.utils.CatalogCompat;
+import org.apache.gravitino.flink.connector.utils.DefaultCatalogCompat;
+import org.apache.iceberg.flink.FlinkCatalogFactory;
+
+/** {@link GravitinoIcebergCatalog} implementation for Flink 1.19. */
+public class GravitinoIcebergCatalogFlink119 extends GravitinoIcebergCatalog {
+
+ protected GravitinoIcebergCatalogFlink119(
+ String catalogName,
+ String defaultDatabase,
+ SchemaAndTablePropertiesConverter schemaAndTablePropertiesConverter,
+ PartitionConverter partitionConverter,
+ Map<String, String> catalogOptions,
+ Map<String, String> icebergCatalogProperties) {
+ super(
+ catalogName,
+ defaultDatabase,
+ schemaAndTablePropertiesConverter,
+ partitionConverter,
+ catalogOptions,
+ createIcebergCatalog(catalogName, icebergCatalogProperties));
+ }
+
+ @Override
+ protected CatalogCompat catalogCompat() {
+ return DefaultCatalogCompat.INSTANCE;
+ }
+
+ private static AbstractCatalog createIcebergCatalog(
+ String catalogName, Map<String, String> icebergCatalogProperties) {
+ return asAbstractCatalog(
+ new FlinkCatalogFactory().createCatalog(catalogName,
icebergCatalogProperties));
+ }
+}
diff --git
a/flink-connector/v1.19/flink/src/main/java/org/apache/gravitino/flink/connector/jdbc/GravitinoJdbcCatalogFlink119.java
b/flink-connector/v1.19/flink/src/main/java/org/apache/gravitino/flink/connector/jdbc/GravitinoJdbcCatalogFlink119.java
new file mode 100644
index 0000000000..d65ea32505
--- /dev/null
+++
b/flink-connector/v1.19/flink/src/main/java/org/apache/gravitino/flink/connector/jdbc/GravitinoJdbcCatalogFlink119.java
@@ -0,0 +1,62 @@
+/*
+ * 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.jdbc;
+
+import java.util.Optional;
+import
org.apache.flink.connector.jdbc.core.database.catalog.factory.JdbcCatalogFactory;
+import org.apache.flink.connector.jdbc.core.table.JdbcDynamicTableFactory;
+import org.apache.flink.table.catalog.AbstractCatalog;
+import org.apache.flink.table.factories.CatalogFactory;
+import org.apache.flink.table.factories.Factory;
+import org.apache.gravitino.flink.connector.PartitionConverter;
+import org.apache.gravitino.flink.connector.SchemaAndTablePropertiesConverter;
+import org.apache.gravitino.flink.connector.utils.CatalogCompat;
+import org.apache.gravitino.flink.connector.utils.DefaultCatalogCompat;
+
+/**
+ * {@link GravitinoJdbcCatalog} implementation for Flink 1.19. Uses the
relocated {@code
+ * org.apache.flink.connector.jdbc.core.*} factory packages shipped with
{@code flink-connector-jdbc
+ * 3.3.x}.
+ */
+public class GravitinoJdbcCatalogFlink119 extends GravitinoJdbcCatalog {
+
+ public GravitinoJdbcCatalogFlink119(
+ CatalogFactory.Context context,
+ String defaultDatabase,
+ SchemaAndTablePropertiesConverter schemaAndTablePropertiesConverter,
+ PartitionConverter partitionConverter) {
+ super(
+ context,
+ defaultDatabase,
+ schemaAndTablePropertiesConverter,
+ partitionConverter,
+ (AbstractCatalog) new JdbcCatalogFactory().createCatalog(context));
+ }
+
+ @Override
+ public Optional<Factory> getFactory() {
+ return Optional.of(new JdbcDynamicTableFactory());
+ }
+
+ @Override
+ protected CatalogCompat catalogCompat() {
+ return DefaultCatalogCompat.INSTANCE;
+ }
+}
diff --git
a/flink-connector/v1.19/flink/src/main/java/org/apache/gravitino/flink/connector/jdbc/mysql/GravitinoMysqlJdbcCatalogFactoryFlink119.java
b/flink-connector/v1.19/flink/src/main/java/org/apache/gravitino/flink/connector/jdbc/mysql/GravitinoMysqlJdbcCatalogFactoryFlink119.java
new file mode 100644
index 0000000000..91bf581434
--- /dev/null
+++
b/flink-connector/v1.19/flink/src/main/java/org/apache/gravitino/flink/connector/jdbc/mysql/GravitinoMysqlJdbcCatalogFactoryFlink119.java
@@ -0,0 +1,40 @@
+/*
+ * 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.jdbc.mysql;
+
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.factories.CatalogFactory;
+import org.apache.gravitino.flink.connector.PartitionConverter;
+import org.apache.gravitino.flink.connector.SchemaAndTablePropertiesConverter;
+import org.apache.gravitino.flink.connector.jdbc.GravitinoJdbcCatalogFlink119;
+
+/** {@link GravitinoMysqlJdbcCatalogFactory} implementation for Flink 1.19. */
+public class GravitinoMysqlJdbcCatalogFactoryFlink119 extends
GravitinoMysqlJdbcCatalogFactory {
+
+ @Override
+ protected Catalog newCatalog(
+ CatalogFactory.Context context,
+ String defaultDatabase,
+ SchemaAndTablePropertiesConverter schemaAndTablePropertiesConverter,
+ PartitionConverter partitionConverter) {
+ return new GravitinoJdbcCatalogFlink119(
+ context, defaultDatabase, schemaAndTablePropertiesConverter,
partitionConverter);
+ }
+}
diff --git
a/flink-connector/v1.19/flink/src/main/java/org/apache/gravitino/flink/connector/jdbc/postgresql/GravitinoPostgresJdbcCatalogFactoryFlink119.java
b/flink-connector/v1.19/flink/src/main/java/org/apache/gravitino/flink/connector/jdbc/postgresql/GravitinoPostgresJdbcCatalogFactoryFlink119.java
new file mode 100644
index 0000000000..c8ce2def9c
--- /dev/null
+++
b/flink-connector/v1.19/flink/src/main/java/org/apache/gravitino/flink/connector/jdbc/postgresql/GravitinoPostgresJdbcCatalogFactoryFlink119.java
@@ -0,0 +1,41 @@
+/*
+ * 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.jdbc.postgresql;
+
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.factories.CatalogFactory;
+import org.apache.gravitino.flink.connector.PartitionConverter;
+import org.apache.gravitino.flink.connector.SchemaAndTablePropertiesConverter;
+import org.apache.gravitino.flink.connector.jdbc.GravitinoJdbcCatalogFlink119;
+
+/** {@link GravitinoPostgresJdbcCatalogFactory} implementation for Flink 1.19.
*/
+public class GravitinoPostgresJdbcCatalogFactoryFlink119
+ extends GravitinoPostgresJdbcCatalogFactory {
+
+ @Override
+ protected Catalog newCatalog(
+ CatalogFactory.Context context,
+ String defaultDatabase,
+ SchemaAndTablePropertiesConverter schemaAndTablePropertiesConverter,
+ PartitionConverter partitionConverter) {
+ return new GravitinoJdbcCatalogFlink119(
+ context, defaultDatabase, schemaAndTablePropertiesConverter,
partitionConverter);
+ }
+}
diff --git
a/flink-connector/v1.19/flink/src/main/java/org/apache/gravitino/flink/connector/paimon/GravitinoPaimonCatalogFactoryFlink119.java
b/flink-connector/v1.19/flink/src/main/java/org/apache/gravitino/flink/connector/paimon/GravitinoPaimonCatalogFactoryFlink119.java
new file mode 100644
index 0000000000..6b9026c334
--- /dev/null
+++
b/flink-connector/v1.19/flink/src/main/java/org/apache/gravitino/flink/connector/paimon/GravitinoPaimonCatalogFactoryFlink119.java
@@ -0,0 +1,38 @@
+/*
+ * 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.paimon;
+
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.gravitino.flink.connector.PartitionConverter;
+import org.apache.gravitino.flink.connector.SchemaAndTablePropertiesConverter;
+
+/** {@link GravitinoPaimonCatalogFactory} implementation for Flink 1.19. */
+public class GravitinoPaimonCatalogFactoryFlink119 extends
GravitinoPaimonCatalogFactory {
+
+ @Override
+ protected Catalog newCatalog(
+ Context context,
+ String defaultDatabase,
+ SchemaAndTablePropertiesConverter schemaAndTablePropertiesConverter,
+ PartitionConverter partitionConverter) {
+ return new GravitinoPaimonCatalogFlink119(
+ context, defaultDatabase, schemaAndTablePropertiesConverter,
partitionConverter);
+ }
+}
diff --git
a/flink-connector/v1.19/flink/src/main/java/org/apache/gravitino/flink/connector/paimon/GravitinoPaimonCatalogFlink119.java
b/flink-connector/v1.19/flink/src/main/java/org/apache/gravitino/flink/connector/paimon/GravitinoPaimonCatalogFlink119.java
new file mode 100644
index 0000000000..2c5adf8b48
--- /dev/null
+++
b/flink-connector/v1.19/flink/src/main/java/org/apache/gravitino/flink/connector/paimon/GravitinoPaimonCatalogFlink119.java
@@ -0,0 +1,43 @@
+/*
+ * 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.paimon;
+
+import org.apache.flink.table.factories.CatalogFactory;
+import org.apache.gravitino.flink.connector.PartitionConverter;
+import org.apache.gravitino.flink.connector.SchemaAndTablePropertiesConverter;
+import org.apache.gravitino.flink.connector.utils.CatalogCompat;
+import org.apache.gravitino.flink.connector.utils.DefaultCatalogCompat;
+
+/** {@link GravitinoPaimonCatalog} implementation for Flink 1.19. */
+public class GravitinoPaimonCatalogFlink119 extends GravitinoPaimonCatalog {
+
+ protected GravitinoPaimonCatalogFlink119(
+ CatalogFactory.Context context,
+ String defaultDatabase,
+ SchemaAndTablePropertiesConverter schemaAndTablePropertiesConverter,
+ PartitionConverter partitionConverter) {
+ super(context, defaultDatabase, schemaAndTablePropertiesConverter,
partitionConverter);
+ }
+
+ @Override
+ protected CatalogCompat catalogCompat() {
+ return DefaultCatalogCompat.INSTANCE;
+ }
+}
diff --git
a/flink-connector/v1.19/flink/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory
b/flink-connector/v1.19/flink/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory
new file mode 100644
index 0000000000..3d3491ed8f
--- /dev/null
+++
b/flink-connector/v1.19/flink/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory
@@ -0,0 +1,24 @@
+#
+# 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.
+#
+
+org.apache.gravitino.flink.connector.hive.GravitinoHiveCatalogFactoryFlink119
+org.apache.gravitino.flink.connector.paimon.GravitinoPaimonCatalogFactoryFlink119
+org.apache.gravitino.flink.connector.iceberg.GravitinoIcebergCatalogFactoryFlink119
+org.apache.gravitino.flink.connector.jdbc.mysql.GravitinoMysqlJdbcCatalogFactoryFlink119
+org.apache.gravitino.flink.connector.jdbc.postgresql.GravitinoPostgresJdbcCatalogFactoryFlink119
diff --git
a/flink-connector/v1.19/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/catalog/GravitinoCatalogManagerIT119.java
b/flink-connector/v1.19/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/catalog/GravitinoCatalogManagerIT119.java
new file mode 100644
index 0000000000..4b14165f2e
--- /dev/null
+++
b/flink-connector/v1.19/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/catalog/GravitinoCatalogManagerIT119.java
@@ -0,0 +1,22 @@
+/*
+ * 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.catalog;
+
+public class GravitinoCatalogManagerIT119 extends GravitinoCatalogManagerIT {}
diff --git
a/flink-connector/v1.19/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/hive/FlinkHiveCatalogIT119.java
b/flink-connector/v1.19/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/hive/FlinkHiveCatalogIT119.java
new file mode 100644
index 0000000000..8a69694354
--- /dev/null
+++
b/flink-connector/v1.19/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/hive/FlinkHiveCatalogIT119.java
@@ -0,0 +1,22 @@
+/*
+ * 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.hive;
+
+public class FlinkHiveCatalogIT119 extends FlinkHiveCatalogIT {}
diff --git
a/flink-connector/v1.19/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/hive/FlinkHiveKerberosClientIT119.java
b/flink-connector/v1.19/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/hive/FlinkHiveKerberosClientIT119.java
new file mode 100644
index 0000000000..c43f16984c
--- /dev/null
+++
b/flink-connector/v1.19/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/hive/FlinkHiveKerberosClientIT119.java
@@ -0,0 +1,22 @@
+/*
+ * 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.hive;
+
+public class FlinkHiveKerberosClientIT119 extends FlinkHiveKerberosClientIT {}
diff --git
a/flink-connector/v1.19/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/iceberg/FlinkIcebergHiveCatalogIT119.java
b/flink-connector/v1.19/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/iceberg/FlinkIcebergHiveCatalogIT119.java
new file mode 100644
index 0000000000..9a744e02e8
--- /dev/null
+++
b/flink-connector/v1.19/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/iceberg/FlinkIcebergHiveCatalogIT119.java
@@ -0,0 +1,22 @@
+/*
+ * 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;
+
+public class FlinkIcebergHiveCatalogIT119 extends FlinkIcebergHiveCatalogIT {}
diff --git
a/flink-connector/v1.19/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/iceberg/FlinkIcebergRestCatalogIT119.java
b/flink-connector/v1.19/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/iceberg/FlinkIcebergRestCatalogIT119.java
new file mode 100644
index 0000000000..b84263ea0f
--- /dev/null
+++
b/flink-connector/v1.19/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/iceberg/FlinkIcebergRestCatalogIT119.java
@@ -0,0 +1,22 @@
+/*
+ * 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;
+
+public class FlinkIcebergRestCatalogIT119 extends FlinkIcebergRestCatalogIT {}
diff --git
a/flink-connector/v1.19/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/jdbc/FlinkJdbcMysqlCatalogIT119.java
b/flink-connector/v1.19/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/jdbc/FlinkJdbcMysqlCatalogIT119.java
new file mode 100644
index 0000000000..5d44479f7d
--- /dev/null
+++
b/flink-connector/v1.19/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/jdbc/FlinkJdbcMysqlCatalogIT119.java
@@ -0,0 +1,22 @@
+/*
+ * 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.jdbc;
+
+public class FlinkJdbcMysqlCatalogIT119 extends FlinkJdbcMysqlCatalogIT {}
diff --git
a/flink-connector/v1.19/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/paimon/FlinkPaimonHiveBackendIT119.java
b/flink-connector/v1.19/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/paimon/FlinkPaimonHiveBackendIT119.java
new file mode 100644
index 0000000000..6bf1b87abd
--- /dev/null
+++
b/flink-connector/v1.19/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/paimon/FlinkPaimonHiveBackendIT119.java
@@ -0,0 +1,22 @@
+/*
+ * 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.paimon;
+
+public class FlinkPaimonHiveBackendIT119 extends FlinkPaimonHiveBackendIT {}
diff --git
a/flink-connector/v1.19/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/paimon/FlinkPaimonJdbcBackendIT119.java
b/flink-connector/v1.19/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/paimon/FlinkPaimonJdbcBackendIT119.java
new file mode 100644
index 0000000000..d09d3d466c
--- /dev/null
+++
b/flink-connector/v1.19/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/paimon/FlinkPaimonJdbcBackendIT119.java
@@ -0,0 +1,22 @@
+/*
+ * 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.paimon;
+
+public class FlinkPaimonJdbcBackendIT119 extends FlinkPaimonJdbcBackendIT {}
diff --git
a/flink-connector/v1.19/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/paimon/FlinkPaimonLocalFileSystemBackendIT119.java
b/flink-connector/v1.19/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/paimon/FlinkPaimonLocalFileSystemBackendIT119.java
new file mode 100644
index 0000000000..bec18c7e62
--- /dev/null
+++
b/flink-connector/v1.19/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/paimon/FlinkPaimonLocalFileSystemBackendIT119.java
@@ -0,0 +1,22 @@
+/*
+ * 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.paimon;
+
+public class FlinkPaimonLocalFileSystemBackendIT119 extends
FlinkPaimonLocalFileSystemBackendIT {}
diff --git a/flink-connector/v1.20/flink-runtime/build.gradle.kts
b/flink-connector/v1.20/flink-runtime/build.gradle.kts
new file mode 100644
index 0000000000..da3dd17670
--- /dev/null
+++ b/flink-connector/v1.20/flink-runtime/build.gradle.kts
@@ -0,0 +1,90 @@
+/*
+ * 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.
+ */
+import com.github.jengelman.gradle.plugins.shadow.tasks.ShadowJar
+
+plugins {
+ `maven-publish`
+ id("java")
+ id("idea")
+ alias(libs.plugins.shadow)
+}
+
+repositories {
+ mavenCentral()
+}
+
+val flinkVersion: String = libs.versions.flink120.get()
+val flinkMajorVersion: String = flinkVersion.substringBeforeLast(".")
+val scalaVersion: String = "2.12"
+val artifactName = "gravitino-${project.name}_$scalaVersion"
+val baseName =
"${rootProject.name}-flink-connector-runtime-${flinkMajorVersion}_$scalaVersion"
+
+configurations.all {
+ resolutionStrategy.eachDependency {
+ if (requested.group == "org.apache.logging.log4j") {
+ throw GradleException("Dependency 'org.apache.logging.log4j' is not
allowed.")
+ }
+ }
+}
+
+dependencies {
+ implementation(project(":clients:client-java-runtime", configuration =
"shadow"))
+ implementation(project(":flink-connector:flink-1.20"))
+
+ testImplementation(libs.junit.jupiter.api)
+ testRuntimeOnly(libs.junit.jupiter.engine)
+}
+
+val shadowJarTask = tasks.named<ShadowJar>("shadowJar")
+
+shadowJarTask.configure {
+ isZip64 = true
+ configurations = listOf(project.configurations.runtimeClasspath.get())
+ archiveFileName.set("$baseName-$version.jar")
+ archiveClassifier.set("")
+ mergeServiceFiles()
+
+ exclude("org/slf4j/**")
+ exclude("META-INF/maven/org.slf4j/**")
+
+ relocate("com.google", "org.apache.gravitino.shaded.com.google")
+ relocate("google", "org.apache.gravitino.shaded.google")
+ relocate("org.apache.hc", "org.apache.gravitino.shaded.org.apache.hc")
+}
+
+tasks.test {
+ useJUnitPlatform()
+ dependsOn(shadowJarTask)
+ doFirst {
+ systemProperty("shadowJarPath",
shadowJarTask.get().archiveFile.get().asFile.absolutePath)
+ }
+}
+
+publishing {
+ publications {
+ withType<MavenPublication>().configureEach {
+ artifactId = baseName
+ }
+ }
+}
+
+tasks.jar {
+ dependsOn(tasks.named("shadowJar"))
+ archiveClassifier.set("empty")
+}
diff --git
a/flink-connector/v1.20/flink-runtime/src/test/java/org/apache/gravitino/flink/runtime/TestRuntimeJarDependencies.java
b/flink-connector/v1.20/flink-runtime/src/test/java/org/apache/gravitino/flink/runtime/TestRuntimeJarDependencies.java
new file mode 100644
index 0000000000..4522b2458f
--- /dev/null
+++
b/flink-connector/v1.20/flink-runtime/src/test/java/org/apache/gravitino/flink/runtime/TestRuntimeJarDependencies.java
@@ -0,0 +1,80 @@
+/*
+ * 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.runtime;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.jar.JarEntry;
+import java.util.jar.JarFile;
+import java.util.stream.Collectors;
+import org.junit.jupiter.api.Test;
+
+class TestRuntimeJarDependencies {
+ private static final String FACTORY_SERVICE_ENTRY =
+ "META-INF/services/org.apache.flink.table.factories.Factory";
+
+ @Test
+ void shadowJarShouldNotBundleSlf4j() throws IOException {
+ String jarPath = System.getProperty("shadowJarPath");
+ assertNotNull(jarPath, "shadowJarPath system property should be provided
by the build");
+
+ File runtimeJar = new File(jarPath);
+ assertTrue(runtimeJar.exists(), "shadow jar does not exist: " +
runtimeJar);
+
+ try (JarFile jarFile = new JarFile(runtimeJar)) {
+ List<String> entries =
jarFile.stream().map(JarEntry::getName).collect(Collectors.toList());
+ boolean hasSlf4jClasses = entries.stream().anyMatch(entry ->
entry.startsWith("org/slf4j/"));
+ boolean hasSlf4jMetadata =
+ entries.stream().anyMatch(entry ->
entry.startsWith("META-INF/maven/org.slf4j/"));
+ assertFalse(
+ hasSlf4jClasses,
+ "Flink connector runtime jar should rely on Flink provided slf4j
instead of shading it");
+ assertFalse(hasSlf4jMetadata, "SLF4J metadata should not be packaged in
runtime jar");
+ }
+ }
+
+ @Test
+ void shadowJarShouldMergeFactoryServiceDescriptors() throws IOException {
+ String jarPath = System.getProperty("shadowJarPath");
+ assertNotNull(jarPath, "shadowJarPath system property should be provided
by the build");
+
+ try (JarFile jarFile = new JarFile(new File(jarPath))) {
+ JarEntry factoryServiceEntry =
jarFile.getJarEntry(FACTORY_SERVICE_ENTRY);
+ assertNotNull(factoryServiceEntry, "Factory service descriptor should
exist in runtime jar");
+
+ String factoryServices =
+ new String(
+ jarFile.getInputStream(factoryServiceEntry).readAllBytes(),
StandardCharsets.UTF_8);
+ assertTrue(
+ factoryServices.contains(
+
"org.apache.gravitino.flink.connector.store.GravitinoCatalogStoreFactory"),
+ "Runtime jar should include the shared catalog-store factory
descriptor");
+ assertTrue(
+ factoryServices.contains(
+
"org.apache.gravitino.flink.connector.hive.GravitinoHiveCatalogFactoryFlink120"),
+ "Runtime jar should include the version-specific catalog factory
descriptor");
+ }
+ }
+}
diff --git a/flink-connector/v1.20/flink/build.gradle.kts
b/flink-connector/v1.20/flink/build.gradle.kts
new file mode 100644
index 0000000000..bbcd018e1c
--- /dev/null
+++ b/flink-connector/v1.20/flink/build.gradle.kts
@@ -0,0 +1,207 @@
+/*
+ * 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.
+ */
+import org.gradle.api.tasks.SourceSetContainer
+
+plugins {
+ `maven-publish`
+ id("java")
+ id("idea")
+}
+
+repositories {
+ mavenCentral()
+}
+
+val commonProject = project(":flink-connector:flink-common")
+val commonSourceSets = commonProject.extensions.getByType<SourceSetContainer>()
+val commonTestOutput = commonSourceSets.named("test").get().output
+val flinkVersion: String = libs.versions.flink120.get()
+val flinkMajorVersion: String = flinkVersion.substringBeforeLast(".")
+val icebergVersion: String = libs.versions.iceberg4flink120.get()
+val paimonVersion: String = libs.versions.paimon4flink120.get()
+val scalaVersion: String = "2.12"
+val artifactName =
"${rootProject.name}-flink-${flinkMajorVersion}_$scalaVersion"
+
+dependencies {
+ implementation(commonProject)
+
+ compileOnly(project(":clients:client-java-runtime", configuration =
"shadow"))
+
compileOnly("org.apache.iceberg:iceberg-flink-runtime-$flinkMajorVersion:$icebergVersion")
+
compileOnly("org.apache.flink:flink-connector-hive_$scalaVersion:$flinkVersion")
+ compileOnly("org.apache.flink:flink-table-common:$flinkVersion")
+ compileOnly("org.apache.flink:flink-table-api-java:$flinkVersion")
+
compileOnly("org.apache.paimon:paimon-flink-$flinkMajorVersion:$paimonVersion")
+ compileOnly(libs.flinkjdbc120)
+ compileOnly(libs.hive2.common) {
+ exclude("org.eclipse.jetty.aggregate", "jetty-all")
+ exclude("org.eclipse.jetty.orbit", "javax.servlet")
+ }
+
+ testImplementation(project(":api"))
+ testImplementation(project(":catalogs:catalog-jdbc-common")) {
+ exclude("org.apache.logging.log4j")
+ }
+ testImplementation(project(":clients:client-java"))
+ testImplementation(project(":common"))
+ testImplementation(project(":core"))
+ testImplementation(project(":flink-connector:flink-common", "testArtifacts"))
+ testImplementation(project(":integration-test-common", "testArtifacts"))
+ testImplementation(project(":server"))
+ testImplementation(project(":server-common"))
+ testImplementation(libs.awaitility)
+ testImplementation(libs.flinkjdbc120)
+ testImplementation(libs.junit.jupiter.api)
+ testImplementation(libs.junit.jupiter.params)
+ testImplementation(libs.metrics.core)
+ testImplementation(libs.minikdc)
+ testImplementation(libs.mockito.core)
+ testImplementation(libs.mysql.driver)
+ testImplementation(libs.postgresql.driver)
+ testImplementation(libs.sqlite.jdbc)
+ testImplementation(libs.testcontainers)
+ testImplementation(libs.testcontainers.junit.jupiter)
+ testImplementation(libs.testcontainers.mysql)
+
+ testImplementation("org.apache.iceberg:iceberg-core:$icebergVersion")
+
testImplementation("org.apache.iceberg:iceberg-hive-metastore:$icebergVersion")
+
testImplementation("org.apache.iceberg:iceberg-flink-runtime-$flinkMajorVersion:$icebergVersion")
+
testImplementation("org.apache.flink:flink-connector-hive_$scalaVersion:$flinkVersion")
+ testImplementation("org.apache.flink:flink-table-common:$flinkVersion")
+ testImplementation("org.apache.flink:flink-table-api-java:$flinkVersion")
+ testImplementation("org.apache.flink:flink-sql-gateway:$flinkVersion")
+
testImplementation("org.apache.paimon:paimon-flink-$flinkMajorVersion:$paimonVersion")
+
+ testImplementation(libs.hive2.exec) {
+ artifact {
+ classifier = "core"
+ }
+ exclude("com.fasterxml.jackson.core")
+ exclude("com.google.code.findbugs", "jsr305")
+ exclude("com.google.protobuf")
+ exclude("org.apache.avro")
+ exclude("org.apache.calcite")
+ exclude("org.apache.calcite.avatica")
+ exclude("org.apache.curator")
+ exclude("org.apache.hadoop", "hadoop-yarn-server-resourcemanager")
+ exclude("org.apache.logging.log4j")
+ exclude("org.apache.zookeeper")
+ exclude("org.eclipse.jetty.aggregate", "jetty-all")
+ exclude("org.eclipse.jetty.orbit", "javax.servlet")
+ exclude("org.openjdk.jol")
+ exclude("org.pentaho")
+ exclude("org.slf4j")
+ }
+
+ testImplementation(libs.hadoop3.common) {
+ exclude("*")
+ }
+ testImplementation(libs.hadoop3.hdfs) {
+ exclude("com.sun.jersey")
+ exclude("commons-cli", "commons-cli")
+ exclude("commons-io", "commons-io")
+ exclude("commons-codec", "commons-codec")
+ exclude("commons-logging", "commons-logging")
+ exclude("javax.servlet", "servlet-api")
+ exclude("org.mortbay.jetty")
+ }
+ testImplementation(libs.hadoop3.hdfs.client)
+ testImplementation(libs.hadoop3.mapreduce.client.core) {
+ exclude("*")
+ }
+ // Hadoop 3.x runtime requirements (stripped by exclude("*") above)
+ testImplementation(libs.hadoop3.shaded.guava)
+ testImplementation(libs.hadoop3.shaded.protobuf)
+ testImplementation(libs.commons.configuration2)
+ testImplementation(libs.re2j)
+ testImplementation(libs.hive2.common) {
+ exclude("org.eclipse.jetty.aggregate", "jetty-all")
+ exclude("org.eclipse.jetty.orbit", "javax.servlet")
+ }
+ testImplementation(libs.hive2.metastore) {
+ exclude("co.cask.tephra")
+ exclude("com.github.joshelser")
+ exclude("com.google.code.findbugs", "jsr305")
+ exclude("com.google.code.findbugs", "sr305")
+ exclude("com.tdunning", "json")
+ exclude("com.zaxxer", "HikariCP")
+ exclude("io.dropwizard.metrics")
+ exclude("javax.transaction", "transaction-api")
+ exclude("org.apache.avro")
+ exclude("org.apache.curator")
+ exclude("org.apache.hbase")
+ exclude("org.apache.hadoop", "hadoop-yarn-server-resourcemanager")
+ exclude("org.apache.logging.log4j")
+ exclude("org.apache.parquet", "parquet-hadoop-bundle")
+ exclude("org.apache.zookeeper")
+ exclude("org.eclipse.jetty.aggregate", "jetty-all")
+ exclude("org.eclipse.jetty.orbit", "javax.servlet")
+ exclude("org.slf4j")
+ }
+
testImplementation("org.apache.flink:flink-table-api-bridge-base:$flinkVersion")
{
+ exclude("commons-cli", "commons-cli")
+ exclude("commons-io", "commons-io")
+ exclude("com.google.code.findbugs", "jsr305")
+ }
+
testImplementation("org.apache.flink:flink-table-planner_$scalaVersion:$flinkVersion")
+ testImplementation("org.apache.flink:flink-test-utils:$flinkVersion")
+
+ testRuntimeOnly(libs.junit.jupiter.engine)
+}
+
+tasks.test {
+ dependsOn(commonProject.tasks.named("testClasses"))
+ // A test artifact dependency only adds common test classes to the
classpath. Add the common
+ // test output explicitly so Gradle discovers and executes those shared
tests in this variant.
+ testClassesDirs = files(commonTestOutput.classesDirs,
sourceSets["test"].output.classesDirs)
+ classpath = files(commonTestOutput, sourceSets["test"].runtimeClasspath)
+
+ val skipITs = project.hasProperty("skipITs")
+ if (skipITs) {
+ exclude("**/integration/test/**")
+ } else {
+ dependsOn(tasks.jar)
+ dependsOn(":catalogs:catalog-hive:jar")
+ dependsOn(":catalogs:catalog-hive:runtimeJars")
+ dependsOn(":catalogs:catalog-lakehouse-iceberg:jar")
+ dependsOn(":catalogs:catalog-lakehouse-iceberg:runtimeJars")
+ dependsOn(":iceberg:iceberg-rest-server:jar")
+ dependsOn(":catalogs:catalog-lakehouse-paimon:jar")
+ dependsOn(":catalogs:catalog-lakehouse-paimon:runtimeJars")
+ dependsOn(":catalogs:catalog-jdbc-mysql:jar")
+ dependsOn(":catalogs:catalog-jdbc-mysql:runtimeJars")
+ dependsOn(":catalogs:catalog-jdbc-postgresql:jar")
+ dependsOn(":catalogs:catalog-jdbc-postgresql:runtimeJars")
+ }
+}
+
+tasks.withType<Jar> {
+ archiveBaseName.set(artifactName)
+}
+
+publishing {
+ publications {
+ withType<MavenPublication>().configureEach {
+ artifactId = artifactName
+ }
+ }
+}
+
+tasks.named<Jar>("sourcesJar") {
+ duplicatesStrategy = DuplicatesStrategy.EXCLUDE
+}
diff --git
a/flink-connector/v1.20/flink/src/main/java/org/apache/gravitino/flink/connector/hive/GravitinoHiveCatalogFactoryFlink120.java
b/flink-connector/v1.20/flink/src/main/java/org/apache/gravitino/flink/connector/hive/GravitinoHiveCatalogFactoryFlink120.java
new file mode 100644
index 0000000000..5ed11f0e16
--- /dev/null
+++
b/flink-connector/v1.20/flink/src/main/java/org/apache/gravitino/flink/connector/hive/GravitinoHiveCatalogFactoryFlink120.java
@@ -0,0 +1,50 @@
+/*
+ * 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.hive;
+
+import java.util.Map;
+import javax.annotation.Nullable;
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.gravitino.flink.connector.PartitionConverter;
+import org.apache.gravitino.flink.connector.SchemaAndTablePropertiesConverter;
+import org.apache.hadoop.hive.conf.HiveConf;
+
+/** {@link GravitinoHiveCatalogFactory} implementation for Flink 1.20. */
+public class GravitinoHiveCatalogFactoryFlink120 extends
GravitinoHiveCatalogFactory {
+
+ @Override
+ protected Catalog newCatalog(
+ String catalogName,
+ String defaultDatabase,
+ Map<String, String> catalogOptions,
+ SchemaAndTablePropertiesConverter schemaAndTablePropertiesConverter,
+ PartitionConverter partitionConverter,
+ @Nullable HiveConf hiveConf,
+ @Nullable String hiveVersion) {
+ return new GravitinoHiveCatalogFlink120(
+ catalogName,
+ defaultDatabase,
+ catalogOptions,
+ schemaAndTablePropertiesConverter,
+ partitionConverter,
+ hiveConf,
+ hiveVersion);
+ }
+}
diff --git
a/flink-connector/v1.20/flink/src/main/java/org/apache/gravitino/flink/connector/hive/GravitinoHiveCatalogFlink120.java
b/flink-connector/v1.20/flink/src/main/java/org/apache/gravitino/flink/connector/hive/GravitinoHiveCatalogFlink120.java
new file mode 100644
index 0000000000..679e991bba
--- /dev/null
+++
b/flink-connector/v1.20/flink/src/main/java/org/apache/gravitino/flink/connector/hive/GravitinoHiveCatalogFlink120.java
@@ -0,0 +1,55 @@
+/*
+ * 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.hive;
+
+import java.util.Map;
+import javax.annotation.Nullable;
+import org.apache.gravitino.flink.connector.PartitionConverter;
+import org.apache.gravitino.flink.connector.SchemaAndTablePropertiesConverter;
+import org.apache.gravitino.flink.connector.utils.CatalogCompat;
+import org.apache.gravitino.flink.connector.utils.CatalogCompatFlink120;
+import org.apache.hadoop.hive.conf.HiveConf;
+
+/** {@link GravitinoHiveCatalog} implementation for Flink 1.20. */
+public class GravitinoHiveCatalogFlink120 extends GravitinoHiveCatalog {
+
+ GravitinoHiveCatalogFlink120(
+ String catalogName,
+ String defaultDatabase,
+ Map<String, String> catalogOptions,
+ SchemaAndTablePropertiesConverter schemaAndTablePropertiesConverter,
+ PartitionConverter partitionConverter,
+ @Nullable HiveConf hiveConf,
+ @Nullable String hiveVersion) {
+ super(
+ catalogName,
+ defaultDatabase,
+ catalogOptions,
+ schemaAndTablePropertiesConverter,
+ partitionConverter,
+ hiveConf,
+ hiveVersion);
+ }
+
+ @Override
+ protected CatalogCompat catalogCompat() {
+ return CatalogCompatFlink120.INSTANCE;
+ }
+}
diff --git
a/flink-connector/v1.20/flink/src/main/java/org/apache/gravitino/flink/connector/iceberg/GravitinoIcebergCatalogFactoryFlink120.java
b/flink-connector/v1.20/flink/src/main/java/org/apache/gravitino/flink/connector/iceberg/GravitinoIcebergCatalogFactoryFlink120.java
new file mode 100644
index 0000000000..d9f2e8b37f
--- /dev/null
+++
b/flink-connector/v1.20/flink/src/main/java/org/apache/gravitino/flink/connector/iceberg/GravitinoIcebergCatalogFactoryFlink120.java
@@ -0,0 +1,46 @@
+/*
+ * 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 java.util.Map;
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.gravitino.flink.connector.PartitionConverter;
+import org.apache.gravitino.flink.connector.SchemaAndTablePropertiesConverter;
+
+/** {@link GravitinoIcebergCatalogFactory} implementation for Flink 1.20. */
+public class GravitinoIcebergCatalogFactoryFlink120 extends
GravitinoIcebergCatalogFactory {
+
+ @Override
+ protected Catalog newCatalog(
+ String catalogName,
+ String defaultDatabase,
+ SchemaAndTablePropertiesConverter schemaAndTablePropertiesConverter,
+ PartitionConverter partitionConverter,
+ Map<String, String> catalogOptions,
+ Map<String, String> icebergCatalogProperties) {
+ return new GravitinoIcebergCatalogFlink120(
+ catalogName,
+ defaultDatabase,
+ schemaAndTablePropertiesConverter,
+ partitionConverter,
+ catalogOptions,
+ icebergCatalogProperties);
+ }
+}
diff --git
a/flink-connector/v1.20/flink/src/main/java/org/apache/gravitino/flink/connector/iceberg/GravitinoIcebergCatalogFlink120.java
b/flink-connector/v1.20/flink/src/main/java/org/apache/gravitino/flink/connector/iceberg/GravitinoIcebergCatalogFlink120.java
new file mode 100644
index 0000000000..f105b8b545
--- /dev/null
+++
b/flink-connector/v1.20/flink/src/main/java/org/apache/gravitino/flink/connector/iceberg/GravitinoIcebergCatalogFlink120.java
@@ -0,0 +1,59 @@
+/*
+ * 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 java.util.Map;
+import org.apache.flink.table.catalog.AbstractCatalog;
+import org.apache.gravitino.flink.connector.PartitionConverter;
+import org.apache.gravitino.flink.connector.SchemaAndTablePropertiesConverter;
+import org.apache.gravitino.flink.connector.utils.CatalogCompat;
+import org.apache.gravitino.flink.connector.utils.CatalogCompatFlink120;
+import org.apache.iceberg.flink.FlinkCatalogFactory;
+
+/** {@link GravitinoIcebergCatalog} implementation for Flink 1.20. */
+public class GravitinoIcebergCatalogFlink120 extends GravitinoIcebergCatalog {
+
+ protected GravitinoIcebergCatalogFlink120(
+ String catalogName,
+ String defaultDatabase,
+ SchemaAndTablePropertiesConverter schemaAndTablePropertiesConverter,
+ PartitionConverter partitionConverter,
+ Map<String, String> catalogOptions,
+ Map<String, String> icebergCatalogProperties) {
+ super(
+ catalogName,
+ defaultDatabase,
+ schemaAndTablePropertiesConverter,
+ partitionConverter,
+ catalogOptions,
+ createIcebergCatalog(catalogName, icebergCatalogProperties));
+ }
+
+ @Override
+ protected CatalogCompat catalogCompat() {
+ return CatalogCompatFlink120.INSTANCE;
+ }
+
+ private static AbstractCatalog createIcebergCatalog(
+ String catalogName, Map<String, String> icebergCatalogProperties) {
+ return asAbstractCatalog(
+ new FlinkCatalogFactory().createCatalog(catalogName,
icebergCatalogProperties));
+ }
+}
diff --git
a/flink-connector/v1.20/flink/src/main/java/org/apache/gravitino/flink/connector/jdbc/GravitinoJdbcCatalogFlink120.java
b/flink-connector/v1.20/flink/src/main/java/org/apache/gravitino/flink/connector/jdbc/GravitinoJdbcCatalogFlink120.java
new file mode 100644
index 0000000000..bb9b899462
--- /dev/null
+++
b/flink-connector/v1.20/flink/src/main/java/org/apache/gravitino/flink/connector/jdbc/GravitinoJdbcCatalogFlink120.java
@@ -0,0 +1,62 @@
+/*
+ * 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.jdbc;
+
+import java.util.Optional;
+import
org.apache.flink.connector.jdbc.core.database.catalog.factory.JdbcCatalogFactory;
+import org.apache.flink.connector.jdbc.core.table.JdbcDynamicTableFactory;
+import org.apache.flink.table.catalog.AbstractCatalog;
+import org.apache.flink.table.factories.CatalogFactory;
+import org.apache.flink.table.factories.Factory;
+import org.apache.gravitino.flink.connector.PartitionConverter;
+import org.apache.gravitino.flink.connector.SchemaAndTablePropertiesConverter;
+import org.apache.gravitino.flink.connector.utils.CatalogCompat;
+import org.apache.gravitino.flink.connector.utils.CatalogCompatFlink120;
+
+/**
+ * {@link GravitinoJdbcCatalog} implementation for Flink 1.20. Uses the
relocated {@code
+ * org.apache.flink.connector.jdbc.core.*} factory packages shipped with
{@code flink-connector-jdbc
+ * 3.3.x}.
+ */
+public class GravitinoJdbcCatalogFlink120 extends GravitinoJdbcCatalog {
+
+ public GravitinoJdbcCatalogFlink120(
+ CatalogFactory.Context context,
+ String defaultDatabase,
+ SchemaAndTablePropertiesConverter schemaAndTablePropertiesConverter,
+ PartitionConverter partitionConverter) {
+ super(
+ context,
+ defaultDatabase,
+ schemaAndTablePropertiesConverter,
+ partitionConverter,
+ (AbstractCatalog) new JdbcCatalogFactory().createCatalog(context));
+ }
+
+ @Override
+ public Optional<Factory> getFactory() {
+ return Optional.of(new JdbcDynamicTableFactory());
+ }
+
+ @Override
+ protected CatalogCompat catalogCompat() {
+ return CatalogCompatFlink120.INSTANCE;
+ }
+}
diff --git
a/flink-connector/v1.20/flink/src/main/java/org/apache/gravitino/flink/connector/jdbc/mysql/GravitinoMysqlJdbcCatalogFactoryFlink120.java
b/flink-connector/v1.20/flink/src/main/java/org/apache/gravitino/flink/connector/jdbc/mysql/GravitinoMysqlJdbcCatalogFactoryFlink120.java
new file mode 100644
index 0000000000..a0ff8eadc7
--- /dev/null
+++
b/flink-connector/v1.20/flink/src/main/java/org/apache/gravitino/flink/connector/jdbc/mysql/GravitinoMysqlJdbcCatalogFactoryFlink120.java
@@ -0,0 +1,40 @@
+/*
+ * 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.jdbc.mysql;
+
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.factories.CatalogFactory;
+import org.apache.gravitino.flink.connector.PartitionConverter;
+import org.apache.gravitino.flink.connector.SchemaAndTablePropertiesConverter;
+import org.apache.gravitino.flink.connector.jdbc.GravitinoJdbcCatalogFlink120;
+
+/** {@link GravitinoMysqlJdbcCatalogFactory} implementation for Flink 1.20. */
+public class GravitinoMysqlJdbcCatalogFactoryFlink120 extends
GravitinoMysqlJdbcCatalogFactory {
+
+ @Override
+ protected Catalog newCatalog(
+ CatalogFactory.Context context,
+ String defaultDatabase,
+ SchemaAndTablePropertiesConverter schemaAndTablePropertiesConverter,
+ PartitionConverter partitionConverter) {
+ return new GravitinoJdbcCatalogFlink120(
+ context, defaultDatabase, schemaAndTablePropertiesConverter,
partitionConverter);
+ }
+}
diff --git
a/flink-connector/v1.20/flink/src/main/java/org/apache/gravitino/flink/connector/jdbc/postgresql/GravitinoPostgresJdbcCatalogFactoryFlink120.java
b/flink-connector/v1.20/flink/src/main/java/org/apache/gravitino/flink/connector/jdbc/postgresql/GravitinoPostgresJdbcCatalogFactoryFlink120.java
new file mode 100644
index 0000000000..45637561a0
--- /dev/null
+++
b/flink-connector/v1.20/flink/src/main/java/org/apache/gravitino/flink/connector/jdbc/postgresql/GravitinoPostgresJdbcCatalogFactoryFlink120.java
@@ -0,0 +1,41 @@
+/*
+ * 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.jdbc.postgresql;
+
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.factories.CatalogFactory;
+import org.apache.gravitino.flink.connector.PartitionConverter;
+import org.apache.gravitino.flink.connector.SchemaAndTablePropertiesConverter;
+import org.apache.gravitino.flink.connector.jdbc.GravitinoJdbcCatalogFlink120;
+
+/** {@link GravitinoPostgresJdbcCatalogFactory} implementation for Flink 1.20.
*/
+public class GravitinoPostgresJdbcCatalogFactoryFlink120
+ extends GravitinoPostgresJdbcCatalogFactory {
+
+ @Override
+ protected Catalog newCatalog(
+ CatalogFactory.Context context,
+ String defaultDatabase,
+ SchemaAndTablePropertiesConverter schemaAndTablePropertiesConverter,
+ PartitionConverter partitionConverter) {
+ return new GravitinoJdbcCatalogFlink120(
+ context, defaultDatabase, schemaAndTablePropertiesConverter,
partitionConverter);
+ }
+}
diff --git
a/flink-connector/v1.20/flink/src/main/java/org/apache/gravitino/flink/connector/paimon/GravitinoPaimonCatalogFactoryFlink120.java
b/flink-connector/v1.20/flink/src/main/java/org/apache/gravitino/flink/connector/paimon/GravitinoPaimonCatalogFactoryFlink120.java
new file mode 100644
index 0000000000..de47221a5a
--- /dev/null
+++
b/flink-connector/v1.20/flink/src/main/java/org/apache/gravitino/flink/connector/paimon/GravitinoPaimonCatalogFactoryFlink120.java
@@ -0,0 +1,38 @@
+/*
+ * 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.paimon;
+
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.gravitino.flink.connector.PartitionConverter;
+import org.apache.gravitino.flink.connector.SchemaAndTablePropertiesConverter;
+
+/** {@link GravitinoPaimonCatalogFactory} implementation for Flink 1.20. */
+public class GravitinoPaimonCatalogFactoryFlink120 extends
GravitinoPaimonCatalogFactory {
+
+ @Override
+ protected Catalog newCatalog(
+ Context context,
+ String defaultDatabase,
+ SchemaAndTablePropertiesConverter schemaAndTablePropertiesConverter,
+ PartitionConverter partitionConverter) {
+ return new GravitinoPaimonCatalogFlink120(
+ context, defaultDatabase, schemaAndTablePropertiesConverter,
partitionConverter);
+ }
+}
diff --git
a/flink-connector/v1.20/flink/src/main/java/org/apache/gravitino/flink/connector/paimon/GravitinoPaimonCatalogFlink120.java
b/flink-connector/v1.20/flink/src/main/java/org/apache/gravitino/flink/connector/paimon/GravitinoPaimonCatalogFlink120.java
new file mode 100644
index 0000000000..464578849b
--- /dev/null
+++
b/flink-connector/v1.20/flink/src/main/java/org/apache/gravitino/flink/connector/paimon/GravitinoPaimonCatalogFlink120.java
@@ -0,0 +1,43 @@
+/*
+ * 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.paimon;
+
+import org.apache.flink.table.factories.CatalogFactory;
+import org.apache.gravitino.flink.connector.PartitionConverter;
+import org.apache.gravitino.flink.connector.SchemaAndTablePropertiesConverter;
+import org.apache.gravitino.flink.connector.utils.CatalogCompat;
+import org.apache.gravitino.flink.connector.utils.CatalogCompatFlink120;
+
+/** {@link GravitinoPaimonCatalog} implementation for Flink 1.20. */
+public class GravitinoPaimonCatalogFlink120 extends GravitinoPaimonCatalog {
+
+ protected GravitinoPaimonCatalogFlink120(
+ CatalogFactory.Context context,
+ String defaultDatabase,
+ SchemaAndTablePropertiesConverter schemaAndTablePropertiesConverter,
+ PartitionConverter partitionConverter) {
+ super(context, defaultDatabase, schemaAndTablePropertiesConverter,
partitionConverter);
+ }
+
+ @Override
+ protected CatalogCompat catalogCompat() {
+ return CatalogCompatFlink120.INSTANCE;
+ }
+}
diff --git
a/flink-connector/v1.20/flink/src/main/java/org/apache/gravitino/flink/connector/utils/CatalogCompatFlink120.java
b/flink-connector/v1.20/flink/src/main/java/org/apache/gravitino/flink/connector/utils/CatalogCompatFlink120.java
new file mode 100644
index 0000000000..11052f8150
--- /dev/null
+++
b/flink-connector/v1.20/flink/src/main/java/org/apache/gravitino/flink/connector/utils/CatalogCompatFlink120.java
@@ -0,0 +1,49 @@
+/*
+ * 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.utils;
+
+import java.util.List;
+import java.util.Map;
+import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.catalog.CatalogPropertiesUtil;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.ResolvedCatalogTable;
+
+/** {@link CatalogCompat} implementation for Flink 1.20. */
+public enum CatalogCompatFlink120 implements CatalogCompat {
+ INSTANCE;
+
+ @Override
+ public CatalogTable createCatalogTable(
+ Schema schema, String comment, List<String> partitionKeys, Map<String,
String> options) {
+ // CatalogTable.of(...) is deprecated in Flink 1.20; use the builder API
instead.
+ return CatalogTable.newBuilder()
+ .schema(schema)
+ .comment(comment)
+ .partitionKeys(partitionKeys)
+ .options(options)
+ .build();
+ }
+
+ @Override
+ public Map<String, String> serializeCatalogTable(ResolvedCatalogTable
resolvedTable) {
+ return CatalogPropertiesUtil.serializeCatalogTable(resolvedTable);
+ }
+}
diff --git
a/flink-connector/v1.20/flink/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory
b/flink-connector/v1.20/flink/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory
new file mode 100644
index 0000000000..d6b935d39c
--- /dev/null
+++
b/flink-connector/v1.20/flink/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory
@@ -0,0 +1,24 @@
+#
+# 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.
+#
+
+org.apache.gravitino.flink.connector.hive.GravitinoHiveCatalogFactoryFlink120
+org.apache.gravitino.flink.connector.paimon.GravitinoPaimonCatalogFactoryFlink120
+org.apache.gravitino.flink.connector.iceberg.GravitinoIcebergCatalogFactoryFlink120
+org.apache.gravitino.flink.connector.jdbc.mysql.GravitinoMysqlJdbcCatalogFactoryFlink120
+org.apache.gravitino.flink.connector.jdbc.postgresql.GravitinoPostgresJdbcCatalogFactoryFlink120
diff --git
a/flink-connector/v1.20/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/catalog/GravitinoCatalogManagerIT120.java
b/flink-connector/v1.20/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/catalog/GravitinoCatalogManagerIT120.java
new file mode 100644
index 0000000000..efc2c84251
--- /dev/null
+++
b/flink-connector/v1.20/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/catalog/GravitinoCatalogManagerIT120.java
@@ -0,0 +1,22 @@
+/*
+ * 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.catalog;
+
+public class GravitinoCatalogManagerIT120 extends GravitinoCatalogManagerIT {}
diff --git
a/flink-connector/v1.20/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/hive/FlinkHiveCatalogIT120.java
b/flink-connector/v1.20/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/hive/FlinkHiveCatalogIT120.java
new file mode 100644
index 0000000000..89b8f2d2d0
--- /dev/null
+++
b/flink-connector/v1.20/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/hive/FlinkHiveCatalogIT120.java
@@ -0,0 +1,78 @@
+/*
+ * 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.hive;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class FlinkHiveCatalogIT120 extends FlinkHiveCatalogIT {
+
+ @Override
+ @Test
+ public void testCreateGravitinoHiveCatalogRequireOptions() {
+ // Flink 1.20 reports missing required catalog options through
IllegalArgumentException
+ // rather than the ValidationException thrown by older minor versions.
+ tableEnv.useCatalog(DEFAULT_CATALOG);
+
+ String catalogName = "gravitino_hive_sql2";
+ IllegalArgumentException exception =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ tableEnv.executeSql(
+ String.format(
+ "create catalog %s with ("
+ + "'type'='gravitino-hive', "
+ + "'hive-conf-dir'='%s'"
+ + ")",
+ catalogName, getSharedHiveConfDir())));
+
+ Assertions.assertTrue(containsMessage(exception, "metastore.uris"),
collectMessages(exception));
+ Assertions.assertFalse(metalake.catalogExists(catalogName));
+ }
+
+ private static boolean containsMessage(Throwable throwable, String
expectedMessage) {
+ Throwable current = throwable;
+ while (current != null) {
+ String message = current.getMessage();
+ if (message != null && message.contains(expectedMessage)) {
+ return true;
+ }
+ current = current.getCause();
+ }
+
+ return false;
+ }
+
+ private static String collectMessages(Throwable throwable) {
+ StringBuilder messages = new StringBuilder();
+ Throwable current = throwable;
+ while (current != null) {
+ messages
+ .append(current.getClass().getName())
+ .append(": ")
+ .append(current.getMessage())
+ .append('\n');
+ current = current.getCause();
+ }
+
+ return messages.toString();
+ }
+}
diff --git
a/flink-connector/v1.20/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/hive/FlinkHiveKerberosClientIT120.java
b/flink-connector/v1.20/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/hive/FlinkHiveKerberosClientIT120.java
new file mode 100644
index 0000000000..662d836718
--- /dev/null
+++
b/flink-connector/v1.20/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/hive/FlinkHiveKerberosClientIT120.java
@@ -0,0 +1,22 @@
+/*
+ * 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.hive;
+
+public class FlinkHiveKerberosClientIT120 extends FlinkHiveKerberosClientIT {}
diff --git
a/flink-connector/v1.20/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/iceberg/FlinkIcebergHiveCatalogIT120.java
b/flink-connector/v1.20/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/iceberg/FlinkIcebergHiveCatalogIT120.java
new file mode 100644
index 0000000000..a0003c57fb
--- /dev/null
+++
b/flink-connector/v1.20/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/iceberg/FlinkIcebergHiveCatalogIT120.java
@@ -0,0 +1,22 @@
+/*
+ * 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;
+
+public class FlinkIcebergHiveCatalogIT120 extends FlinkIcebergHiveCatalogIT {}
diff --git
a/flink-connector/v1.20/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/iceberg/FlinkIcebergRestCatalogIT120.java
b/flink-connector/v1.20/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/iceberg/FlinkIcebergRestCatalogIT120.java
new file mode 100644
index 0000000000..1d272f11bd
--- /dev/null
+++
b/flink-connector/v1.20/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/iceberg/FlinkIcebergRestCatalogIT120.java
@@ -0,0 +1,22 @@
+/*
+ * 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;
+
+public class FlinkIcebergRestCatalogIT120 extends FlinkIcebergRestCatalogIT {}
diff --git
a/flink-connector/v1.20/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/jdbc/FlinkJdbcMysqlCatalogIT120.java
b/flink-connector/v1.20/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/jdbc/FlinkJdbcMysqlCatalogIT120.java
new file mode 100644
index 0000000000..ee2c77f436
--- /dev/null
+++
b/flink-connector/v1.20/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/jdbc/FlinkJdbcMysqlCatalogIT120.java
@@ -0,0 +1,51 @@
+/*
+ * 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.jdbc;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class FlinkJdbcMysqlCatalogIT120 extends FlinkJdbcMysqlCatalogIT {
+
+ @Override
+ @Test
+ public void testCreateGravitinoJdbcCatalogUsingSQLMissingOptions() {
+ // Flink 1.20 reports missing required catalog options through
IllegalArgumentException
+ // rather than the ValidationException thrown by older minor versions.
+ tableEnv.useCatalog(DEFAULT_CATALOG);
+ String catalogName = "gravitino_mysql_jdbc_catalog_missing_options";
+ IllegalArgumentException exception =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ tableEnv.executeSql(
+ String.format(
+ "create catalog %s with ("
+ + "'type'='gravitino-jdbc-mysql', "
+ + "'base-url'='%s',"
+ + "'username'='%s',"
+ + "'default-database'='%s'"
+ + ")",
+ catalogName, mysqlUrl, mysqlUsername,
mysqlDefaultDatabase)));
+
+ Assertions.assertTrue(exception.getMessage().contains("password"));
+ Assertions.assertFalse(metalake.catalogExists(catalogName));
+ }
+}
diff --git
a/flink-connector/v1.20/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/paimon/FlinkPaimonHiveBackendIT120.java
b/flink-connector/v1.20/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/paimon/FlinkPaimonHiveBackendIT120.java
new file mode 100644
index 0000000000..c3305a864f
--- /dev/null
+++
b/flink-connector/v1.20/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/paimon/FlinkPaimonHiveBackendIT120.java
@@ -0,0 +1,22 @@
+/*
+ * 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.paimon;
+
+public class FlinkPaimonHiveBackendIT120 extends FlinkPaimonHiveBackendIT {}
diff --git
a/flink-connector/v1.20/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/paimon/FlinkPaimonJdbcBackendIT120.java
b/flink-connector/v1.20/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/paimon/FlinkPaimonJdbcBackendIT120.java
new file mode 100644
index 0000000000..a718916e8e
--- /dev/null
+++
b/flink-connector/v1.20/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/paimon/FlinkPaimonJdbcBackendIT120.java
@@ -0,0 +1,22 @@
+/*
+ * 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.paimon;
+
+public class FlinkPaimonJdbcBackendIT120 extends FlinkPaimonJdbcBackendIT {}
diff --git
a/flink-connector/v1.20/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/paimon/FlinkPaimonLocalFileSystemBackendIT120.java
b/flink-connector/v1.20/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/paimon/FlinkPaimonLocalFileSystemBackendIT120.java
new file mode 100644
index 0000000000..26e5154080
--- /dev/null
+++
b/flink-connector/v1.20/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/paimon/FlinkPaimonLocalFileSystemBackendIT120.java
@@ -0,0 +1,22 @@
+/*
+ * 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.paimon;
+
+public class FlinkPaimonLocalFileSystemBackendIT120 extends
FlinkPaimonLocalFileSystemBackendIT {}
diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml
index 3f0945f9c2..ae67ef69ef 100644
--- a/gradle/libs.versions.toml
+++ b/gradle/libs.versions.toml
@@ -101,10 +101,18 @@ bouncycastle = "1.84"
jodd = "3.5.2"
flink = "1.18.0"
flink18 = "1.18.0"
+flink119 = "1.19.0"
+flink120 = "1.20.0"
flinkjdbc = "3.2.0-1.18"
flinkjdbc18 = "3.2.0-1.18"
+flinkjdbc119 = "3.3.0-1.19"
+flinkjdbc120 = "3.3.0-1.20"
iceberg4flink18 = "1.9.2"
+iceberg4flink119 = "1.9.2"
+iceberg4flink120 = "1.9.2"
paimon4flink18 = "1.2.0"
+paimon4flink119 = "1.2.0"
+paimon4flink120 = "1.2.0"
# cglib upgraded from 2. to 3.3.0 for Java 17 compatibility.
# Explicit ASM dependency (9.8) is required due to cglib's internal use and
Java 17 module system.
cglib = "3.3.0"
@@ -334,6 +342,8 @@ aliyun-credentials-sdk = { group='com.aliyun',
name='credentials-java', version.
aliyun-sdk-oss = { module = "com.aliyun.oss:aliyun-sdk-oss", version.ref =
"aliyun-sdk-oss" }
flinkjdbc = {group='org.apache.flink',name='flink-connector-jdbc',
version.ref='flinkjdbc'}
flinkjdbc18 = {group='org.apache.flink',name='flink-connector-jdbc',
version.ref='flinkjdbc18'}
+flinkjdbc119 = {group='org.apache.flink',name='flink-connector-jdbc',
version.ref='flinkjdbc119'}
+flinkjdbc120 = {group='org.apache.flink',name='flink-connector-jdbc',
version.ref='flinkjdbc120'}
concurrent-trees = { group = "com.googlecode.concurrent-trees", name =
"concurrent-trees", version.ref = "concurrent-trees" }
jcasbin = { group='org.casbin', name='jcasbin', version.ref="jcasbin" }
openlineage-java= { group = "io.openlineage", name = "openlineage-java",
version.ref = "openlineage" }
diff --git a/settings.gradle.kts b/settings.gradle.kts
index b3a9d6fb48..5f21cc7790 100644
--- a/settings.gradle.kts
+++ b/settings.gradle.kts
@@ -89,10 +89,23 @@ include("spark-connector:spark-common")
if (scalaVersion == "2.12") {
// flink only support scala 2.12
include("flink-connector:flink-common")
- include("flink-connector:flink-1.18", "flink-connector:flink-runtime-1.18")
+ include(
+ "flink-connector:flink-1.18",
+ "flink-connector:flink-runtime-1.18",
+ "flink-connector:flink-1.19",
+ "flink-connector:flink-runtime-1.19",
+ "flink-connector:flink-1.20",
+ "flink-connector:flink-runtime-1.20"
+ )
project(":flink-connector:flink-1.18").projectDir =
file("flink-connector/v1.18/flink")
project(":flink-connector:flink-runtime-1.18").projectDir =
file("flink-connector/v1.18/flink-runtime")
+ project(":flink-connector:flink-1.19").projectDir =
file("flink-connector/v1.19/flink")
+ project(":flink-connector:flink-runtime-1.19").projectDir =
+ file("flink-connector/v1.19/flink-runtime")
+ project(":flink-connector:flink-1.20").projectDir =
file("flink-connector/v1.20/flink")
+ project(":flink-connector:flink-runtime-1.20").projectDir =
+ file("flink-connector/v1.20/flink-runtime")
}
include("spark-connector:spark-3.3", "spark-connector:spark-runtime-3.3")
project(":spark-connector:spark-3.3").projectDir =
file("spark-connector/v3.3/spark")
