class Elasticsearch7DynamicSinkFactory implements DynamicTableSinkFactory {

	@Override
	public DynamicTableSink createDynamicTableSink(Context context) {
		TableSchema tableSchema = context.getCatalogTable().getSchema();
		ElasticsearchValidationUtils.validatePrimaryKey(tableSchema);

		final FactoryUtil.TableFactoryHelper helper = FactoryUtil.createTableFactoryHelper(this, context);

		final EncodingFormat<SerializationSchema<RowData>> format = helper.discoverEncodingFormat(
			SerializationFormatFactory.class,
			FORMAT_OPTION);

		helper.validate();
		Configuration configuration = new Configuration();
		context.getCatalogTable()
			.getOptions()
			.forEach(configuration::setString);
		Elasticsearch7Configuration config = new Elasticsearch7Configuration(configuration, context.getClassLoader());

		validate(config, configuration);

		return new Elasticsearch7DynamicSink(
			format,
			config,
			TableSchemaUtils.getPhysicalSchema(tableSchema));
	}

}