private SerializationSchema<Row> getSerializationSchema(Map<String, String> properties) {
		final String formatType = properties.get(FORMAT_TYPE);
		// we could have added this check to the table factory context
		// but this approach allows to throw more helpful error messages
		// if the supported format has not been added
		if (formatType == null || !formatType.equals(SUPPORTED_FORMAT_TYPE)) {
			throw new ValidationException(
				"The Elasticsearch sink requires a '" + SUPPORTED_FORMAT_TYPE + "' format.");
		}

		@SuppressWarnings("unchecked")
		final SerializationSchemaFactory<Row> formatFactory = TableFactoryService.find(
			SerializationSchemaFactory.class,
			properties,
			this.getClass().getClassLoader());
		return formatFactory.createSerializationSchema(properties);
	}
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