@Override
    public DataSink createDataSink(Context context) {
        FactoryHelper.createFactoryHelper(this, context)
                .validateExcept(TABLE_CREATE_PROPERTIES_PREFIX, SINK_PROPERTIES_PREFIX);

        StarRocksSinkOptions sinkOptions =
                buildSinkConnectorOptions(context.getFactoryConfiguration());
        TableCreateConfig tableCreateConfig =
                TableCreateConfig.from(context.getFactoryConfiguration());
        SchemaChangeConfig schemaChangeConfig =
                SchemaChangeConfig.from(context.getFactoryConfiguration());
        String zoneStr = context.getFactoryConfiguration().get(PIPELINE_LOCAL_TIME_ZONE);
        ZoneId zoneId =
                PIPELINE_LOCAL_TIME_ZONE.defaultValue().equals(zoneStr)
                        ? ZoneId.systemDefault()
                        : ZoneId.of(zoneStr);
        return new StarRocksDataSink(sinkOptions, tableCreateConfig, schemaChangeConfig, zoneId);
    }
@Test
    void testCreateDataSink() {
        DataSinkFactory sinkFactory =
                FactoryDiscoveryUtils.getFactoryByIdentifier("starrocks", DataSinkFactory.class);
        Assertions.assertThat(sinkFactory).isInstanceOf(StarRocksDataSinkFactory.class);

        Configuration conf =
                Configuration.fromMap(
                        ImmutableMap.<String, String>builder()
                                .put("jdbc-url", "jdbc:mysql://127.0.0.1:9030")
                                .put("load-url", "127.0.0.1:8030")
                                .put("username", "root")
                                .put("password", "")
                                .build());
        DataSink dataSink =
                sinkFactory.createDataSink(
                        new FactoryHelper.DefaultContext(
                                conf, conf, Thread.currentThread().getContextClassLoader()));
        Assertions.assertThat(dataSink).isInstanceOf(StarRocksDataSink.class);
    }