@Override
            public DataStreamSink<?> consumeDataStream(
                    ProviderContext providerContext, DataStream<RowData> dataStream) {
                return FlinkSink.forRowData(dataStream)
                        .tableLoader(tableLoader)
                        .tableSchema(tableSchema)
                        .equalityFieldColumns(equalityColumns)
                        .overwrite(overwrite)
                        .setAll(writeProps)
                        .flinkConf(readableConfig)
                        .SinkMetadataUtils(sinkMetadataUtils)
                        .append();
            }
Copy from iceberg-flink:iceberg-flink-1.15:1.3.1
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/IcebergTableSink.java