public static SinkFunctionFactory.SinkVersion chooseSinkVersionAutomatically(StarRocksSinkOptions sinkOptions) {
        if (StarRocksSinkSemantic.AT_LEAST_ONCE.equals(sinkOptions.getSemantic())) {
            LOG.info("Choose sink version V2 for at-least-once.");
            return SinkVersion.V2;
        }

        if (sinkOptions.isSupportTransactionStreamLoad()) {
            LOG.info("StarRocks supports transaction load, and choose sink version V2");
            return SinkVersion.V2;
        } else {
            LOG.info("StarRocks does not support transaction load, and choose sink version V1");
            return SinkVersion.V1;
        }
    }
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