You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by mx...@apache.org on 2026/09/30 10:04:06 UTC
(iceberg) 04/04: Flink: Re-apply v2.3 changes
This is an automated email from the ASF dual-hosted git repository.
mxm pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
commit 9daf2df445af5eced1f37b7b9e697669070caa4d
Author: Maximilian Michels <mx...@apache.org>
AuthorDate: Wed Sep 30 11:13:28 2026 +0200
Flink: Re-apply v2.3 changes
---
flink/v2.3/build.gradle | 34 +++++++++++-----------
.../iceberg/flink/TestFlinkCatalogTable.java | 8 ++---
.../maintenance/operator/TestMonitorSource.java | 2 ++
.../iceberg/flink/util/TestFlinkPackage.java | 2 +-
4 files changed, 24 insertions(+), 22 deletions(-)
diff --git a/flink/v2.3/build.gradle b/flink/v2.3/build.gradle
index a2b21227ed..d3b432152c 100644
--- a/flink/v2.3/build.gradle
+++ b/flink/v2.3/build.gradle
@@ -17,7 +17,7 @@
* under the License.
*/
-String flinkMajorVersion = '2.1'
+String flinkMajorVersion = '2.3'
String scalaVersion = System.getProperty("scalaVersion") != null ? System.getProperty("scalaVersion") : System.getProperty("defaultScalaVersion")
project(":iceberg-flink:iceberg-flink-${flinkMajorVersion}") {
@@ -32,16 +32,16 @@ project(":iceberg-flink:iceberg-flink-${flinkMajorVersion}") {
implementation project(':iceberg-parquet')
implementation project(':iceberg-hive-metastore')
- compileOnly libs.flink21.avro
+ compileOnly libs.flink23.avro
compileOnly libs.joda.time
// dropwizard histogram metrics (optional in Flink)
- compileOnly libs.flink21.metrics.dropwizard
- compileOnly libs.flink21.streaming.java
- compileOnly "${libs.flink21.streaming.java.get().module}:${libs.flink21.streaming.java.get().getVersion()}:tests"
- compileOnly libs.flink21.table.api.java.bridge
- compileOnly "org.apache.flink:flink-table-planner_${scalaVersion}:${libs.versions.flink21.get()}"
- compileOnly libs.flink21.connector.base
- compileOnly libs.flink21.connector.files
+ compileOnly libs.flink23.metrics.dropwizard
+ compileOnly libs.flink23.streaming.java
+ compileOnly "${libs.flink23.streaming.java.get().module}:${libs.flink23.streaming.java.get().getVersion()}:tests"
+ compileOnly libs.flink23.table.api.java.bridge
+ compileOnly "org.apache.flink:flink-table-planner_${scalaVersion}:${libs.versions.flink23.get()}"
+ compileOnly libs.flink23.connector.base
+ compileOnly libs.flink23.connector.files
compileOnly libs.hadoop3.hdfs
compileOnly libs.hadoop3.common
@@ -70,13 +70,13 @@ project(":iceberg-flink:iceberg-flink-${flinkMajorVersion}") {
implementation libs.datasketches
implementation libs.roaringbitmap
- testImplementation libs.flink21.connector.test.utils
- testImplementation libs.flink21.core
- testImplementation libs.flink21.runtime
- testImplementation(libs.flink21.test.utilsjunit) {
+ testImplementation libs.flink23.connector.test.utils
+ testImplementation libs.flink23.core
+ testImplementation libs.flink23.runtime
+ testImplementation(libs.flink23.test.utilsjunit) {
exclude group: 'junit'
}
- testImplementation(libs.flink21.test.utils) {
+ testImplementation(libs.flink23.test.utils) {
exclude group: "org.apache.curator", module: 'curator-test'
exclude group: 'junit'
}
@@ -181,13 +181,13 @@ project(":iceberg-flink:iceberg-flink-runtime-${flinkMajorVersion}") {
integrationImplementation project(path: ":iceberg-flink:iceberg-flink-${flinkMajorVersion}", configuration: "testArtifacts")
integrationImplementation project(path: ':iceberg-api', configuration: 'testArtifacts')
integrationImplementation project(path: ':iceberg-hive-metastore', configuration: 'testArtifacts')
- integrationImplementation(libs.flink21.test.utils) {
+ integrationImplementation(libs.flink23.test.utils) {
exclude group: "org.apache.curator", module: 'curator-test'
exclude group: 'junit'
}
- integrationImplementation libs.flink21.table.api.java.bridge
- integrationImplementation "org.apache.flink:flink-table-planner_${scalaVersion}:${libs.versions.flink21.get()}"
+ integrationImplementation libs.flink23.table.api.java.bridge
+ integrationImplementation "org.apache.flink:flink-table-planner_${scalaVersion}:${libs.versions.flink23.get()}"
integrationImplementation libs.hadoop3.common
integrationImplementation libs.hadoop3.hdfs
diff --git a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java
index 30daf6d55e..aea9cda11d 100644
--- a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java
+++ b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java
@@ -417,7 +417,7 @@ public class TestFlinkCatalogTable extends CatalogTestBase {
// Adding an existing field should fail due to Flink's internal validation.
assertThatThrownBy(() -> sql("ALTER TABLE tl ADD (id STRING)"))
.isInstanceOf(ValidationException.class)
- .hasMessageContaining("Try to add a column `id` which already exists in the table.");
+ .hasMessageContaining("Column `id` already exists in the table.");
}
@TestTemplate
@@ -585,8 +585,7 @@ public class TestFlinkCatalogTable extends CatalogTestBase {
// validation.
assertThatThrownBy(() -> sql("ALTER TABLE tl MODIFY (non_existing STRING FIRST)"))
.isInstanceOf(ValidationException.class)
- .hasMessageContaining(
- "Try to modify a column `non_existing` which does not exist in the table.");
+ .hasMessageContaining("Column `non_existing` does not exist in the table.");
// Moving a column after a non-existing column should fail due to Flink's internal validation.
assertThatThrownBy(() -> sql("ALTER TABLE tl MODIFY (dt STRING AFTER non_existing)"))
@@ -755,7 +754,8 @@ public class TestFlinkCatalogTable extends CatalogTestBase {
org.apache.flink.table.api.Schema.newBuilder()
.column("id", DataTypes.BIGINT())
.build())
- .definitionQuery("SELECT id FROM tl")
+ .originalQuery("SELECT id FROM tl")
+ .expandedQuery("SELECT id FROM tl")
.freshness(IntervalFreshness.ofMinute("5"))
.logicalRefreshMode(CatalogMaterializedTable.LogicalRefreshMode.AUTOMATIC)
.refreshMode(CatalogMaterializedTable.RefreshMode.CONTINUOUS)
diff --git a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestMonitorSource.java b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestMonitorSource.java
index 3dca6c421c..04bac0cee7 100644
--- a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestMonitorSource.java
+++ b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestMonitorSource.java
@@ -226,6 +226,8 @@ class TestMonitorSource extends OperatorTestBase {
// Make sure that the job with restored source does not read new records from the table
StreamGraph streamGraph = env.getStreamGraph();
+ // Submitting the graph directly skips the executor, which is what normally assigns the job id
+ streamGraph.setJobId(JobID.generate());
streamGraph.setSavepointRestoreSettings(SavepointRestoreSettings.forPath(savepointPath));
CompletableFuture<JobID> jobIDCompletableFuture = clusterClient.submitJob(streamGraph);
try {
diff --git a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/util/TestFlinkPackage.java b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/util/TestFlinkPackage.java
index 8f765ff21f..918c6466a3 100644
--- a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/util/TestFlinkPackage.java
+++ b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/util/TestFlinkPackage.java
@@ -29,7 +29,7 @@ public class TestFlinkPackage {
/** This unit test would need to be adjusted as new Flink version is supported. */
@Test
public void testVersion() {
- assertThat(FlinkPackage.version()).isEqualTo("2.1.0");
+ assertThat(FlinkPackage.version()).isEqualTo("2.3.0");
}
@Test