@Override
	public void setKeyFields(String[] keyNames) {
		if (keyNames == null) {
			this.keyFieldIndices = new int[0];
			return;
		}

		final String[] fieldNames = getFieldNames();
		final int[] keyFieldIndices = new int[keyNames.length];
		for (int i = 0; i < keyNames.length; i++) {
			keyFieldIndices[i] = -1;
			for (int j = 0; j < fieldNames.length; j++) {
				if (keyNames[i].equals(fieldNames[j])) {
					keyFieldIndices[i] = j;
					break;
				}
			}
			if (keyFieldIndices[i] == -1) {
				throw new RuntimeException("Invalid key fields: " + Arrays.toString(keyNames));
			}
		}

		validateKeyTypes(keyFieldIndices);

		this.keyFieldIndices = keyFieldIndices;
	}
Key field indices determined by the query.
elasticsearch5-xh/elasticsearch5-xh-sink/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchUpsertTableSinkBase.java