private SingleOutputStreamOperator<WriteResult> appendWriter(
                DataStream<RowData> input, RowType flinkRowType, List<Integer> equalityFieldIds,
                MetricOption metricOption) {
            // Validate the equality fields and partition fields if we enable the upsert mode.
            if (flinkWriteConf.upsertMode()) {
                Preconditions.checkState(
                        !flinkWriteConf.overwriteMode(),
                        "OVERWRITE mode shouldn't be enable when configuring to use UPSERT data stream.");
                Preconditions.checkState(
                        !equalityFieldIds.isEmpty(),
                        "Equality field columns shouldn't be empty when configuring to use UPSERT data stream.");
                if (!table.spec().isUnpartitioned()) {
                    for (PartitionField partitionField : table.spec().fields()) {
                        Preconditions.checkState(
                                equalityFieldIds.contains(partitionField.sourceId()),
                                "In UPSERT mode, partition field '%s' should be included in equality fields: '%s'",
                                partitionField,
                                equalityFieldColumns);
                    }
                }
            }

            IcebergStreamWriter<RowData> streamWriter =
                    createStreamWriter(table, flinkWriteConf, flinkRowType, equalityFieldIds, sinkMetadataUtils,
                            metricOption);

            int parallelism =
                    flinkWriteConf.writeParallelism() == null
                            ? input.getParallelism()
                            : flinkWriteConf.writeParallelism();
            SingleOutputStreamOperator<WriteResult> writerStream =
                    input
                            .transform(
                                    operatorName(ICEBERG_STREAM_WRITER_NAME),
                                    TypeInformation.of(WriteResult.class),
                                    streamWriter)
                            .setParallelism(parallelism);
            if (uidPrefix != null) {
                writerStream = writerStream.uid(uidPrefix + "-writer");
            }
            return writerStream;
        }
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