private List<Host> getHosts(DescriptorProperties descriptorProperties) {
		if (descriptorProperties.containsKey(CONNECTOR_HOSTS)) {
			return validateAndParseHostsString(descriptorProperties);
		} else {
			final List<Map<String, String>> hosts = descriptorProperties.getFixedIndexedProperties(
				CONNECTOR_HOSTS,
				Arrays.asList(CONNECTOR_HOSTS_HOSTNAME, CONNECTOR_HOSTS_PORT, CONNECTOR_HOSTS_PROTOCOL));
			return hosts.stream()
				.map(host -> new Host(
					descriptorProperties.getString(host.get(CONNECTOR_HOSTS_HOSTNAME)),
					descriptorProperties.getInt(host.get(CONNECTOR_HOSTS_PORT)),
					descriptorProperties.getString(host.get(CONNECTOR_HOSTS_PROTOCOL))))
				.collect(Collectors.toList());
		}
	}
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