Flink: table supplier in writer operator for optional table refresh

flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/FlinkWriteConf.java

@@ -18,7 +18,9 @@
  */
 package org.apache.iceberg.flink;
 
+import java.time.Duration;
 import java.util.Map;
+import org.apache.calcite.linq4j.function.Experimental;
 import org.apache.flink.configuration.ReadableConfig;
 import org.apache.iceberg.DistributionMode;
 import org.apache.iceberg.FileFormat;
@@ -184,4 +186,27 @@ public String branch() {
   public Integer writeParallelism() {
     return confParser.intConf().option(FlinkWriteOptions.WRITE_PARALLELISM.key()).parseOptional();
   }
+
+  /**
+   * NOTE: This may be removed or changed in a future release. This value specifies the interval for
+   * refreshing the table instance in {@link IcebergStreamWriter}. If not specified then the default
+   * behavior is to not refresh the table.
+   *
+   * @return the interval for refreshing the table in writer subtasks
+   */
+  @Experimental
+  public Duration tableRefreshInterval() {
+    Long intervalMs =
+        confParser
+            .longConf()
+            .option(FlinkWriteOptions.TABLE_REFRSH_MS.key())
+            .flinkConfig(FlinkWriteOptions.TABLE_REFRSH_MS)
+            .parseOptional();
+
+    if (intervalMs == null) {
+      return null;
+    }
+
+    return Duration.ofMillis(intervalMs);
+  }
 }

flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/FlinkWriteOptions.java

@@ -18,6 +18,7 @@
  */
 package org.apache.iceberg.flink;
 
+import org.apache.calcite.linq4j.function.Experimental;
 import org.apache.flink.configuration.ConfigOption;
 import org.apache.flink.configuration.ConfigOptions;
 import org.apache.iceberg.SnapshotRef;
@@ -64,4 +65,8 @@ private FlinkWriteOptions() {}
 
   public static final ConfigOption<Integer> WRITE_PARALLELISM =
       ConfigOptions.key("write-parallelism").intType().noDefaultValue();
+
+  @Experimental
+  public static final ConfigOption<Long> TABLE_REFRSH_MS =
+      ConfigOptions.key("table-refresh-ms").longType().noDefaultValue();
 }

flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java

@@ -33,7 +33,6 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.function.Function;
-import org.apache.flink.annotation.Experimental;
 import org.apache.flink.api.common.functions.MapFunction;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.common.typeinfo.Types;
@@ -133,7 +132,6 @@ public static Builder forRowData(DataStream<RowData> input) {
   public static class Builder {
     private Function<String, DataStream<RowData>> inputCreator = null;
     private TableLoader tableLoader;
-    private Duration tableRefreshInterval;
     private Table table;
     private TableSchema tableSchema;
     private List<String> equalityFieldColumns = null;
@@ -270,20 +268,6 @@ public Builder upsert(boolean enabled) {
       return this;
     }
 
-    /**
-     * Sets the interval for refreshing the table instance in {@link IcebergStreamWriter}. If not
-     * specified then the default behavior is to not refresh the table, and the initial table
-     * instance initialized is used for the lifetime of the job.
-     *
-     * @param refreshInterval the interval for refreshing the table in writer subtasks
-     * @return {@link Builder} to connect the iceberg table.
-     */
-    @Experimental
-    public Builder tableRefreshInterval(Duration refreshInterval) {
-      this.tableRefreshInterval = refreshInterval;
-      return this;
-    }
-
     /**
      * Configuring the equality field columns for iceberg table that accept CDC or UPSERT events.
      *
@@ -484,6 +468,8 @@ private SingleOutputStreamOperator<WriteResult> appendWriter(
       }
 
       SerializableTable serializableTable = (SerializableTable) SerializableTable.copyOf(table);
+      Duration tableRefreshInterval = flinkWriteConf.tableRefreshInterval();
+
       SerializableSupplier<Table> tableSupplier;
       if (tableRefreshInterval != null) {
         tableSupplier =

flink/v1.17/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSink.java

@@ -19,10 +19,10 @@
 package org.apache.iceberg.flink.sink;
 
 import java.io.IOException;
-import java.time.Duration;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import org.apache.flink.configuration.Configuration;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.table.api.TableSchema;
@@ -378,10 +378,13 @@ public void testWriteRowWithTableRefreshInterval() throws Exception {
         env.addSource(createBoundedSource(rows), ROW_TYPE_INFO)
             .map(CONVERTER::toInternal, FlinkCompatibilityUtil.toTypeInfo(SimpleDataUtil.ROW_TYPE));
 
+    Configuration flinkConf = new Configuration();
+    flinkConf.setLong(FlinkWriteOptions.TABLE_REFRSH_MS.key(), 100L);
+
     FlinkSink.forRowData(dataStream)
         .table(table)
         .tableLoader(tableLoader)
-        .tableRefreshInterval(Duration.ofMillis(100))
+        .flinkConf(flinkConf)
         .writeParallelism(parallelism)
         .append();