private String createKey(Row row) {
			final StringBuilder builder = new StringBuilder();
			for (int i = 0; i < keyFieldIndices.length; i++) {
				final int keyFieldIndex = keyFieldIndices[i];
				if (i > 0) {
					builder.append(keyDelimiter);
				}
				final Object value = row.getField(keyFieldIndex);
				if (value == null) {
					builder.append(keyNullLiteral);
				} else {
					builder.append(value.toString());
				}
			}
			return builder.toString();
		}
Sink function for converting upserts into Elasticsearch {@link ActionRequest}s.
elasticsearch5-xh/elasticsearch5-xh-sink/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchUpsertTableSinkBase.java