private <T> DataStream<T> appendWithRateLimit(DataStream<T> input) {
            if (tableOptions.get(WRITE_RATE_LIMIT) <= 0) {
                return input;
            }

            int parallelism = writeParallelism == null ? input.getParallelism() : writeParallelism;
            SingleOutputStreamOperator<T> inputWithRateLimit = input
                    .map(new RateLimitMapFunction(tableOptions.get(WRITE_RATE_LIMIT)))
                    .name("rate_limit")
                    .setParallelism(parallelism);

            if (uidPrefix != null) {
                ((SingleOutputStreamOperator) inputWithRateLimit).uid(uidPrefix + "_rate_limit");
            }
            return inputWithRateLimit;
        }
Append the iceberg sink operators to write records to iceberg table.

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