@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 testLackRequiredOption() {
        DataSinkFactory sinkFactory = getElasticsearchDataSinkFactory();
        List<String> requiredKeys = getRequiredKeys(sinkFactory);
        for (String requiredKey : requiredKeys) {
            // 创建一个新的配置 Map，包含所有必需选项
            Map<String, String> options = new HashMap<>(createValidOptions());
            // 移除当前正在测试的必需选项
            options.remove(requiredKey);
            Configuration conf = Configuration.fromMap(options);
            // 打印日志以确保我们在测试缺少必需选项的情况
            System.out.println("Testing missing required option: " + requiredKey);

            // 添加创建 DataSink 对象的代码
            Assertions.assertThatThrownBy(() -> createDataSink(sinkFactory, conf))

                    // Assertions to check for missing required option
                    .isInstanceOf(ValidationException.class)
                    .hasMessageContaining(
                            String.format(
                                    "One or more required options are missing.\n\n"
                                            + "Missing required options are:\n\n"
                                            + "%s",
                                    requiredKey));
        }
    }