public DataStreamSink<?> consumeStream(DataStream<RowData> dataStream) {

        SinkFunction<RowData> kuduSinkFunction = createSinkFunction();

        return dataStream
                .addSink(kuduSinkFunction)
                .setParallelism(dataStream.getParallelism())
                .name(TableConnectorUtils.generateRuntimeName(this.getClass(), kuduTableInfo.getFieldNames()));
    }
The names of the key fields of the upsert stream consumed by this sink.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/kudu/src/main/java/org/apache/inlong/sort/kudu/table/KuduDynamicTableSink.java