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

        Configuration configuration =
                Configuration.fromMap(context.getFactoryConfiguration().toMap());
        DeliveryGuarantee deliveryGuarantee =
                context.getFactoryConfiguration().get(KafkaDataSinkOptions.DELIVERY_GUARANTEE);
        ZoneId zoneId = ZoneId.systemDefault();
        if (!Objects.equals(
                context.getPipelineConfiguration().get(PipelineOptions.PIPELINE_LOCAL_TIME_ZONE),
                PipelineOptions.PIPELINE_LOCAL_TIME_ZONE.defaultValue())) {
            zoneId =
                    ZoneId.of(
                            context.getPipelineConfiguration()
                                    .get(PipelineOptions.PIPELINE_LOCAL_TIME_ZONE));
        }
        KeyFormat keyFormat = context.getFactoryConfiguration().get(KEY_FORMAT);
        SerializationSchema<Event> keySerialization =
                KeySerializationFactory.createSerializationSchema(configuration, keyFormat, zoneId);
        JsonSerializationType jsonSerializationType =
                context.getFactoryConfiguration().get(KafkaDataSinkOptions.VALUE_FORMAT);
        SerializationSchema<Event> valueSerialization =
                ChangeLogJsonFormatFactory.createSerializationSchema(
                        configuration, jsonSerializationType, zoneId);
        final Properties kafkaProperties = new Properties();
        Map<String, String> allOptions = context.getFactoryConfiguration().toMap();
        allOptions.keySet().stream()
                .filter(key -> key.startsWith(PROPERTIES_PREFIX))
                .forEach(
                        key -> {
                            final String value = allOptions.get(key);
                            final String subKey = key.substring((PROPERTIES_PREFIX).length());
                            kafkaProperties.put(subKey, value);
                        });
        String topic = context.getFactoryConfiguration().get(KafkaDataSinkOptions.TOPIC);
        boolean addTableToHeaderEnabled =
                context.getFactoryConfiguration()
                        .get(KafkaDataSinkOptions.SINK_ADD_TABLEID_TO_HEADER_ENABLED);
        String customHeaders =
                context.getFactoryConfiguration().get(KafkaDataSinkOptions.SINK_CUSTOM_HEADER);
        PartitionStrategy partitionStrategy =
                context.getFactoryConfiguration().get(KafkaDataSinkOptions.PARTITION_STRATEGY);
        return new KafkaDataSink(
                deliveryGuarantee,
                kafkaProperties,
                partitionStrategy,
                zoneId,
                keySerialization,
                valueSerialization,
                topic,
                addTableToHeaderEnabled,
                customHeaders);
    }
@Test
    void testCreateDataSink() {
        DataSinkFactory sinkFactory =
                FactoryDiscoveryUtils.getFactoryByIdentifier("kafka", DataSinkFactory.class);
        Assertions.assertThat(sinkFactory).isInstanceOf(KafkaDataSinkFactory.class);

        Configuration conf = Configuration.fromMap(ImmutableMap.<String, String>builder().build());
        DataSink dataSink =
                sinkFactory.createDataSink(
                        new FactoryHelper.DefaultContext(
                                conf, conf, Thread.currentThread().getContextClassLoader()));
        Assertions.assertThat(dataSink).isInstanceOf(KafkaDataSink.class);
    }