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 <[email protected]> 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
