@Override
    @SuppressWarnings("unchecked")
    public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {

        final TypeInformation<RowData> rowDataTypeInfo = context.createTypeInformation(flinkSchema.toRowDataType());
        StarRocksDynamicSinkFunctionBase<RowData> starrocksSinkFunction = SinkFunctionFactory.createSinkFunction(
                sinkOptions,
                flinkSchema,
                new StarRocksTableRowTransformer(rowDataTypeInfo, ignoreJsonParseError),
                inlongMetric,
                auditHostAndPorts,
                auditKeys);
        return SinkFunctionProvider.of(starrocksSinkFunction, sinkOptions.getSinkParallelism());

    }
This class is used to create a DynamicTableSink for Starrocks.
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/starrocks/src/main/java/org/apache/inlong/sort/starrocks/table/sink/table/StarRocksDynamicTableSink.java