Flink: Add support for Flink 2.0

Author: mxm

.github/workflows/flink-ci.yml

@@ -73,7 +73,7 @@ jobs:
     strategy:
       matrix:
         jvm: [11, 17, 21]
-        flink: ['1.18', '1.19', '1.20']
+        flink: ['1.19', '1.20', '2.0']
     env:
       SPARK_LOCAL_IP: localhost
     steps:

dev/stage-binaries.sh

@@ -19,7 +19,7 @@
 #
 
 SCALA_VERSION=2.12
-FLINK_VERSIONS=1.18,1.19,1.20
+FLINK_VERSIONS=1.19,1.20,2.0
 SPARK_VERSIONS=3.4,3.5
 KAFKA_VERSIONS=3
 

flink/build.gradle

@@ -29,4 +29,8 @@ if (flinkVersions.contains("1.19")) {
 
 if (flinkVersions.contains("1.20")) {
   apply from: file("$projectDir/v1.20/build.gradle")
+}
+
+if (flinkVersions.contains("2.0")) {
+  apply from: file("$projectDir/v2.0/build.gradle")
 }

flink/v2.0/build.gradle

@@ -17,7 +17,7 @@
  * under the License.
  */
 
-String flinkMajorVersion = '1.20'
+String flinkMajorVersion = '2.0'
 String scalaVersion = System.getProperty("scalaVersion") != null ? System.getProperty("scalaVersion") : System.getProperty("defaultScalaVersion")
 
 project(":iceberg-flink:iceberg-flink-${flinkMajorVersion}") {
@@ -32,15 +32,15 @@ project(":iceberg-flink:iceberg-flink-${flinkMajorVersion}") {
     implementation project(':iceberg-parquet')
     implementation project(':iceberg-hive-metastore')
 
-    compileOnly libs.flink120.avro
+    compileOnly libs.flink20.avro
     // for dropwizard histogram metrics implementation
-    compileOnly libs.flink120.metrics.dropwizard
-    compileOnly libs.flink120.streaming.java
-    compileOnly "${libs.flink120.streaming.java.get().module}:${libs.flink120.streaming.java.get().getVersion()}:tests"
-    compileOnly libs.flink120.table.api.java.bridge
-    compileOnly "org.apache.flink:flink-table-planner_${scalaVersion}:${libs.versions.flink120.get()}"
-    compileOnly libs.flink120.connector.base
-    compileOnly libs.flink120.connector.files
+    compileOnly libs.flink20.metrics.dropwizard
+    compileOnly libs.flink20.streaming.java
+    compileOnly "${libs.flink20.streaming.java.get().module}:${libs.flink20.streaming.java.get().getVersion()}:tests"
+    compileOnly libs.flink20.table.api.java.bridge
+    compileOnly "org.apache.flink:flink-table-planner_${scalaVersion}:${libs.versions.flink20.get()}"
+    compileOnly libs.flink20.connector.base
+    compileOnly libs.flink20.connector.files
 
     compileOnly libs.hadoop3.hdfs
     compileOnly libs.hadoop3.common
@@ -68,13 +68,13 @@ project(":iceberg-flink:iceberg-flink-${flinkMajorVersion}") {
 
     implementation libs.datasketches
 
-    testImplementation libs.flink120.connector.test.utils
-    testImplementation libs.flink120.core
-    testImplementation libs.flink120.runtime
-    testImplementation(libs.flink120.test.utilsjunit) {
+    testImplementation libs.flink20.connector.test.utils
+    testImplementation libs.flink20.core
+    testImplementation libs.flink20.runtime
+    testImplementation(libs.flink20.test.utilsjunit) {
       exclude group: 'junit'
     }
-    testImplementation(libs.flink120.test.utils) {
+    testImplementation(libs.flink20.test.utils) {
       exclude group: "org.apache.curator", module: 'curator-test'
       exclude group: 'junit'
     }
@@ -169,7 +169,7 @@ project(":iceberg-flink:iceberg-flink-runtime-${flinkMajorVersion}") {
     }
 
     // for dropwizard histogram metrics implementation
-    implementation libs.flink120.metrics.dropwizard
+    implementation libs.flink20.metrics.dropwizard
 
     // for integration testing with the flink-runtime-jar
     // all of those dependencies are required because the integration test extends FlinkTestBase
@@ -179,13 +179,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.flink120.test.utils) {
+    integrationImplementation(libs.flink20.test.utils) {
       exclude group: "org.apache.curator", module: 'curator-test'
       exclude group: 'junit'
     }
 
-    integrationImplementation libs.flink120.table.api.java.bridge
-    integrationImplementation "org.apache.flink:flink-table-planner_${scalaVersion}:${libs.versions.flink120.get()}"
+    integrationImplementation libs.flink20.table.api.java.bridge
+    integrationImplementation "org.apache.flink:flink-table-planner_${scalaVersion}:${libs.versions.flink20.get()}"
 
     integrationImplementation libs.hadoop3.common
     integrationImplementation libs.hadoop3.hdfs

flink/v2.0/flink/src/main/java/org/apache/flink/table/api/runtime/types/FlinkScalaKryoInstantiator.java

@@ -0,0 +1,26 @@
+/*
+ * 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.flink.table.api.runtime.types;
+
+/**
+ * Override Flink's internal FlinkScalaKryoInstantiator to avoid loading the Scala extensions for
+ * the KryoSerializer. This is a workaround until Kryo-related issues with the Scala extensions are
+ * fixed. See: https://issues.apache.org/jira/browse/FLINK-37546
+ */
+public class FlinkScalaKryoInstantiator {}

flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java

@@ -28,7 +28,6 @@
 import java.util.Set;
 import java.util.stream.Collectors;
 import org.apache.flink.annotation.Internal;
-import org.apache.flink.table.api.TableSchema;
 import org.apache.flink.table.catalog.AbstractCatalog;
 import org.apache.flink.table.catalog.CatalogBaseTable;
 import org.apache.flink.table.catalog.CatalogDatabase;
@@ -37,7 +36,6 @@
 import org.apache.flink.table.catalog.CatalogPartition;
 import org.apache.flink.table.catalog.CatalogPartitionSpec;
 import org.apache.flink.table.catalog.CatalogTable;
-import org.apache.flink.table.catalog.CatalogTableImpl;
 import org.apache.flink.table.catalog.ObjectPath;
 import org.apache.flink.table.catalog.ResolvedCatalogTable;
 import org.apache.flink.table.catalog.TableChange;
@@ -53,6 +51,7 @@
 import org.apache.flink.table.catalog.stats.CatalogTableStatistics;
 import org.apache.flink.table.expressions.Expression;
 import org.apache.flink.table.factories.Factory;
+import org.apache.flink.table.legacy.api.TableSchema;
 import org.apache.flink.util.StringUtils;
 import org.apache.iceberg.CachingCatalog;
 import org.apache.iceberg.DataFile;
@@ -680,7 +679,11 @@ static CatalogTable toCatalogTableWithProps(Table table, Map<String, String> pro
     // CatalogTableImpl to copy a new catalog table.
     // Let's re-loading table from Iceberg catalog when creating source/sink operators.
     // Iceberg does not have Table comment, so pass a null (Default comment value in Flink).
-    return new CatalogTableImpl(schema, partitionKeys, props, null);
+    return CatalogTable.newBuilder()
+        .schema(schema.toSchema())
+        .partitionKeys(partitionKeys)
+        .options(props)
+        .build();
   }
 
   static CatalogTable toCatalogTable(Table table) {

flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalogFactory.java

@@ -21,9 +21,10 @@
 import java.net.URL;
 import java.nio.file.Files;
 import java.nio.file.Paths;
-import java.util.List;
 import java.util.Locale;
 import java.util.Map;
+import java.util.Set;
+import org.apache.flink.configuration.ConfigOption;
 import org.apache.flink.configuration.GlobalConfiguration;
 import org.apache.flink.runtime.util.HadoopUtils;
 import org.apache.flink.table.catalog.Catalog;
@@ -34,8 +35,7 @@
 import org.apache.iceberg.catalog.Namespace;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.base.Strings;
-import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
-import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
 import org.apache.iceberg.util.PropertyUtil;
 
 /**
@@ -60,6 +60,8 @@
  */
 public class FlinkCatalogFactory implements CatalogFactory {
 
+  public static final String FACTORY_IDENTIFIER = "iceberg";
+
   // Can not just use "type", it conflicts with CATALOG_TYPE.
   public static final String ICEBERG_CATALOG_TYPE = "catalog-type";
   public static final String ICEBERG_CATALOG_TYPE_HADOOP = "hadoop";
@@ -72,8 +74,6 @@ public class FlinkCatalogFactory implements CatalogFactory {
   public static final String DEFAULT_DATABASE_NAME = "default";
   public static final String DEFAULT_CATALOG_NAME = "default_catalog";
   public static final String BASE_NAMESPACE = "base-namespace";
-  public static final String TYPE = "type";
-  public static final String PROPERTY_VERSION = "property-version";
 
   /**
    * Create an Iceberg {@link org.apache.iceberg.catalog.Catalog} loader to be used by this Flink
@@ -122,21 +122,23 @@ static CatalogLoader createCatalogLoader(
   }
 
   @Override
-  public Map<String, String> requiredContext() {
-    Map<String, String> context = Maps.newHashMap();
-    context.put(TYPE, "iceberg");
-    context.put(PROPERTY_VERSION, "1");
-    return context;
+  public String factoryIdentifier() {
+    return FACTORY_IDENTIFIER;
+  }
+
+  @Override
+  public Set<ConfigOption<?>> requiredOptions() {
+    return ImmutableSet.<ConfigOption<?>>builder().build();
   }
 
   @Override
-  public List<String> supportedProperties() {
-    return ImmutableList.of("*");
+  public Set<ConfigOption<?>> optionalOptions() {
+    return ImmutableSet.<ConfigOption<?>>builder().build();
   }
 
   @Override
-  public Catalog createCatalog(String name, Map<String, String> properties) {
-    return createCatalog(name, properties, clusterHadoopConf());
+  public Catalog createCatalog(Context context) {
+    return createCatalog(context.getName(), context.getOptions(), clusterHadoopConf());
   }
 
   protected Catalog createCatalog(

flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java

@@ -23,7 +23,6 @@
 import java.util.Set;
 import org.apache.flink.configuration.ConfigOption;
 import org.apache.flink.configuration.Configuration;
-import org.apache.flink.table.api.TableSchema;
 import org.apache.flink.table.catalog.CatalogDatabaseImpl;
 import org.apache.flink.table.catalog.ObjectIdentifier;
 import org.apache.flink.table.catalog.ObjectPath;
@@ -34,6 +33,7 @@
 import org.apache.flink.table.connector.source.DynamicTableSource;
 import org.apache.flink.table.factories.DynamicTableSinkFactory;
 import org.apache.flink.table.factories.DynamicTableSourceFactory;
+import org.apache.flink.table.legacy.api.TableSchema;
 import org.apache.flink.table.utils.TableSchemaUtils;
 import org.apache.flink.util.Preconditions;
 import org.apache.iceberg.catalog.TableIdentifier;
@@ -131,17 +131,16 @@ private static TableLoader createTableLoader(
 
     mergedProps.forEach(flinkConf::setString);
 
-    String catalogName = flinkConf.getString(FlinkCreateTableOptions.CATALOG_NAME);
+    String catalogName = flinkConf.get(FlinkCreateTableOptions.CATALOG_NAME);
     Preconditions.checkNotNull(
         catalogName,
         "Table property '%s' cannot be null",
         FlinkCreateTableOptions.CATALOG_NAME.key());
 
-    String catalogDatabase =
-        flinkConf.getString(FlinkCreateTableOptions.CATALOG_DATABASE, databaseName);
+    String catalogDatabase = flinkConf.get(FlinkCreateTableOptions.CATALOG_DATABASE, databaseName);
     Preconditions.checkNotNull(catalogDatabase, "The iceberg database name cannot be null");
 
-    String catalogTable = flinkConf.getString(FlinkCreateTableOptions.CATALOG_TABLE, tableName);
+    String catalogTable = flinkConf.get(FlinkCreateTableOptions.CATALOG_TABLE, tableName);
     Preconditions.checkNotNull(catalogTable, "The iceberg table name cannot be null");
 
     org.apache.hadoop.conf.Configuration hadoopConf = FlinkCatalogFactory.clusterHadoopConf();

flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/FlinkSchemaUtil.java

@@ -21,9 +21,9 @@
 import java.util.List;
 import java.util.Set;
 import org.apache.flink.table.api.DataTypes;
-import org.apache.flink.table.api.TableSchema;
 import org.apache.flink.table.catalog.Column;
 import org.apache.flink.table.catalog.ResolvedSchema;
+import org.apache.flink.table.legacy.api.TableSchema;
 import org.apache.flink.table.types.logical.LogicalType;
 import org.apache.flink.table.types.logical.RowType;
 import org.apache.flink.table.types.utils.TypeConversions;

flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/IcebergTableSink.java

@@ -23,15 +23,15 @@
 import org.apache.flink.configuration.ReadableConfig;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.datastream.DataStreamSink;
-import org.apache.flink.table.api.TableSchema;
-import org.apache.flink.table.api.constraints.UniqueConstraint;
 import org.apache.flink.table.connector.ChangelogMode;
 import org.apache.flink.table.connector.ProviderContext;
 import org.apache.flink.table.connector.sink.DataStreamSinkProvider;
 import org.apache.flink.table.connector.sink.DynamicTableSink;
 import org.apache.flink.table.connector.sink.abilities.SupportsOverwrite;
 import org.apache.flink.table.connector.sink.abilities.SupportsPartitioning;
 import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.legacy.api.TableSchema;
+import org.apache.flink.table.legacy.api.constraints.UniqueConstraint;
 import org.apache.flink.types.RowKind;
 import org.apache.flink.util.Preconditions;
 import org.apache.iceberg.flink.sink.FlinkSink;