This is an automated email from the ASF dual-hosted git repository.
diqiu50 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new ff256c0ea5 [#13739] feat(flink-connector): Support Apache Flink 2.2
(#13740)
ff256c0ea5 is described below
commit ff256c0ea584f63d95e44e1c1add271b4dbb41bc
Author: Xu Bai <[email protected]>
AuthorDate: Wed Oct 14 22:13:43 2026 +0800
[#13739] feat(flink-connector): Support Apache Flink 2.2 (#13740)
### What changes were proposed in this pull request?
Add Flink 2.2 connector/runtime modules and update CI, Docker setup and
connector documentation.
### Why are the changes needed?
Enable Gravitino catalog workflows on Flink 2.2 while retaining existing
version support.
Fix: #13739
### Does this PR introduce _any_ user-facing change?
Yes. Add `gravitino-flink-connector-runtime-2.2-${version}.jar` for
Flink 2.2.1, supporting Iceberg, Paimon, JDBC, Catalog Store and session
catalogs. Hive remains unsupported on Flink 2.x.
### How was this patch tested?
- Flink 2.2 connector/runtime builds and unit tests passed, including 2
runtime packaging tests and a Scala 2.13 runtime build.
- 14 Iceberg unit tests and 7 REST deploy integration tests passed with
Iceberg 1.12.0 client and 1.11.0 server.
- Repository unit tests excluding Docker tests and formatting checks
passed.
---
.../workflows/backend-integration-test-action.yml | 1 +
.../workflows/flink-integration-test-action.yml | 4 +-
.../workflows/trino-integration-test-action.yml | 2 +
.../workflows/trino-multi-version-test-action.yml | 2 +
dev/docker/flink-connectors/Dockerfile | 11 ++-
dev/docker/flink-connectors/README.md | 7 +-
dev/docker/flink-connectors/copy-connector.sh | 10 +-
.../flink-connectors-dependency.sh | 9 +-
docs/flink-connector/flink-catalog-iceberg.md | 1 +
docs/flink-connector/flink-catalog-jdbc.md | 8 +-
docs/flink-connector/flink-catalog-paimon.md | 16 ++--
docs/flink-connector/flink-connector.md | 3 +-
.../v2.1/flink-runtime/build.gradle.kts | 5 +-
.../flink/runtime/TestRuntimeJarDependencies.java | 37 ++++++--
flink-connector/v2.1/flink/build.gradle.kts | 7 +-
.../{v2.1 => v2.2}/flink-runtime/build.gradle.kts | 9 +-
.../flink/runtime/TestRuntimeJarDependencies.java | 37 ++++++--
.../{v2.1 => v2.2}/flink/build.gradle.kts | 25 ++---
.../GravitinoIcebergCatalogFactoryFlink22.java | 47 +++++++++
.../iceberg/GravitinoIcebergCatalogFlink22.java | 105 +++++++++++++++++++++
.../jdbc/GravitinoJdbcCatalogFlink22.java | 64 +++++++++++++
.../GravitinoMysqlJdbcCatalogFactoryFlink22.java | 41 ++++++++
...GravitinoPostgresJdbcCatalogFactoryFlink22.java | 42 +++++++++
.../GravitinoPaimonCatalogFactoryFlink22.java | 39 ++++++++
.../paimon/GravitinoPaimonCatalogFlink22.java | 52 ++++++++++
.../connector/utils/CatalogCompatFlink22.java | 64 +++++++++++++
.../org.apache.flink.table.factories.Factory | 29 ++++++
.../connector/catalog/TestBaseCatalogFlink22.java | 31 ++++++
.../hive/TestFlinkGenericTableUtilFlink22.java | 31 ++++++
.../hive/TestGravitinoHiveCatalogFlink22.java | 31 ++++++
.../TestGravitinoIcebergCatalogFlink22.java | 84 +++++++++++++++++
.../test/catalog/GravitinoCatalogManagerIT22.java | 22 +++++
.../test/iceberg/FlinkIcebergHiveCatalogIT22.java | 22 +++++
.../test/iceberg/FlinkIcebergJdbcCatalogIT22.java | 29 ++++++
.../test/iceberg/FlinkIcebergRestCatalogIT22.java | 22 +++++
.../test/jdbc/FlinkJdbcMysqlCatalogIT22.java | 53 +++++++++++
.../test/paimon/FlinkPaimonHiveBackendIT22.java | 22 +++++
.../test/paimon/FlinkPaimonJdbcBackendIT22.java | 22 +++++
.../FlinkPaimonLocalFileSystemBackendIT22.java | 22 +++++
.../jdbc/TestGravitinoJdbcCatalogFlink22.java | 31 ++++++
.../paimon/TestGravitinoPaimonCatalogFlink22.java | 31 ++++++
.../connector/utils/TestCatalogCompatFlink22.java | 80 ++++++++++++++++
gradle/libs.versions.toml | 13 ++-
settings.gradle.kts | 7 +-
44 files changed, 1162 insertions(+), 68 deletions(-)
diff --git a/.github/workflows/backend-integration-test-action.yml
b/.github/workflows/backend-integration-test-action.yml
index cd26d94d6d..b5de5e641e 100644
--- a/.github/workflows/backend-integration-test-action.yml
+++ b/.github/workflows/backend-integration-test-action.yml
@@ -78,6 +78,7 @@ jobs:
-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 :flink-connector:flink-2.1:test -x
:flink-connector:flink-runtime-2.1:test \
+ -x :flink-connector:flink-2.2:test -x
:flink-connector:flink-runtime-2.2:test \
-x :spark-connector:spark-3.5:test -x
:spark-connector:spark-4.0:test \
-x :spark-connector:spark-runtime-3.5:test -x
:spark-connector:spark-runtime-4.0:test \
-x :trino-connector:integration-test:test -x
:trino-connector:trino-connector:test \
diff --git a/.github/workflows/flink-integration-test-action.yml
b/.github/workflows/flink-integration-test-action.yml
index efe786b92b..db792b457c 100644
--- a/.github/workflows/flink-integration-test-action.yml
+++ b/.github/workflows/flink-integration-test-action.yml
@@ -48,7 +48,7 @@ jobs:
id: integrationTest
# run embedded mode and deploy mode integration tests for every
supported Flink minor
run: |
- for v in 1.19 1.20 2.1; do
+ for v in 1.19 1.20 2.1 2.2; do
./gradlew -PskipTests -PtestMode=embedded -PskipDockerTests=false
-PskipWeb=true \
:flink-connector:flink-$v:test \
--tests
"org.apache.gravitino.flink.connector.integration.test.**"
@@ -70,6 +70,8 @@ jobs:
flink-connector/v1.20/flink/build/*.tar
flink-connector/v2.1/flink/build/*.log
flink-connector/v2.1/flink/build/*.tar
+ flink-connector/v2.2/flink/build/*.log
+ flink-connector/v2.2/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-action.yml
b/.github/workflows/trino-integration-test-action.yml
index 165706a59a..8435aa0494 100644
--- a/.github/workflows/trino-integration-test-action.yml
+++ b/.github/workflows/trino-integration-test-action.yml
@@ -46,6 +46,8 @@ jobs:
-x :flink-connector:flink-runtime-1.20:build \
-x :flink-connector:flink-2.1:build \
-x :flink-connector:flink-runtime-2.1:build \
+ -x :flink-connector:flink-2.2:build \
+ -x :flink-connector:flink-runtime-2.2:build \
-x :spark-connector:spark-3.5:build -x
:spark-connector:spark-4.0:build \
-x :spark-connector:spark-runtime-3.5:build -x
:spark-connector:spark-runtime-4.0:build \
-x :mcp-server:build -x :lineage:build \
diff --git a/.github/workflows/trino-multi-version-test-action.yml
b/.github/workflows/trino-multi-version-test-action.yml
index 0e81613f8f..9542928161 100644
--- a/.github/workflows/trino-multi-version-test-action.yml
+++ b/.github/workflows/trino-multi-version-test-action.yml
@@ -33,6 +33,8 @@ jobs:
-x :flink-connector:flink-runtime-1.20:build \
-x :flink-connector:flink-2.1:build \
-x :flink-connector:flink-runtime-2.1:build \
+ -x :flink-connector:flink-2.2:build \
+ -x :flink-connector:flink-runtime-2.2:build \
-x :spark-connector:spark-3.5:build -x
:spark-connector:spark-4.0:build \
-x :spark-connector:spark-runtime-3.5:build -x
:spark-connector:spark-runtime-4.0:build \
-x :mcp-server:build -x :lineage:build \
diff --git a/dev/docker/flink-connectors/Dockerfile
b/dev/docker/flink-connectors/Dockerfile
index a9a11439ba..2b6f164991 100644
--- a/dev/docker/flink-connectors/Dockerfile
+++ b/dev/docker/flink-connectors/Dockerfile
@@ -20,9 +20,10 @@
#
==============================================================================
# Apache Gravitino Flink Connector — UBI 10-based, certification-oriented image
# Contains the Apache Gravitino Flink connector runtime jars for every
-# supported Flink version present in the source tree (Scala 2.12; Flink does
-# not support Scala 2.13). The connector code is Apache-2.0; the shaded runtime
-# jar bundles third-party open source components (see the licenses/ directory).
+# supported Flink version present in the source tree. Flink 1.x jars use the
+# Scala 2.12 suffix; Flink 2.x jars have no Scala suffix. The connector code is
+# Apache-2.0; the shaded runtime jar bundles third-party open source components
+# (see the licenses/ directory).
# Used as a Kubernetes init-container to inject connector jars into Flink pods.
#
# The set of Flink versions is discovered at build time from the Gravitino
@@ -56,8 +57,8 @@ LABEL name="gravitino-flink-connector" \
vendor="The Apache Software Foundation" \
version="${IMAGE_VERSION}" \
release="${IMAGE_RELEASE}" \
- summary="Apache Gravitino Flink Connector (Scala 2.12)" \
- description="Apache Gravitino Flink connector runtime jars for all
supported Flink versions (Scala 2.12), for Kubernetes init-container
deployment. The version is selected at runtime via the FLINK_VERSION
environment variable." \
+ summary="Apache Gravitino Flink Connector" \
+ description="Apache Gravitino Flink connector runtime jars for all
supported Flink versions, for Kubernetes init-container deployment. Flink 1.x
jars use the Scala 2.12 suffix; Flink 2.x jars have no Scala suffix. The
version is selected at runtime via the FLINK_VERSION environment variable." \
maintainer="Apache Gravitino <[email protected]>"
# Default Flink version (can be overridden at runtime)
diff --git a/dev/docker/flink-connectors/README.md
b/dev/docker/flink-connectors/README.md
index 68d35d708a..6b54fb383a 100644
--- a/dev/docker/flink-connectors/README.md
+++ b/dev/docker/flink-connectors/README.md
@@ -59,13 +59,14 @@ server version, or keep it lower.
## Supported versions
-Each Flink minor version ships as its own directory under `/connectors`
-(Scala 2.12; Flink does not support Scala 2.13). The exact set baked into an
+Flink 1.19, 1.20, 2.1, and 2.2 are supported. Each Flink minor version ships
as its own directory under
+`/connectors`. Flink 1.x runtime JAR names include the `_2.12` Scala suffix;
+Flink 2.x runtime JAR names have no Scala suffix. The exact set baked into an
image depends on the Gravitino source branch it was built from; list them
with:
```bash
-docker run --rm apache/gravitino-flink-connector:{version}
+docker run --rm -e LIST_VERSIONS=true
apache/gravitino-flink-connector:{version}
```
Each directory contains one shaded runtime JAR. Do not mix JARs from
diff --git a/dev/docker/flink-connectors/copy-connector.sh
b/dev/docker/flink-connectors/copy-connector.sh
index 591aa7c6dc..12f4b35ba2 100755
--- a/dev/docker/flink-connectors/copy-connector.sh
+++ b/dev/docker/flink-connectors/copy-connector.sh
@@ -27,8 +27,8 @@
# Useful for `docker run --rm <image>`.
#
# The available versions are DISCOVERED at runtime from the directories baked
-# into /connectors (flink-<ver>). All versions are Scala 2.12 only (Flink does
-# not support Scala 2.13).
+# into /connectors (flink-<ver>). Flink 1.x jars use the Scala 2.12 suffix;
+# Flink 2.x jars have no Scala suffix.
#
# As an init container, a missing /target volume is treated as an error so a
# misconfigured pod fails fast instead of letting the engine start without the
@@ -44,7 +44,7 @@ list_available_versions() {
}
if [ "${LIST_VERSIONS}" = "true" ]; then
- echo "Apache Gravitino Flink connector jars available at /connectors/ (all
Scala 2.12):"
+ echo "Apache Gravitino Flink connector jars available at /connectors/:"
echo ""
list_available_versions
echo ""
@@ -64,12 +64,12 @@ SOURCE_DIR="/connectors/flink-${FLINK_VERSION}"
if [ ! -d "$SOURCE_DIR" ]; then
echo "ERROR: Flink version ${FLINK_VERSION} is not supported by this image."
>&2
echo "" >&2
- echo "Available versions (all Scala 2.12):" >&2
+ echo "Available versions:" >&2
list_available_versions >&2
exit 1
fi
-echo "Copying Flink ${FLINK_VERSION} connector (Scala 2.12) to /target/..."
+echo "Copying Flink ${FLINK_VERSION} connector to /target/..."
cp "${SOURCE_DIR}"/*.jar /target/
echo "Done. Jars copied to /target/:"
ls -1 /target/*.jar 2>/dev/null
diff --git a/dev/docker/flink-connectors/flink-connectors-dependency.sh
b/dev/docker/flink-connectors/flink-connectors-dependency.sh
index 5085460b87..2cd64ec8e9 100755
--- a/dev/docker/flink-connectors/flink-connectors-dependency.sh
+++ b/dev/docker/flink-connectors/flink-connectors-dependency.sh
@@ -20,14 +20,15 @@
# Build every Apache Gravitino Flink connector runtime shadow jar present in
# the source tree and prepare the layout consumed by the Docker image.
-# Flink only supports Scala 2.12.
+# Flink 1.x jars use the Scala 2.12 suffix; Flink 2.x jars have no Scala
suffix.
#
# The set of Flink runtime modules is DISCOVERED from the Gradle project, so
# this script does not hard-code which versions exist. Whatever the checked-out
-# branch supports (for example 1.19/1.20/2.1) is built automatically.
+# branch supports (for example 1.19/1.20/2.1/2.2) is built automatically.
#
# Output layout:
-#
packages/connectors/flink-<ver>/gravitino-flink-connector-runtime-<ver>_2.12-*.jar
+#
packages/connectors/flink-1.<minor>/gravitino-flink-connector-runtime-1.<minor>_2.12-*.jar
+#
packages/connectors/flink-2.<minor>/gravitino-flink-connector-runtime-2.<minor>-*.jar
set -euo pipefail
@@ -52,7 +53,7 @@ versions="$(echo "${runtime_modules}" | sed
's/flink-runtime-//' | sort -u)"
echo "Discovered Flink connector versions:"
echo "${versions}" | sed 's/^/ - /'
-# Assemble each discovered runtime shadow jar (Scala 2.12 only).
+# Assemble each discovered runtime shadow jar.
tasks=""
for m in ${runtime_modules}; do
tasks="${tasks} :flink-connector:${m}:shadowJar"
diff --git a/docs/flink-connector/flink-catalog-iceberg.md
b/docs/flink-connector/flink-catalog-iceberg.md
index c347d93682..db88e7a2b4 100644
--- a/docs/flink-connector/flink-catalog-iceberg.md
+++ b/docs/flink-connector/flink-catalog-iceberg.md
@@ -45,6 +45,7 @@ Flink clients use a different Iceberg version than the
Gravitino server (1.11.0)
| 1.19 | 2.12 | 1.10.2 |
`iceberg-flink-runtime-1.19-1.10.2.jar` |
`gravitino-flink-connector-runtime-1.19_2.12-${gravitino-version}.jar` |
| 1.20 | 2.12 | 1.11.0 |
`iceberg-flink-runtime-1.20-1.11.0.jar` |
`gravitino-flink-connector-runtime-1.20_2.12-${gravitino-version}.jar` |
| 2.1 | n/a | 1.11.0 |
`iceberg-flink-runtime-2.1-1.11.0.jar` |
`gravitino-flink-connector-runtime-2.1-${gravitino-version}.jar` |
+| 2.2 | n/a | 1.12.0 |
`iceberg-flink-runtime-2.2-1.12.0.jar` |
`gravitino-flink-connector-runtime-2.2-${gravitino-version}.jar` |
Replace `${gravitino-version}` with your Gravitino release version.
diff --git a/docs/flink-connector/flink-catalog-jdbc.md
b/docs/flink-connector/flink-catalog-jdbc.md
index d50cc4f18b..8b8ac74226 100644
--- a/docs/flink-connector/flink-catalog-jdbc.md
+++ b/docs/flink-connector/flink-catalog-jdbc.md
@@ -31,8 +31,14 @@ Place the following JAR files in the lib directory of your
Flink installation:
| 1.19 | `3.3.0-1.19` |
`gravitino-flink-connector-runtime-1.19_2.12-${gravitino-version}.jar` |
| 1.20 | `3.3.0-1.20` |
`gravitino-flink-connector-runtime-1.20_2.12-${gravitino-version}.jar` |
| 2.1 | `4.1.0-2.1` |
`gravitino-flink-connector-runtime-2.1-${gravitino-version}.jar` |
+| 2.2 | `4.1.0-2.2` |
`gravitino-flink-connector-runtime-2.2-${gravitino-version}.jar` |
-Starting with Flink 2.1, `flink-connector-jdbc` is split into per-database
artifacts: place `flink-connector-jdbc-core-4.1.0-2.1.jar`,
`flink-connector-jdbc-mysql-4.1.0-2.1.jar`, and
`flink-connector-jdbc-postgres-4.1.0-2.1.jar` together in the classpath instead
of a single combined JAR.
+For Flink 2.x, `flink-connector-jdbc` is split into core and per-database
artifacts. Place all three JARs from the matching row in the classpath instead
of a single combined JAR:
+
+| Flink version | Core artifact | MySQL artifact
| PostgreSQL artifact |
+|---------------|-------------------------------------------|--------------------------------------------|-----------------------------------------------|
+| 2.1 | `flink-connector-jdbc-core-4.1.0-2.1.jar` |
`flink-connector-jdbc-mysql-4.1.0-2.1.jar` |
`flink-connector-jdbc-postgres-4.1.0-2.1.jar` |
+| 2.2 | `flink-connector-jdbc-core-4.1.0-2.2.jar` |
`flink-connector-jdbc-mysql-4.1.0-2.2.jar` |
`flink-connector-jdbc-postgres-4.1.0-2.2.jar` |
Next, when you create the JDBC catalog in Gravitino, add the
`flink.bypass.default-database` property with the value of the default database
name.
diff --git a/docs/flink-connector/flink-catalog-paimon.md
b/docs/flink-connector/flink-catalog-paimon.md
index 09669b7568..e21a70a077 100644
--- a/docs/flink-connector/flink-catalog-paimon.md
+++ b/docs/flink-connector/flink-catalog-paimon.md
@@ -39,9 +39,10 @@ Supports most DDL and DML operations in Flink SQL, except
such operations:
## Prerequisites
-* Paimon 1.2.0 is fully tested.
+* Flink 1.19 and 1.20 use Paimon 1.2.0.
+* Flink 2.1 and 2.2 use Paimon 1.4.1.
-Other Paimon versions may also work but have not been tested fully.
+The supported Paimon versions are pinned in the table below. Other Paimon
versions may also work but have not been tested fully.
## Getting Started
@@ -52,11 +53,12 @@ Place the following JAR files in the lib directory of your
Flink installation:
- The Paimon Flink connector JAR that matches your Flink minor version
- The Gravitino Flink connector runtime JAR that matches your Flink minor
version
-| Flink version | Paimon connector artifact | Gravitino
runtime artifact |
-|---------------|-------------------------------------------|------------------------------------------------------------------------|
-| 1.19 | `paimon-flink-1.19-${paimon-version}.jar` |
`gravitino-flink-connector-runtime-1.19_2.12-${gravitino-version}.jar` |
-| 1.20 | `paimon-flink-1.20-${paimon-version}.jar` |
`gravitino-flink-connector-runtime-1.20_2.12-${gravitino-version}.jar` |
-| 2.1 | `paimon-flink-2.1-${paimon-version}.jar` |
`gravitino-flink-connector-runtime-2.1-${gravitino-version}.jar` |
+| Flink version | Paimon version | Paimon connector artifact |
Gravitino runtime artifact |
+|---------------|----------------|-------------------------------------|------------------------------------------------------------------------|
+| 1.19 | 1.2.0 | `paimon-flink-1.19-1.2.0.jar` |
`gravitino-flink-connector-runtime-1.19_2.12-${gravitino-version}.jar` |
+| 1.20 | 1.2.0 | `paimon-flink-1.20-1.2.0.jar` |
`gravitino-flink-connector-runtime-1.20_2.12-${gravitino-version}.jar` |
+| 2.1 | 1.4.1 | `paimon-flink-2.1-1.4.1.jar` |
`gravitino-flink-connector-runtime-2.1-${gravitino-version}.jar` |
+| 2.2 | 1.4.1 | `paimon-flink-2.2-1.4.1.jar` |
`gravitino-flink-connector-runtime-2.2-${gravitino-version}.jar` |
### SQL Example
diff --git a/docs/flink-connector/flink-connector.md
b/docs/flink-connector/flink-connector.md
index f239eba324..1340e2f383 100644
--- a/docs/flink-connector/flink-connector.md
+++ b/docs/flink-connector/flink-connector.md
@@ -24,7 +24,7 @@ The connector is published as a version-specific runtime JAR
for each supported
## Prerequisites
* Scala 2.12 (Flink 1.19 and 1.20 only; Flink 2.x removed the Scala APIs)
-* Flink 1.19, 1.20, or 2.1
+* Flink 1.19, 1.20, 2.1, or 2.2
* JDK 8, 11 or 17 for Flink 1.x; JDK 17 for Flink 2.x
On Flink 2.x, the Hive catalog is not supported: `flink-connector-hive` has
not published a
@@ -41,6 +41,7 @@ catalogs are supported on Flink 2.x.
| 1.19 |
`gravitino-flink-connector-runtime-1.19_2.12-${gravitino-version}.jar` |
| 1.20 |
`gravitino-flink-connector-runtime-1.20_2.12-${gravitino-version}.jar` |
| 2.1 |
`gravitino-flink-connector-runtime-2.1-${gravitino-version}.jar` |
+| 2.2 |
`gravitino-flink-connector-runtime-2.2-${gravitino-version}.jar` |
Do not mix runtime JARs from different Flink minor versions in the same Flink
deployment.
diff --git a/flink-connector/v2.1/flink-runtime/build.gradle.kts
b/flink-connector/v2.1/flink-runtime/build.gradle.kts
index 61c21a2556..84d56553b5 100644
--- a/flink-connector/v2.1/flink-runtime/build.gradle.kts
+++ b/flink-connector/v2.1/flink-runtime/build.gradle.kts
@@ -46,11 +46,11 @@ configurations.all {
dependencies {
implementation(project(":clients:client-java-runtime", configuration =
"shadow"))
implementation(project(":flink-connector:flink-2.1"))
- // flink-2.1's own openlineageSqlJava21 constraint only applies to that
module's compileOnly and
+ // flink-2.1's own openlineageSqlJava constraint only applies to that
module's compileOnly and
// testImplementation configurations, so it never reaches this module's
runtimeClasspath (which
// shadowJar packages). Depend on it directly here so the shaded runtime jar
ships the patched
// 1.52.0 release instead of the vulnerable 1.32.0 pulled transitively by
flink-connector-jdbc.
- implementation(libs.openlineageSqlJava21)
+ implementation(libs.openlineageSqlJava)
testImplementation(libs.junit.jupiter.api)
testRuntimeOnly(libs.junit.jupiter.engine)
@@ -76,6 +76,7 @@ shadowJarTask.configure {
tasks.test {
useJUnitPlatform()
dependsOn(shadowJarTask)
+ inputs.file(shadowJarTask.flatMap { it.archiveFile })
doFirst {
systemProperty("shadowJarPath",
shadowJarTask.get().archiveFile.get().asFile.absolutePath)
}
diff --git
a/flink-connector/v2.1/flink-runtime/src/test/java/org/apache/gravitino/flink/runtime/TestRuntimeJarDependencies.java
b/flink-connector/v2.1/flink-runtime/src/test/java/org/apache/gravitino/flink/runtime/TestRuntimeJarDependencies.java
index 067cd616a6..d7307b874d 100644
---
a/flink-connector/v2.1/flink-runtime/src/test/java/org/apache/gravitino/flink/runtime/TestRuntimeJarDependencies.java
+++
b/flink-connector/v2.1/flink-runtime/src/test/java/org/apache/gravitino/flink/runtime/TestRuntimeJarDependencies.java
@@ -18,6 +18,7 @@
*/
package org.apache.gravitino.flink.runtime;
+import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -26,7 +27,9 @@ import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
+import java.util.HashSet;
import java.util.List;
+import java.util.Set;
import java.util.jar.JarEntry;
import java.util.jar.JarFile;
import java.util.stream.Collectors;
@@ -69,14 +72,32 @@ class TestRuntimeJarDependencies {
try (InputStream in = jarFile.getInputStream(factoryServiceEntry)) {
factoryServices = new String(in.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.paimon.GravitinoPaimonCatalogFactoryFlink21"),
- "Runtime jar should include the version-specific catalog factory
descriptor");
+ List<String> gravitinoFactories =
+ factoryServices
+ .lines()
+ .map(line -> line.split("#", 2)[0].trim())
+ .filter(line -> line.startsWith("org.apache.gravitino.flink."))
+ .collect(Collectors.toList());
+ Set<String> expectedFactories =
+ Set.of(
+
"org.apache.gravitino.flink.connector.store.GravitinoCatalogStoreFactory",
+
"org.apache.gravitino.flink.connector.paimon.GravitinoPaimonCatalogFactoryFlink21",
+
"org.apache.gravitino.flink.connector.iceberg.GravitinoIcebergCatalogFactoryFlink21",
+
"org.apache.gravitino.flink.connector.jdbc.mysql.GravitinoMysqlJdbcCatalogFactoryFlink21",
+
"org.apache.gravitino.flink.connector.jdbc.postgresql.GravitinoPostgresJdbcCatalogFactoryFlink21");
+ assertEquals(
+ expectedFactories,
+ new HashSet<>(gravitinoFactories),
+ "Runtime jar should register exactly the shared catalog-store and
four Flink 2.1 factories");
+ assertEquals(
+ expectedFactories.size(),
+ gravitinoFactories.size(),
+ "Runtime jar should not contain duplicate Gravitino factory
registrations");
+ for (String factory : expectedFactories) {
+ assertNotNull(
+ jarFile.getJarEntry(factory.replace('.', '/') + ".class"),
+ "Registered factory class should exist in runtime jar: " +
factory);
+ }
}
}
}
diff --git a/flink-connector/v2.1/flink/build.gradle.kts
b/flink-connector/v2.1/flink/build.gradle.kts
index 803eb3e52c..e8758b7086 100644
--- a/flink-connector/v2.1/flink/build.gradle.kts
+++ b/flink-connector/v2.1/flink/build.gradle.kts
@@ -56,7 +56,8 @@ val flinkVersion: String = libs.versions.flink21.get()
val flinkMajorVersion: String = flinkVersion.substringBeforeLast(".")
val icebergVersion: String = libs.versions.iceberg4flink21.get()
val paimonVersion: String = libs.versions.paimon4flink21.get()
-// Flink 2.x removed the Scala APIs entirely, so unlike the 1.x modules there
is no scala suffix.
+// This connector uses Flink Java APIs, so its artifacts need no Scala-version
suffix.
+// Flink internals still use Scala.
val artifactName = "${rootProject.name}-flink-$flinkMajorVersion"
dependencies {
@@ -70,8 +71,8 @@ dependencies {
// can't simply be excluded: the connector's lineage extraction actually
calls into it at
// runtime (JdbcSource.getLineageVertex). Force this newer release
instead, which ships an
// up-to-date bundled copy.
- compileOnly(libs.openlineageSqlJava21)
- testImplementation(libs.openlineageSqlJava21)
+ compileOnly(libs.openlineageSqlJava)
+ testImplementation(libs.openlineageSqlJava)
}
// Dependencies needed to compile flink-common's sources (mirrors
flink-common/build.gradle.kts).
diff --git a/flink-connector/v2.1/flink-runtime/build.gradle.kts
b/flink-connector/v2.2/flink-runtime/build.gradle.kts
similarity index 90%
copy from flink-connector/v2.1/flink-runtime/build.gradle.kts
copy to flink-connector/v2.2/flink-runtime/build.gradle.kts
index 61c21a2556..e3b307d8c4 100644
--- a/flink-connector/v2.1/flink-runtime/build.gradle.kts
+++ b/flink-connector/v2.2/flink-runtime/build.gradle.kts
@@ -29,7 +29,7 @@ repositories {
mavenCentral()
}
-val flinkVersion: String = libs.versions.flink21.get()
+val flinkVersion: String = libs.versions.flink22.get()
val flinkMajorVersion: String = flinkVersion.substringBeforeLast(".")
// Flink 2.x removed the Scala APIs entirely, so unlike the 1.x modules there
is no scala suffix.
val artifactName = "gravitino-${project.name}"
@@ -45,12 +45,12 @@ configurations.all {
dependencies {
implementation(project(":clients:client-java-runtime", configuration =
"shadow"))
- implementation(project(":flink-connector:flink-2.1"))
- // flink-2.1's own openlineageSqlJava21 constraint only applies to that
module's compileOnly and
+ implementation(project(":flink-connector:flink-2.2"))
+ // flink-2.2's own openlineageSqlJava constraint only applies to that
module's compileOnly and
// testImplementation configurations, so it never reaches this module's
runtimeClasspath (which
// shadowJar packages). Depend on it directly here so the shaded runtime jar
ships the patched
// 1.52.0 release instead of the vulnerable 1.32.0 pulled transitively by
flink-connector-jdbc.
- implementation(libs.openlineageSqlJava21)
+ implementation(libs.openlineageSqlJava)
testImplementation(libs.junit.jupiter.api)
testRuntimeOnly(libs.junit.jupiter.engine)
@@ -76,6 +76,7 @@ shadowJarTask.configure {
tasks.test {
useJUnitPlatform()
dependsOn(shadowJarTask)
+ inputs.file(shadowJarTask.flatMap { it.archiveFile })
doFirst {
systemProperty("shadowJarPath",
shadowJarTask.get().archiveFile.get().asFile.absolutePath)
}
diff --git
a/flink-connector/v2.1/flink-runtime/src/test/java/org/apache/gravitino/flink/runtime/TestRuntimeJarDependencies.java
b/flink-connector/v2.2/flink-runtime/src/test/java/org/apache/gravitino/flink/runtime/TestRuntimeJarDependencies.java
similarity index 69%
copy from
flink-connector/v2.1/flink-runtime/src/test/java/org/apache/gravitino/flink/runtime/TestRuntimeJarDependencies.java
copy to
flink-connector/v2.2/flink-runtime/src/test/java/org/apache/gravitino/flink/runtime/TestRuntimeJarDependencies.java
index 067cd616a6..1187a46b33 100644
---
a/flink-connector/v2.1/flink-runtime/src/test/java/org/apache/gravitino/flink/runtime/TestRuntimeJarDependencies.java
+++
b/flink-connector/v2.2/flink-runtime/src/test/java/org/apache/gravitino/flink/runtime/TestRuntimeJarDependencies.java
@@ -18,6 +18,7 @@
*/
package org.apache.gravitino.flink.runtime;
+import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -26,7 +27,9 @@ import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
+import java.util.HashSet;
import java.util.List;
+import java.util.Set;
import java.util.jar.JarEntry;
import java.util.jar.JarFile;
import java.util.stream.Collectors;
@@ -69,14 +72,32 @@ class TestRuntimeJarDependencies {
try (InputStream in = jarFile.getInputStream(factoryServiceEntry)) {
factoryServices = new String(in.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.paimon.GravitinoPaimonCatalogFactoryFlink21"),
- "Runtime jar should include the version-specific catalog factory
descriptor");
+ List<String> gravitinoFactories =
+ factoryServices
+ .lines()
+ .map(line -> line.split("#", 2)[0].trim())
+ .filter(line -> line.startsWith("org.apache.gravitino.flink."))
+ .collect(Collectors.toList());
+ Set<String> expectedFactories =
+ Set.of(
+
"org.apache.gravitino.flink.connector.store.GravitinoCatalogStoreFactory",
+
"org.apache.gravitino.flink.connector.paimon.GravitinoPaimonCatalogFactoryFlink22",
+
"org.apache.gravitino.flink.connector.iceberg.GravitinoIcebergCatalogFactoryFlink22",
+
"org.apache.gravitino.flink.connector.jdbc.mysql.GravitinoMysqlJdbcCatalogFactoryFlink22",
+
"org.apache.gravitino.flink.connector.jdbc.postgresql.GravitinoPostgresJdbcCatalogFactoryFlink22");
+ assertEquals(
+ expectedFactories,
+ new HashSet<>(gravitinoFactories),
+ "Runtime jar should register exactly the shared catalog-store and
four Flink 2.2 factories");
+ assertEquals(
+ expectedFactories.size(),
+ gravitinoFactories.size(),
+ "Runtime jar should not contain duplicate Gravitino factory
registrations");
+ for (String factory : expectedFactories) {
+ assertNotNull(
+ jarFile.getJarEntry(factory.replace('.', '/') + ".class"),
+ "Registered factory class should exist in runtime jar: " +
factory);
+ }
}
}
}
diff --git a/flink-connector/v2.1/flink/build.gradle.kts
b/flink-connector/v2.2/flink/build.gradle.kts
similarity index 94%
copy from flink-connector/v2.1/flink/build.gradle.kts
copy to flink-connector/v2.2/flink/build.gradle.kts
index 803eb3e52c..e0a3bae162 100644
--- a/flink-connector/v2.1/flink/build.gradle.kts
+++ b/flink-connector/v2.2/flink/build.gradle.kts
@@ -52,11 +52,12 @@
plugins.withType<com.diffplug.gradle.spotless.SpotlessPlugin>().configureEach {
}
}
-val flinkVersion: String = libs.versions.flink21.get()
+val flinkVersion: String = libs.versions.flink22.get()
val flinkMajorVersion: String = flinkVersion.substringBeforeLast(".")
-val icebergVersion: String = libs.versions.iceberg4flink21.get()
-val paimonVersion: String = libs.versions.paimon4flink21.get()
-// Flink 2.x removed the Scala APIs entirely, so unlike the 1.x modules there
is no scala suffix.
+val icebergVersion: String = libs.versions.iceberg4flink22.get()
+val paimonVersion: String = libs.versions.paimon4flink22.get()
+// This connector uses Flink Java APIs, so its artifacts need no Scala-version
suffix.
+// Flink internals still use Scala.
val artifactName = "${rootProject.name}-flink-$flinkMajorVersion"
dependencies {
@@ -70,8 +71,8 @@ dependencies {
// can't simply be excluded: the connector's lineage extraction actually
calls into it at
// runtime (JdbcSource.getLineageVertex). Force this newer release
instead, which ships an
// up-to-date bundled copy.
- compileOnly(libs.openlineageSqlJava21)
- testImplementation(libs.openlineageSqlJava21)
+ compileOnly(libs.openlineageSqlJava)
+ testImplementation(libs.openlineageSqlJava)
}
// Dependencies needed to compile flink-common's sources (mirrors
flink-common/build.gradle.kts).
@@ -93,9 +94,9 @@ dependencies {
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.flinkjdbc21.core)
- compileOnly(libs.flinkjdbc21.mysql)
- compileOnly(libs.flinkjdbc21.postgres)
+ compileOnly(libs.flinkjdbc22.core)
+ compileOnly(libs.flinkjdbc22.mysql)
+ compileOnly(libs.flinkjdbc22.postgres)
compileOnly(libs.hive2.common) {
exclude("org.eclipse.jetty.aggregate", "jetty-all")
exclude("org.eclipse.jetty.orbit", "javax.servlet")
@@ -133,9 +134,9 @@ dependencies {
testImplementation(project(":server"))
testImplementation(project(":server-common"))
testImplementation(libs.awaitility)
- testImplementation(libs.flinkjdbc21.core)
- testImplementation(libs.flinkjdbc21.mysql)
- testImplementation(libs.flinkjdbc21.postgres)
+ testImplementation(libs.flinkjdbc22.core)
+ testImplementation(libs.flinkjdbc22.mysql)
+ testImplementation(libs.flinkjdbc22.postgres)
testImplementation(libs.junit.jupiter.api)
testImplementation(libs.junit.jupiter.params)
testImplementation(libs.metrics.core)
diff --git
a/flink-connector/v2.2/flink/src/main/java/org/apache/gravitino/flink/connector/iceberg/GravitinoIcebergCatalogFactoryFlink22.java
b/flink-connector/v2.2/flink/src/main/java/org/apache/gravitino/flink/connector/iceberg/GravitinoIcebergCatalogFactoryFlink22.java
new file mode 100644
index 0000000000..1d09ae3ed8
--- /dev/null
+++
b/flink-connector/v2.2/flink/src/main/java/org/apache/gravitino/flink/connector/iceberg/GravitinoIcebergCatalogFactoryFlink22.java
@@ -0,0 +1,47 @@
+/*
+ * 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 2.2. */
+public class GravitinoIcebergCatalogFactoryFlink22 extends
GravitinoIcebergCatalogFactory {
+
+ /** {@inheritDoc} */
+ @Override
+ protected Catalog newCatalog(
+ String catalogName,
+ String defaultDatabase,
+ SchemaAndTablePropertiesConverter schemaAndTablePropertiesConverter,
+ PartitionConverter partitionConverter,
+ Map<String, String> catalogOptions,
+ Map<String, String> icebergCatalogProperties) {
+ return new GravitinoIcebergCatalogFlink22(
+ catalogName,
+ defaultDatabase,
+ schemaAndTablePropertiesConverter,
+ partitionConverter,
+ catalogOptions,
+ icebergCatalogProperties);
+ }
+}
diff --git
a/flink-connector/v2.2/flink/src/main/java/org/apache/gravitino/flink/connector/iceberg/GravitinoIcebergCatalogFlink22.java
b/flink-connector/v2.2/flink/src/main/java/org/apache/gravitino/flink/connector/iceberg/GravitinoIcebergCatalogFlink22.java
new file mode 100644
index 0000000000..44e8d2cf4f
--- /dev/null
+++
b/flink-connector/v2.2/flink/src/main/java/org/apache/gravitino/flink/connector/iceberg/GravitinoIcebergCatalogFlink22.java
@@ -0,0 +1,105 @@
+/*
+ * 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.configuration.Configuration;
+import org.apache.flink.configuration.ReadableConfig;
+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.CatalogCompatFlink22;
+import org.apache.iceberg.flink.FlinkCatalogFactory;
+
+/** {@link GravitinoIcebergCatalog} implementation for Flink 2.2. */
+public class GravitinoIcebergCatalogFlink22 extends GravitinoIcebergCatalog {
+
+ /**
+ * Creates a catalog backed by the Flink 2.2 connector.
+ *
+ * @param catalogName the catalog name
+ * @param defaultDatabase the default database name
+ * @param schemaAndTablePropertiesConverter the converter for schemas and
table properties
+ * @param partitionConverter the converter for partition specifications
+ * @param catalogOptions the Gravitino catalog options
+ * @param icebergCatalogProperties the properties of the underlying Iceberg
catalog
+ */
+ protected GravitinoIcebergCatalogFlink22(
+ String catalogName,
+ String defaultDatabase,
+ SchemaAndTablePropertiesConverter schemaAndTablePropertiesConverter,
+ PartitionConverter partitionConverter,
+ Map<String, String> catalogOptions,
+ Map<String, String> icebergCatalogProperties) {
+ super(
+ catalogName,
+ defaultDatabase,
+ schemaAndTablePropertiesConverter,
+ partitionConverter,
+ catalogOptions,
+ icebergCatalogProperties);
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ protected Object createInnerIcebergCatalog(String catalogName, Map<String,
String> properties) {
+ // Flink 2.x removed the 2-arg createCatalog(String, Map) overload and
made the 3-arg
+ // (String, Map, org.apache.hadoop.conf.Configuration) overload protected;
only the public
+ // createCatalog(Context) overload remains usable from outside the
iceberg-flink package.
+ return new FlinkCatalogFactory()
+ .createCatalog(
+ new CatalogFactory.Context() {
+ /** {@inheritDoc} */
+ @Override
+ public String getName() {
+ return catalogName;
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public Map<String, String> getOptions() {
+ return properties;
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public ReadableConfig getConfiguration() {
+ // FlinkCatalogFactory.createCatalog(Context) does not
currently read
+ // getConfiguration() (it derives its Hadoop Configuration from
+ // FlinkCatalogFactory.clusterHadoopConf() instead), so an
empty one is a no-op
+ // stub to satisfy the interface. Re-check this on
iceberg-flink upgrades.
+ return new Configuration();
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public ClassLoader getClassLoader() {
+ return GravitinoIcebergCatalogFlink22.class.getClassLoader();
+ }
+ });
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ protected CatalogCompat catalogCompat() {
+ return CatalogCompatFlink22.INSTANCE;
+ }
+}
diff --git
a/flink-connector/v2.2/flink/src/main/java/org/apache/gravitino/flink/connector/jdbc/GravitinoJdbcCatalogFlink22.java
b/flink-connector/v2.2/flink/src/main/java/org/apache/gravitino/flink/connector/jdbc/GravitinoJdbcCatalogFlink22.java
new file mode 100644
index 0000000000..ce0fd3e735
--- /dev/null
+++
b/flink-connector/v2.2/flink/src/main/java/org/apache/gravitino/flink/connector/jdbc/GravitinoJdbcCatalogFlink22.java
@@ -0,0 +1,64 @@
+/*
+ * 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
org.apache.flink.connector.jdbc.core.database.catalog.factory.JdbcCatalogFactory;
+import org.apache.flink.table.catalog.AbstractCatalog;
+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.CatalogCompatFlink22;
+
+/**
+ * {@link GravitinoJdbcCatalog} implementation for Flink 2.2. Uses the {@code
+ * org.apache.flink.connector.jdbc.core.*} factory packages shipped with {@code
+ * flink-connector-jdbc-core}/{@code -mysql}/{@code -postgres} 4.1.0-2.2.
+ */
+public class GravitinoJdbcCatalogFlink22 extends GravitinoJdbcCatalog {
+
+ /**
+ * Creates a catalog backed by the Flink 2.2 connector.
+ *
+ * @param context the catalog factory context
+ * @param defaultDatabase the default database name
+ * @param schemaAndTablePropertiesConverter the converter for schemas and
table properties
+ * @param partitionConverter the converter for partition specifications
+ */
+ public GravitinoJdbcCatalogFlink22(
+ CatalogFactory.Context context,
+ String defaultDatabase,
+ SchemaAndTablePropertiesConverter schemaAndTablePropertiesConverter,
+ PartitionConverter partitionConverter) {
+ super(context, defaultDatabase, schemaAndTablePropertiesConverter,
partitionConverter);
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ protected AbstractCatalog createInnerCatalog(CatalogFactory.Context context)
{
+ return (AbstractCatalog) new JdbcCatalogFactory().createCatalog(context);
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ protected CatalogCompat catalogCompat() {
+ return CatalogCompatFlink22.INSTANCE;
+ }
+}
diff --git
a/flink-connector/v2.2/flink/src/main/java/org/apache/gravitino/flink/connector/jdbc/mysql/GravitinoMysqlJdbcCatalogFactoryFlink22.java
b/flink-connector/v2.2/flink/src/main/java/org/apache/gravitino/flink/connector/jdbc/mysql/GravitinoMysqlJdbcCatalogFactoryFlink22.java
new file mode 100644
index 0000000000..9a6e7e3ce6
--- /dev/null
+++
b/flink-connector/v2.2/flink/src/main/java/org/apache/gravitino/flink/connector/jdbc/mysql/GravitinoMysqlJdbcCatalogFactoryFlink22.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.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.GravitinoJdbcCatalogFlink22;
+
+/** {@link GravitinoMysqlJdbcCatalogFactory} implementation for Flink 2.2. */
+public class GravitinoMysqlJdbcCatalogFactoryFlink22 extends
GravitinoMysqlJdbcCatalogFactory {
+
+ /** {@inheritDoc} */
+ @Override
+ protected Catalog newCatalog(
+ CatalogFactory.Context context,
+ String defaultDatabase,
+ SchemaAndTablePropertiesConverter schemaAndTablePropertiesConverter,
+ PartitionConverter partitionConverter) {
+ return new GravitinoJdbcCatalogFlink22(
+ context, defaultDatabase, schemaAndTablePropertiesConverter,
partitionConverter);
+ }
+}
diff --git
a/flink-connector/v2.2/flink/src/main/java/org/apache/gravitino/flink/connector/jdbc/postgresql/GravitinoPostgresJdbcCatalogFactoryFlink22.java
b/flink-connector/v2.2/flink/src/main/java/org/apache/gravitino/flink/connector/jdbc/postgresql/GravitinoPostgresJdbcCatalogFactoryFlink22.java
new file mode 100644
index 0000000000..e2eb7e9f53
--- /dev/null
+++
b/flink-connector/v2.2/flink/src/main/java/org/apache/gravitino/flink/connector/jdbc/postgresql/GravitinoPostgresJdbcCatalogFactoryFlink22.java
@@ -0,0 +1,42 @@
+/*
+ * 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.GravitinoJdbcCatalogFlink22;
+
+/** {@link GravitinoPostgresJdbcCatalogFactory} implementation for Flink 2.2.
*/
+public class GravitinoPostgresJdbcCatalogFactoryFlink22
+ extends GravitinoPostgresJdbcCatalogFactory {
+
+ /** {@inheritDoc} */
+ @Override
+ protected Catalog newCatalog(
+ CatalogFactory.Context context,
+ String defaultDatabase,
+ SchemaAndTablePropertiesConverter schemaAndTablePropertiesConverter,
+ PartitionConverter partitionConverter) {
+ return new GravitinoJdbcCatalogFlink22(
+ context, defaultDatabase, schemaAndTablePropertiesConverter,
partitionConverter);
+ }
+}
diff --git
a/flink-connector/v2.2/flink/src/main/java/org/apache/gravitino/flink/connector/paimon/GravitinoPaimonCatalogFactoryFlink22.java
b/flink-connector/v2.2/flink/src/main/java/org/apache/gravitino/flink/connector/paimon/GravitinoPaimonCatalogFactoryFlink22.java
new file mode 100644
index 0000000000..0f25d1b690
--- /dev/null
+++
b/flink-connector/v2.2/flink/src/main/java/org/apache/gravitino/flink/connector/paimon/GravitinoPaimonCatalogFactoryFlink22.java
@@ -0,0 +1,39 @@
+/*
+ * 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 2.2. */
+public class GravitinoPaimonCatalogFactoryFlink22 extends
GravitinoPaimonCatalogFactory {
+
+ /** {@inheritDoc} */
+ @Override
+ protected Catalog newCatalog(
+ Context context,
+ String defaultDatabase,
+ SchemaAndTablePropertiesConverter schemaAndTablePropertiesConverter,
+ PartitionConverter partitionConverter) {
+ return new GravitinoPaimonCatalogFlink22(
+ context, defaultDatabase, schemaAndTablePropertiesConverter,
partitionConverter);
+ }
+}
diff --git
a/flink-connector/v2.2/flink/src/main/java/org/apache/gravitino/flink/connector/paimon/GravitinoPaimonCatalogFlink22.java
b/flink-connector/v2.2/flink/src/main/java/org/apache/gravitino/flink/connector/paimon/GravitinoPaimonCatalogFlink22.java
new file mode 100644
index 0000000000..1655545783
--- /dev/null
+++
b/flink-connector/v2.2/flink/src/main/java/org/apache/gravitino/flink/connector/paimon/GravitinoPaimonCatalogFlink22.java
@@ -0,0 +1,52 @@
+/*
+ * 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.CatalogCompatFlink22;
+
+/** {@link GravitinoPaimonCatalog} implementation for Flink 2.2. */
+public class GravitinoPaimonCatalogFlink22 extends GravitinoPaimonCatalog {
+
+ /**
+ * Creates a catalog backed by the Flink 2.2 connector.
+ *
+ * @param context the catalog factory context
+ * @param defaultDatabase the default database name
+ * @param schemaAndTablePropertiesConverter the converter for schemas and
table properties
+ * @param partitionConverter the converter for partition specifications
+ */
+ protected GravitinoPaimonCatalogFlink22(
+ CatalogFactory.Context context,
+ String defaultDatabase,
+ SchemaAndTablePropertiesConverter schemaAndTablePropertiesConverter,
+ PartitionConverter partitionConverter) {
+ super(context, defaultDatabase, schemaAndTablePropertiesConverter,
partitionConverter);
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ protected CatalogCompat catalogCompat() {
+ return CatalogCompatFlink22.INSTANCE;
+ }
+}
diff --git
a/flink-connector/v2.2/flink/src/main/java/org/apache/gravitino/flink/connector/utils/CatalogCompatFlink22.java
b/flink-connector/v2.2/flink/src/main/java/org/apache/gravitino/flink/connector/utils/CatalogCompatFlink22.java
new file mode 100644
index 0000000000..893783eaf8
--- /dev/null
+++
b/flink-connector/v2.2/flink/src/main/java/org/apache/gravitino/flink/connector/utils/CatalogCompatFlink22.java
@@ -0,0 +1,64 @@
+/*
+ * 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.connector.jdbc.core.table.JdbcDynamicTableFactory;
+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;
+import org.apache.flink.table.expressions.DefaultSqlFactory;
+import org.apache.flink.table.factories.Factory;
+
+/** {@link CatalogCompat} implementation for Flink 2.2. */
+public enum CatalogCompatFlink22 implements CatalogCompat {
+ /** Singleton compatibility implementation for Flink 2.2. */
+ INSTANCE;
+
+ /** {@inheritDoc} */
+ @Override
+ public CatalogTable createCatalogTable(
+ Schema schema, String comment, List<String> partitionKeys, Map<String,
String> options) {
+ // CatalogTable.of(...) was removed in Flink 2.x (deprecated as of Flink
1.20); use the
+ // builder API instead.
+ return CatalogTable.newBuilder()
+ .schema(schema)
+ .comment(comment)
+ .partitionKeys(partitionKeys)
+ .options(options)
+ .build();
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public Map<String, String> serializeCatalogTable(ResolvedCatalogTable
resolvedTable) {
+ // Flink 2.x requires a SqlFactory parameter for serializing
computed-column
+ // expressions; DefaultSqlFactory is the standard built-in implementation.
+ return CatalogPropertiesUtil.serializeCatalogTable(resolvedTable,
DefaultSqlFactory.INSTANCE);
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public Factory jdbcDynamicTableFactory() {
+ return new JdbcDynamicTableFactory();
+ }
+}
diff --git
a/flink-connector/v2.2/flink/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory
b/flink-connector/v2.2/flink/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory
new file mode 100644
index 0000000000..0fd9b0b5e7
--- /dev/null
+++
b/flink-connector/v2.2/flink/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory
@@ -0,0 +1,29 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+# Hive catalog support is intentionally omitted for Flink 2.2: no
flink-connector-hive release
+# targets the Flink 2.x line yet (tracked upstream as FLINK-38031), and the
newest available
+# release (built against 1.20) fails at runtime on Flink 2.x because it
references
+# org.apache.flink.api.java.hadoop.mapred.utils.HadoopUtils, a class removed
along with the
+# DataSet API. Revisit once flink-connector-hive ships a Flink 2.x-compatible
release.
+org.apache.gravitino.flink.connector.paimon.GravitinoPaimonCatalogFactoryFlink22
+org.apache.gravitino.flink.connector.iceberg.GravitinoIcebergCatalogFactoryFlink22
+org.apache.gravitino.flink.connector.jdbc.mysql.GravitinoMysqlJdbcCatalogFactoryFlink22
+org.apache.gravitino.flink.connector.jdbc.postgresql.GravitinoPostgresJdbcCatalogFactoryFlink22
+org.apache.gravitino.flink.connector.store.GravitinoCatalogStoreFactory
diff --git
a/flink-connector/v2.2/flink/src/test/java/org/apache/gravitino/flink/connector/catalog/TestBaseCatalogFlink22.java
b/flink-connector/v2.2/flink/src/test/java/org/apache/gravitino/flink/connector/catalog/TestBaseCatalogFlink22.java
new file mode 100644
index 0000000000..adfc181df9
--- /dev/null
+++
b/flink-connector/v2.2/flink/src/test/java/org/apache/gravitino/flink/connector/catalog/TestBaseCatalogFlink22.java
@@ -0,0 +1,31 @@
+/*
+ * 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.catalog;
+
+import org.apache.gravitino.flink.connector.utils.CatalogCompat;
+import org.apache.gravitino.flink.connector.utils.CatalogCompatFlink22;
+
+public class TestBaseCatalogFlink22 extends TestBaseCatalog {
+
+ @Override
+ protected CatalogCompat catalogCompat() {
+ return CatalogCompatFlink22.INSTANCE;
+ }
+}
diff --git
a/flink-connector/v2.2/flink/src/test/java/org/apache/gravitino/flink/connector/hive/TestFlinkGenericTableUtilFlink22.java
b/flink-connector/v2.2/flink/src/test/java/org/apache/gravitino/flink/connector/hive/TestFlinkGenericTableUtilFlink22.java
new file mode 100644
index 0000000000..e14b254496
--- /dev/null
+++
b/flink-connector/v2.2/flink/src/test/java/org/apache/gravitino/flink/connector/hive/TestFlinkGenericTableUtilFlink22.java
@@ -0,0 +1,31 @@
+/*
+ * 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 org.apache.gravitino.flink.connector.utils.CatalogCompat;
+import org.apache.gravitino.flink.connector.utils.CatalogCompatFlink22;
+
+public class TestFlinkGenericTableUtilFlink22 extends
TestFlinkGenericTableUtil {
+
+ @Override
+ protected CatalogCompat catalogCompat() {
+ return CatalogCompatFlink22.INSTANCE;
+ }
+}
diff --git
a/flink-connector/v2.2/flink/src/test/java/org/apache/gravitino/flink/connector/hive/TestGravitinoHiveCatalogFlink22.java
b/flink-connector/v2.2/flink/src/test/java/org/apache/gravitino/flink/connector/hive/TestGravitinoHiveCatalogFlink22.java
new file mode 100644
index 0000000000..b410fda376
--- /dev/null
+++
b/flink-connector/v2.2/flink/src/test/java/org/apache/gravitino/flink/connector/hive/TestGravitinoHiveCatalogFlink22.java
@@ -0,0 +1,31 @@
+/*
+ * 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 org.apache.gravitino.flink.connector.utils.CatalogCompat;
+import org.apache.gravitino.flink.connector.utils.CatalogCompatFlink22;
+
+public class TestGravitinoHiveCatalogFlink22 extends TestGravitinoHiveCatalog {
+
+ @Override
+ protected CatalogCompat catalogCompat() {
+ return CatalogCompatFlink22.INSTANCE;
+ }
+}
diff --git
a/flink-connector/v2.2/flink/src/test/java/org/apache/gravitino/flink/connector/iceberg/TestGravitinoIcebergCatalogFlink22.java
b/flink-connector/v2.2/flink/src/test/java/org/apache/gravitino/flink/connector/iceberg/TestGravitinoIcebergCatalogFlink22.java
new file mode 100644
index 0000000000..34e4515c22
--- /dev/null
+++
b/flink-connector/v2.2/flink/src/test/java/org/apache/gravitino/flink/connector/iceberg/TestGravitinoIcebergCatalogFlink22.java
@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.gravitino.flink.connector.iceberg;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+
+import com.google.common.collect.ImmutableMap;
+import java.nio.file.Path;
+import java.util.List;
+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.iceberg.flink.FlinkCatalog;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+/**
+ * Verifies that {@link
GravitinoIcebergCatalogFlink22#createInnerIcebergCatalog} correctly threads
+ * the given catalog name and properties into the anonymous {@code
CatalogFactory.Context} shim
+ * required to work around Flink 2.x removing {@code FlinkCatalogFactory}'s
2-arg {@code
+ * createCatalog(String, Map)} overload.
+ */
+class TestGravitinoIcebergCatalogFlink22 {
+
+ @TempDir private Path warehouse;
+
+ @Test
+ void createInnerIcebergCatalogThreadsNameAndProperties() throws Exception {
+ Map<String, String> icebergProperties =
+ ImmutableMap.of(
+ IcebergPropertiesConstants.ICEBERG_CATALOG_TYPE,
+ "hadoop",
+ IcebergPropertiesConstants.ICEBERG_CATALOG_WAREHOUSE,
+ "file://" + warehouse);
+
+ GravitinoIcebergCatalogFlink22 catalog =
+ new GravitinoIcebergCatalogFlink22(
+ "my_catalog",
+ "default",
+ mock(SchemaAndTablePropertiesConverter.class),
+ mock(PartitionConverter.class),
+ icebergProperties,
+ icebergProperties);
+
+ Object inner = catalog.createInnerIcebergCatalog("my_catalog",
icebergProperties);
+
+ assertInstanceOf(FlinkCatalog.class, inner);
+ AbstractCatalog innerCatalog = (AbstractCatalog) inner;
+ // The catalog name was threaded through Context.getName().
+ assertEquals("my_catalog", innerCatalog.getName());
+
+ // The warehouse property was threaded through Context.getOptions():
opening the catalog and
+ // listing databases succeeds against the temp warehouse rather than
failing with a
+ // missing-warehouse / uninitialized-catalog error.
+ innerCatalog.open();
+ try {
+ List<String> databases = innerCatalog.listDatabases();
+ assertTrue(databases.isEmpty());
+ } finally {
+ innerCatalog.close();
+ }
+ }
+}
diff --git
a/flink-connector/v2.2/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/catalog/GravitinoCatalogManagerIT22.java
b/flink-connector/v2.2/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/catalog/GravitinoCatalogManagerIT22.java
new file mode 100644
index 0000000000..97f5711873
--- /dev/null
+++
b/flink-connector/v2.2/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/catalog/GravitinoCatalogManagerIT22.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 GravitinoCatalogManagerIT22 extends GravitinoCatalogManagerIT {}
diff --git
a/flink-connector/v2.2/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/iceberg/FlinkIcebergHiveCatalogIT22.java
b/flink-connector/v2.2/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/iceberg/FlinkIcebergHiveCatalogIT22.java
new file mode 100644
index 0000000000..0a8c42e79c
--- /dev/null
+++
b/flink-connector/v2.2/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/iceberg/FlinkIcebergHiveCatalogIT22.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 FlinkIcebergHiveCatalogIT22 extends FlinkIcebergHiveCatalogIT {}
diff --git
a/flink-connector/v2.2/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/iceberg/FlinkIcebergJdbcCatalogIT22.java
b/flink-connector/v2.2/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/iceberg/FlinkIcebergJdbcCatalogIT22.java
new file mode 100644
index 0000000000..b0b9bebc8f
--- /dev/null
+++
b/flink-connector/v2.2/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/iceberg/FlinkIcebergJdbcCatalogIT22.java
@@ -0,0 +1,29 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.gravitino.flink.connector.integration.test.iceberg;
+
+import org.junit.jupiter.api.condition.DisabledIf;
+
+// Even though Flink 2.2 and lakehouse-iceberg both use Iceberg 1.11, a JDBC
backend exercises the
+// server-side Iceberg JdbcCatalog, which hits a cross-classloader
IllegalAccessError when the
+// embedded MiniGravitino server shares the JVM with the Flink Iceberg
runtime. Run in deploy mode
+// only. @DisabledIf is not @Inherited, so each concrete subclass must declare
it explicitly.
+@DisabledIf("org.apache.gravitino.integration.test.util.ITUtils#isEmbedded")
+public class FlinkIcebergJdbcCatalogIT22 extends FlinkIcebergJdbcCatalogIT {}
diff --git
a/flink-connector/v2.2/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/iceberg/FlinkIcebergRestCatalogIT22.java
b/flink-connector/v2.2/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/iceberg/FlinkIcebergRestCatalogIT22.java
new file mode 100644
index 0000000000..feb825ac39
--- /dev/null
+++
b/flink-connector/v2.2/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/iceberg/FlinkIcebergRestCatalogIT22.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 FlinkIcebergRestCatalogIT22 extends FlinkIcebergRestCatalogIT {}
diff --git
a/flink-connector/v2.2/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/jdbc/FlinkJdbcMysqlCatalogIT22.java
b/flink-connector/v2.2/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/jdbc/FlinkJdbcMysqlCatalogIT22.java
new file mode 100644
index 0000000000..010af20fde
--- /dev/null
+++
b/flink-connector/v2.2/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/jdbc/FlinkJdbcMysqlCatalogIT22.java
@@ -0,0 +1,53 @@
+/*
+ * 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.apache.flink.table.api.ValidationException;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class FlinkJdbcMysqlCatalogIT22 extends FlinkJdbcMysqlCatalogIT {
+
+ @Override
+ @Test
+ public void testCreateGravitinoJdbcCatalogUsingSQLMissingOptions() {
+ // Flink 2.2's JdbcCatalogFactory reports missing required catalog options
through Flink's
+ // own ValidationException (via FactoryUtil option validation) rather than
the
+ // IllegalArgumentException thrown by older minor versions.
+ tableEnv.useCatalog(DEFAULT_CATALOG);
+ String catalogName = "gravitino_mysql_jdbc_catalog_missing_options";
+ ValidationException exception =
+ Assertions.assertThrows(
+ ValidationException.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/v2.2/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/paimon/FlinkPaimonHiveBackendIT22.java
b/flink-connector/v2.2/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/paimon/FlinkPaimonHiveBackendIT22.java
new file mode 100644
index 0000000000..18a92be60c
--- /dev/null
+++
b/flink-connector/v2.2/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/paimon/FlinkPaimonHiveBackendIT22.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 FlinkPaimonHiveBackendIT22 extends FlinkPaimonHiveBackendIT {}
diff --git
a/flink-connector/v2.2/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/paimon/FlinkPaimonJdbcBackendIT22.java
b/flink-connector/v2.2/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/paimon/FlinkPaimonJdbcBackendIT22.java
new file mode 100644
index 0000000000..e672250402
--- /dev/null
+++
b/flink-connector/v2.2/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/paimon/FlinkPaimonJdbcBackendIT22.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 FlinkPaimonJdbcBackendIT22 extends FlinkPaimonJdbcBackendIT {}
diff --git
a/flink-connector/v2.2/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/paimon/FlinkPaimonLocalFileSystemBackendIT22.java
b/flink-connector/v2.2/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/paimon/FlinkPaimonLocalFileSystemBackendIT22.java
new file mode 100644
index 0000000000..89e0903a3d
--- /dev/null
+++
b/flink-connector/v2.2/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/paimon/FlinkPaimonLocalFileSystemBackendIT22.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 FlinkPaimonLocalFileSystemBackendIT22 extends
FlinkPaimonLocalFileSystemBackendIT {}
diff --git
a/flink-connector/v2.2/flink/src/test/java/org/apache/gravitino/flink/connector/jdbc/TestGravitinoJdbcCatalogFlink22.java
b/flink-connector/v2.2/flink/src/test/java/org/apache/gravitino/flink/connector/jdbc/TestGravitinoJdbcCatalogFlink22.java
new file mode 100644
index 0000000000..e04bfba386
--- /dev/null
+++
b/flink-connector/v2.2/flink/src/test/java/org/apache/gravitino/flink/connector/jdbc/TestGravitinoJdbcCatalogFlink22.java
@@ -0,0 +1,31 @@
+/*
+ * 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 org.apache.gravitino.flink.connector.utils.CatalogCompat;
+import org.apache.gravitino.flink.connector.utils.CatalogCompatFlink22;
+
+public class TestGravitinoJdbcCatalogFlink22 extends TestGravitinoJdbcCatalog {
+
+ @Override
+ protected CatalogCompat catalogCompat() {
+ return CatalogCompatFlink22.INSTANCE;
+ }
+}
diff --git
a/flink-connector/v2.2/flink/src/test/java/org/apache/gravitino/flink/connector/paimon/TestGravitinoPaimonCatalogFlink22.java
b/flink-connector/v2.2/flink/src/test/java/org/apache/gravitino/flink/connector/paimon/TestGravitinoPaimonCatalogFlink22.java
new file mode 100644
index 0000000000..6bfbc973c8
--- /dev/null
+++
b/flink-connector/v2.2/flink/src/test/java/org/apache/gravitino/flink/connector/paimon/TestGravitinoPaimonCatalogFlink22.java
@@ -0,0 +1,31 @@
+/*
+ * 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.gravitino.flink.connector.utils.CatalogCompat;
+import org.apache.gravitino.flink.connector.utils.CatalogCompatFlink22;
+
+public class TestGravitinoPaimonCatalogFlink22 extends
TestGravitinoPaimonCatalog {
+
+ @Override
+ protected CatalogCompat catalogCompat() {
+ return CatalogCompatFlink22.INSTANCE;
+ }
+}
diff --git
a/flink-connector/v2.2/flink/src/test/java/org/apache/gravitino/flink/connector/utils/TestCatalogCompatFlink22.java
b/flink-connector/v2.2/flink/src/test/java/org/apache/gravitino/flink/connector/utils/TestCatalogCompatFlink22.java
new file mode 100644
index 0000000000..b30215d848
--- /dev/null
+++
b/flink-connector/v2.2/flink/src/test/java/org/apache/gravitino/flink/connector/utils/TestCatalogCompatFlink22.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.connector.utils;
+
+import java.util.Collections;
+import java.util.Map;
+import org.apache.flink.connector.jdbc.core.table.JdbcDynamicTableFactory;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.Column;
+import org.apache.flink.table.catalog.ResolvedCatalogTable;
+import org.apache.flink.table.catalog.ResolvedSchema;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class TestCatalogCompatFlink22 {
+
+ @Test
+ public void testCreateCatalogTable() {
+ CatalogTable catalogTable =
+ CatalogCompatFlink22.INSTANCE.createCatalogTable(
+ Schema.newBuilder().column("id", DataTypes.INT()).build(),
+ "comment",
+ Collections.singletonList("id"),
+ Collections.singletonMap("key", "value"));
+
+ Assertions.assertEquals("comment", catalogTable.getComment());
+ Assertions.assertEquals(Collections.singletonList("id"),
catalogTable.getPartitionKeys());
+ Assertions.assertEquals("value", catalogTable.getOptions().get("key"));
+ Assertions.assertEquals(1,
catalogTable.getUnresolvedSchema().getColumns().size());
+ }
+
+ @Test
+ @SuppressWarnings("deprecation")
+ public void testSerializeCatalogTable() {
+ CatalogTable catalogTable =
+ CatalogCompatFlink22.INSTANCE.createCatalogTable(
+ Schema.newBuilder().column("id", DataTypes.INT()).build(),
+ "comment",
+ Collections.emptyList(),
+ Collections.singletonMap("key", "value"));
+ ResolvedCatalogTable resolvedCatalogTable =
+ new ResolvedCatalogTable(
+ catalogTable,
+ new ResolvedSchema(
+ Collections.singletonList(Column.physical("id",
DataTypes.INT())),
+ Collections.emptyList(),
+ null));
+
+ Map<String, String> serialized =
+
CatalogCompatFlink22.INSTANCE.serializeCatalogTable(resolvedCatalogTable);
+
+ Assertions.assertEquals("id", serialized.get("schema.0.name"));
+ Assertions.assertEquals("value", serialized.get("key"));
+ }
+
+ @Test
+ public void testJdbcDynamicTableFactory() {
+ Assertions.assertInstanceOf(
+ JdbcDynamicTableFactory.class,
CatalogCompatFlink22.INSTANCE.jdbcDynamicTableFactory());
+ }
+}
diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml
index cb827cdacb..9b2f9a9168 100644
--- a/gradle/libs.versions.toml
+++ b/gradle/libs.versions.toml
@@ -119,6 +119,7 @@ jodd = "3.5.2"
flink119 = "1.19.0"
flink120 = "1.20.0"
flink21 = "2.1.3"
+flink22 = "2.2.1"
flinkjdbc119 = "3.3.0-1.19"
flinkjdbc120 = "3.3.0-1.20"
# flink-connector-jdbc was restructured starting with the 4.x line: the
monolithic
@@ -126,20 +127,23 @@ flinkjdbc120 = "3.3.0-1.20"
# database (flink-connector-jdbc-mysql, flink-connector-jdbc-postgres, ...),
and this split line
# is what publishes real Flink 2.x builds.
flinkjdbc21 = "4.1.0-2.1"
+flinkjdbc22 = "4.1.0-2.2"
# flink-connector-hive has not published a build targeting the Flink 2.x line
yet either;
# reuse the newest externalized release (built against 1.20) until an official
2.x build ships.
flinkHive21 = "3.0.0-1.20"
-# flink-connector-jdbc-core (Flink 2.1) transitively pulls
openlineage-sql-java:1.32.0, whose
+# flink-connector-jdbc-core (Flink 2.x) transitively pulls
openlineage-sql-java:1.32.0, whose
# bundled, unrelocated org.apache.commons.lang3.SystemProperties predates the
getUserName(String)
# overload and shadows the real commons-lang3 on the classpath. Force this
newer release, which
# ships an up-to-date bundled copy.
-openlineageSqlJava21 = "1.52.0"
+openlineageSqlJava = "1.52.0"
iceberg4flink119 = "1.10.2"
iceberg4flink120 = "1.11.0"
iceberg4flink21 = "1.11.0"
+iceberg4flink22 = "1.12.0"
paimon4flink119 = "1.2.0"
paimon4flink120 = "1.2.0"
paimon4flink21 = "1.4.1"
+paimon4flink22 = "1.4.1"
# 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"
@@ -383,8 +387,11 @@ flinkjdbc120 =
{group='org.apache.flink',name='flink-connector-jdbc', version.re
flinkjdbc21-core = {group='org.apache.flink',name='flink-connector-jdbc-core',
version.ref='flinkjdbc21'}
flinkjdbc21-mysql =
{group='org.apache.flink',name='flink-connector-jdbc-mysql',
version.ref='flinkjdbc21'}
flinkjdbc21-postgres =
{group='org.apache.flink',name='flink-connector-jdbc-postgres',
version.ref='flinkjdbc21'}
+flinkjdbc22-core = {group='org.apache.flink',name='flink-connector-jdbc-core',
version.ref='flinkjdbc22'}
+flinkjdbc22-mysql =
{group='org.apache.flink',name='flink-connector-jdbc-mysql',
version.ref='flinkjdbc22'}
+flinkjdbc22-postgres =
{group='org.apache.flink',name='flink-connector-jdbc-postgres',
version.ref='flinkjdbc22'}
flinkHive21 = {group='org.apache.flink',name='flink-connector-hive_2.12',
version.ref='flinkHive21'}
-openlineageSqlJava21 = {group='io.openlineage',name='openlineage-sql-java',
version.ref='openlineageSqlJava21'}
+openlineageSqlJava = {group='io.openlineage',name='openlineage-sql-java',
version.ref='openlineageSqlJava'}
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 edc3da371e..511cdd025d 100644
--- a/settings.gradle.kts
+++ b/settings.gradle.kts
@@ -100,11 +100,16 @@ if (scalaVersion == "2.12") {
project(":flink-connector:flink-runtime-1.20").projectDir =
file("flink-connector/v1.20/flink-runtime")
}
-// Flink 2.x removed the Scala APIs entirely, so the flink-2.1 modules are not
gated by scalaVersion.
+// These connectors use Flink Java APIs, so they do not need scalaVersion
gating.
+// Flink internals still use Scala.
include("flink-connector:flink-2.1", "flink-connector:flink-runtime-2.1")
project(":flink-connector:flink-2.1").projectDir =
file("flink-connector/v2.1/flink")
project(":flink-connector:flink-runtime-2.1").projectDir =
file("flink-connector/v2.1/flink-runtime")
+include("flink-connector:flink-2.2", "flink-connector:flink-runtime-2.2")
+project(":flink-connector:flink-2.2").projectDir =
file("flink-connector/v2.2/flink")
+project(":flink-connector:flink-runtime-2.2").projectDir =
+ file("flink-connector/v2.2/flink-runtime")
include("spark-connector:spark-3.5", "spark-connector:spark-runtime-3.5")
project(":spark-connector:spark-3.5").projectDir =
file("spark-connector/v3.5/spark")
project(":spark-connector:spark-runtime-3.5").projectDir =
file("spark-connector/v3.5/spark-runtime")
