DATAES-839 - ReactiveElasticsearchTemplate should use RequestFactory.

Original PR: spring-projects#466

Author: sothawo

src/main/java/org/springframework/data/elasticsearch/BulkFailureException.java

@@ -0,0 +1,37 @@
+/*
+ * Copyright 2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.elasticsearch;
+
+import org.springframework.dao.DataRetrievalFailureException;
+
+import java.util.Map;
+
+/**
+ * @author Peter-Josef Meisch
+ * @since 4.1
+ */
+public class BulkFailureException extends DataRetrievalFailureException {
+	private final Map<String, String> failedDocuments;
+
+	public BulkFailureException(String msg, Map<String, String> failedDocuments) {
+		super(msg);
+		this.failedDocuments = failedDocuments;
+	}
+
+	public Map<String, String> getFailedDocuments() {
+		return failedDocuments;
+	}
+}

src/main/java/org/springframework/data/elasticsearch/UncategorizedElasticsearchException.java

@@ -22,6 +22,7 @@
  * @since 4.0
  */
 public class UncategorizedElasticsearchException extends UncategorizedDataAccessException {
+
 	public UncategorizedElasticsearchException(String msg, Throwable cause) {
 		super(msg, cause);
 	}

src/main/java/org/springframework/data/elasticsearch/core/AbstractElasticsearchTemplate.java

@@ -36,7 +36,7 @@
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationContextAware;
 import org.springframework.data.convert.EntityReader;
-import org.springframework.data.elasticsearch.ElasticsearchException;
+import org.springframework.data.elasticsearch.BulkFailureException;
 import org.springframework.data.elasticsearch.core.convert.ElasticsearchConverter;
 import org.springframework.data.elasticsearch.core.convert.MappingElasticsearchConverter;
 import org.springframework.data.elasticsearch.core.document.Document;
@@ -405,7 +405,7 @@ protected List<String> checkForBulkOperationFailure(BulkResponse bulkResponse) {
 				if (item.isFailed())
 					failedDocuments.put(item.getId(), item.getFailureMessage());
 			}
-			throw new ElasticsearchException(
+			throw new BulkFailureException(
 					"Bulk operation has failures. Use ElasticsearchException.getFailedDocuments() for detailed messages ["
 							+ failedDocuments + ']',
 					failedDocuments);

src/main/java/org/springframework/data/elasticsearch/core/ElasticsearchRestTemplate.java

@@ -210,7 +210,7 @@ public String delete(String id, IndexCoordinates index) {
 		Assert.notNull(id, "id must not be null");
 		Assert.notNull(index, "index must not be null");
 
-		DeleteRequest request = new DeleteRequest(index.getIndexName(), elasticsearchConverter.convertId(id));
+		DeleteRequest request = requestFactory.deleteRequest(elasticsearchConverter.convertId(id), index);
 		return execute(client -> client.delete(request, RequestOptions.DEFAULT).getId());
 	}