@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 testPrefixedRequiredOption() {
        DataSinkFactory sinkFactory = getElasticsearchDataSinkFactory();

        Configuration conf =
                Configuration.fromMap(
                        ImmutableMap.<String, String>builder()
                                .put("hosts", "localhost:9200")
                                .put("batch.size.max", "500")
                                .put("inflight.requests.max", "5")
                                .put("version", "7") // Added version to the test configuration
                                .build());

        // 打印日志以确保我们在测试带前缀的必需选项
        System.out.println("Testing prefixed required option");

        DataSink dataSink = createDataSink(sinkFactory, conf);
        Assertions.assertThat(dataSink).isInstanceOf(ElasticsearchDataSink.class);
    }