Flink: make FLIP-27 default in SQL and mark the old FlinkSource as deprecated (#11345)

Author: stevenzwu

docs/docs/flink-queries.md

@@ -66,13 +66,17 @@ There are some options that could be set in Flink SQL hint options for streaming
 
 ### FLIP-27 source for SQL
 
-Here are the SQL settings for the [FLIP-27](https://cwiki.apache.org/confluence/display/FLINK/FLIP-27%3A+Refactor+Source+Interface) source. All other SQL settings and options documented above are applicable to the FLIP-27 source.
+Here is the SQL setting to opt in or out of the
+[FLIP-27 source](https://cwiki.apache.org/confluence/display/FLINK/FLIP-27%3A+Refactor+Source+Interface).
 
 ```sql
--- Opt in the FLIP-27 source. Default is false.
-SET table.exec.iceberg.use-flip27-source = true;
+-- Opt out the FLIP-27 source.
+-- Default is false for Flink 1.19 and below, and true for Flink 1.20 and above.
+SET table.exec.iceberg.use-flip27-source = false;
 ```
 
+All other SQL settings and options documented above are applicable to the FLIP-27 source.
+
 ### Reading branches and tags with SQL
 Branch and tags can be read via SQL by specifying options. For more details
 refer to [Flink Configuration](flink-configuration.md#read-options)

flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/source/FlinkSource.java

@@ -28,6 +28,7 @@
 import org.apache.flink.configuration.ReadableConfig;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
 import org.apache.flink.table.api.TableSchema;
 import org.apache.flink.table.data.RowData;
 import org.apache.iceberg.Schema;
@@ -46,6 +47,14 @@
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.apache.iceberg.util.PropertyUtil;
 
+/**
+ * Flink source builder for old {@link SourceFunction} implementation.
+ *
+ * @deprecated since 1.7.0, will be removed in 2.0.0. Use {@link IcebergSource} instead, which
+ *     implement the newer FLIP-27 source interface. This class implements the old {@link
+ *     SourceFunction} that has been marked as deprecated in Flink since Aug 2023.
+ */
+@Deprecated
 public class FlinkSource {
   private FlinkSource() {}
 

flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/source/IcebergSource.java

@@ -27,7 +27,6 @@
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeUnit;
 import javax.annotation.Nullable;
-import org.apache.flink.annotation.Experimental;
 import org.apache.flink.api.common.eventtime.WatermarkStrategy;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.connector.source.Boundedness;
@@ -86,7 +85,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-@Experimental
 public class IcebergSource<T> implements Source<T, IcebergSourceSplit, IcebergEnumeratorState> {
   private static final Logger LOG = LoggerFactory.getLogger(IcebergSource.class);
 

flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/source/FlinkSource.java

@@ -28,6 +28,7 @@
 import org.apache.flink.configuration.ReadableConfig;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
 import org.apache.flink.table.api.TableSchema;
 import org.apache.flink.table.data.RowData;
 import org.apache.iceberg.Schema;
@@ -46,6 +47,14 @@
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.apache.iceberg.util.PropertyUtil;
 
+/**
+ * Flink source builder for old {@link SourceFunction} implementation.
+ *
+ * @deprecated since 1.7.0, will be removed in 2.0.0. Use {@link IcebergSource} instead, which
+ *     implement the newer FLIP-27 source interface. This class implements the old {@link
+ *     SourceFunction} that has been marked as deprecated in Flink since Aug 2023.
+ */
+@Deprecated
 public class FlinkSource {
   private FlinkSource() {}
 

flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/source/IcebergSource.java

@@ -27,7 +27,6 @@
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeUnit;
 import javax.annotation.Nullable;
-import org.apache.flink.annotation.Experimental;
 import org.apache.flink.api.common.eventtime.WatermarkStrategy;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.connector.source.Boundedness;
@@ -86,7 +85,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-@Experimental
 public class IcebergSource<T> implements Source<T, IcebergSourceSplit, IcebergEnumeratorState> {
   private static final Logger LOG = LoggerFactory.getLogger(IcebergSource.class);
 

flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkConfigOptions.java

@@ -88,7 +88,7 @@ private FlinkConfigOptions() {}
   public static final ConfigOption<Boolean> TABLE_EXEC_ICEBERG_USE_FLIP27_SOURCE =
       ConfigOptions.key("table.exec.iceberg.use-flip27-source")
           .booleanType()
-          .defaultValue(false)
+          .defaultValue(true)
           .withDescription("Use the FLIP-27 based Iceberg source implementation.");
 
   public static final ConfigOption<SplitAssignerType> TABLE_EXEC_SPLIT_ASSIGNER_TYPE =

flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/source/FlinkSource.java

@@ -28,6 +28,7 @@
 import org.apache.flink.configuration.ReadableConfig;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
 import org.apache.flink.table.api.TableSchema;
 import org.apache.flink.table.data.RowData;
 import org.apache.iceberg.Schema;
@@ -46,6 +47,14 @@
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.apache.iceberg.util.PropertyUtil;
 
+/**
+ * /** Flink source builder for old {@link SourceFunction} implementation.
+ *
+ * @deprecated since 1.7.0, will be removed in 2.0.0. Use {@link IcebergSource} instead, which
+ *     implement the newer FLIP-27 source interface. This class implements the old {@link
+ *     SourceFunction} that has been marked as deprecated in Flink since Aug 2023.
+ */
+@Deprecated
 public class FlinkSource {
   private FlinkSource() {}
 

flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/source/IcebergSource.java

@@ -27,7 +27,6 @@
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeUnit;
 import javax.annotation.Nullable;
-import org.apache.flink.annotation.Experimental;
 import org.apache.flink.api.common.eventtime.WatermarkStrategy;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.connector.source.Boundedness;
@@ -86,7 +85,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-@Experimental
 public class IcebergSource<T> implements Source<T, IcebergSourceSplit, IcebergEnumeratorState> {
   private static final Logger LOG = LoggerFactory.getLogger(IcebergSource.class);