@Override
	public DataStreamSink<?> consumeDataStream(DataStream<Tuple2<Boolean, Row>> dataStream) {
		final ElasticsearchUpsertSinkFunction upsertFunction =
			new ElasticsearchUpsertSinkFunction(
				index,
				docType,
				keyDelimiter,
				keyNullLiteral,
				serializationSchema,
				contentType,
				requestFactory,
				keyFieldIndices);
		final SinkFunction<Tuple2<Boolean, Row>> sinkFunction = createSinkFunction(
			hosts,
			failureHandler,
			sinkOptions,
			upsertFunction);
		return dataStream.addSink(sinkFunction)
			.setParallelism(dataStream.getParallelism())
			.name(TableConnectorUtils.generateRuntimeName(this.getClass(), getFieldNames()));
	}
Key field indices determined by the query.
elasticsearch5-xh/elasticsearch5-xh-sink/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchUpsertTableSinkBase.java