static IcebergStreamWriter<RowData> createStreamWriter(
            Table table,
            FlinkWriteConf flinkWriteConf,
            RowType flinkRowType,
            List<Integer> equalityFieldIds,
            SinkMetadataUtils sinkMetadataUtils,
            MetricOption metricOption) {
        Preconditions.checkArgument(table != null, "Iceberg table shouldn't be null");

        Table serializableTable = SerializableTable.copyOf(table);
        FileFormat format = flinkWriteConf.dataFileFormat();
        TaskWriterFactory<RowData> taskWriterFactory =
                new RowDataTaskWriterFactory(
                        serializableTable,
                        flinkRowType,
                        flinkWriteConf.targetDataFileSize(),
                        format,
                        writeProperties(table, format, flinkWriteConf),
                        equalityFieldIds,
                        flinkWriteConf.upsertMode());
        return new IcebergStreamWriter<>(table.name(), taskWriterFactory, sinkMetadataUtils, metricOption);
    }
Append the iceberg sink operators to write records to iceberg table.

@return {@link DataStreamSink} for sink.
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/FlinkSink.java