private void validateKeyTypes(int[] keyFieldIndices) {
		final TypeInformation<?>[] types = getFieldTypes();
		for (int keyFieldIndex : keyFieldIndices) {
			final TypeInformation<?> type = types[keyFieldIndex];
			if (!TypeCheckUtils.isSimpleStringRepresentation(type)) {
				throw new ValidationException(
					"Only simple types that can be safely converted into a string representation " +
						"can be used as keys. But was: " + type);
			}
		}
	}
Validate the types that are used for conversion to string.
elasticsearch5-xh/elasticsearch5-xh-sink/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchUpsertTableSinkBase.java