public static <T> StarRocksDynamicSinkFunctionBase<T> createSinkFunction(
            StarRocksSinkOptions sinkOptions, TableSchema schema, StarRocksIRowTransformer<T> rowTransformer,
            String inlongMetric, String auditHostAndPorts, String auditKeys) {
        detectStarRocksFeature(sinkOptions);
        SinkVersion sinkVersion = getSinkVersion(sinkOptions);
        switch (sinkVersion) {
            case V1:
                return new StarRocksDynamicSinkFunction<>(sinkOptions, schema, rowTransformer);
            case V2:
                return new StarRocksDynamicSinkFunctionV2<>(sinkOptions, schema, rowTransformer,
                        inlongMetric, auditHostAndPorts, auditKeys);
            default:
                throw new UnsupportedOperationException("Unsupported sink type " + sinkVersion.name());
        }
    }
Create sink function according to the configuration.
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/starrocks/src/main/java/org/apache/inlong/sort/starrocks/table/sink/table/SinkFunctionFactory.java