static RowType toFlinkRowType(Schema schema, TableSchema requestedSchema) {
        if (requestedSchema != null) {
            // Convert the flink schema to iceberg schema firstly, then reassign ids to match the existing
            // iceberg schema.
            Schema writeSchema = TypeUtil.reassignIds(FlinkSchemaUtil.convert(requestedSchema), schema);
            TypeUtil.validateWriteSchema(schema, writeSchema, true, true);

            // We use this flink schema to read values from RowData. The flink's TINYINT and SMALLINT will
            // be promoted to
            // iceberg INTEGER, that means if we use iceberg's table schema to read TINYINT (backend by 1
            // 'byte'), we will
            // read 4 bytes rather than 1 byte, it will mess up the byte array in BinaryRowData. So here
            // we must use flink
            // schema.
            return (RowType) requestedSchema.toRowDataType().getLogicalType();
        } else {
            return FlinkSchemaUtil.convert(schema);
        }
    }
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