private SinkFunction<RowData> createSinkFunction() {
        boolean sinkWithAsyncMode = configuration.getBoolean(KuduOptions.SINK_WRITE_WITH_ASYNC_MODE);
        if (sinkWithAsyncMode) {
            return new KuduAsyncSinkFunction(
                    kuduTableInfo,
                    configuration,
                    inlongMetric,
                    auditHostAndPorts);
        } else {
            return new KuduSinkFunction(
                    kuduTableInfo,
                    configuration,
                    inlongMetric,
                    auditHostAndPorts);
        }
    }
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