@Override
    public DataSink createDataSink(Context context) {
        // Validate the configuration
        FactoryHelper.createFactoryHelper(this, context).validate();

        // Get the configuration directly from the context
        Configuration configuration =
                Configuration.fromMap(context.getFactoryConfiguration().toMap());

        // Validate required options
        validateRequiredOptions(configuration);

        ZoneId zoneId = determineZoneId(context);
        ElasticsearchSinkOptions sinkOptions = buildSinkConnectorOptions(configuration);
        return new ElasticsearchDataSink(sinkOptions, zoneId);
    }
@Test
    void testUnsupportedOption() {
        DataSinkFactory sinkFactory = getElasticsearchDataSinkFactory();

        Configuration conf =
                Configuration.fromMap(
                        ImmutableMap.<String, String>builder()
                                .put("hosts", "localhost:9200")
                                .put("version", "7")
                                .put("unsupported_key", "unsupported_value")
                                .build());

        // 打印日志以确保我们在测试不受支持的选项
        System.out.println("Testing unsupported option");

        // Assertions to check for unsupported options
        Assertions.assertThatThrownBy(() -> createDataSink(sinkFactory, conf))
                .isInstanceOf(ValidationException.class)
                .hasMessageContaining(
                        "Unsupported options found for 'elasticsearch'.\n\n"
                                + "Unsupported options:\n\n"
                                + "unsupported_key");
    }