private void mapSinkOption(
			DescriptorProperties descriptorProperties,
			Map<SinkOption, String> options,
			String fromKey,
			SinkOption toKey) {
		descriptorProperties.getOptionalString(fromKey).ifPresent(v -> options.put(toKey, v));
	}
Version-agnostic table factory for creating an {@link UpsertStreamTableSink} for Elasticsearch.
elasticsearch5-xh/elasticsearch5-xh-sink/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchUpsertTableSinkFactoryBase.java