Extend BulkFailureException.failedDocuments return type to show more details about failure.

Original Pull Request #2633
Related tickets #2619

Author: ilya-ulyanov

src/main/asciidoc/reference/elasticsearch-migration-guide-5.1-5.2.adoc

@@ -6,6 +6,14 @@ This section describes breaking changes from version 5.1.x to 5.2.x and how remo
 [[elasticsearch-migration-guide-5.1-5.2.breaking-changes]]
 == Breaking Changes
 
+In the `org.springframework.data.elasticsearch.BulkFailureException` class, the return type of the `getFailedDocuments` is changed from `Map<String, String>`
+to `Map<String, FailureDetails>`, which allows to get additional details about failure reasons.
+
+The definition of the `FailureDetails` class (inner to `BulkFailureException`):
+[source,java]
+public record FailureDetails(Integer status, String errorMessage) {
+}
+
 [[elasticsearch-migration-guide-5.1-5.2.deprecations]]
 == Deprecations
 

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

@@ -21,17 +21,27 @@
 
 /**
  * @author Peter-Josef Meisch
+ * @author Illia Ulianov
  * @since 4.1
  */
 public class BulkFailureException extends DataRetrievalFailureException {
-	private final Map<String, String> failedDocuments;
+	private final Map<String, FailureDetails> failedDocuments;
 
-	public BulkFailureException(String msg, Map<String, String> failedDocuments) {
+	public BulkFailureException(String msg, Map<String, FailureDetails> failedDocuments) {
 		super(msg);
 		this.failedDocuments = failedDocuments;
 	}
 
-	public Map<String, String> getFailedDocuments() {
+	public Map<String, FailureDetails> getFailedDocuments() {
 		return failedDocuments;
 	}
+
+	/**
+	 * Details about a document saving failure.
+	 *
+	 * @author Illia Ulianov
+	 * @since 5.2
+	 */
+	public record FailureDetails(Integer status, String errorMessage) {
+	}
 }

src/main/java/org/springframework/data/elasticsearch/client/elc/ElasticsearchTemplate.java

@@ -55,6 +55,7 @@
  *
  * @author Peter-Josef Meisch
  * @author Hamid Rahimi
+ * @author Illia Ulianov
  * @since 4.4
  */
 public class ElasticsearchTemplate extends AbstractElasticsearchTemplate {
@@ -637,11 +638,11 @@ public BaseQueryBuilder queryBuilderWithIds(List<String> ids) {
 	protected List<IndexedObjectInformation> checkForBulkOperationFailure(BulkResponse bulkResponse) {
 
 		if (bulkResponse.errors()) {
-			Map<String, String> failedDocuments = new HashMap<>();
+			Map<String, BulkFailureException.FailureDetails> failedDocuments = new HashMap<>();
 			for (BulkResponseItem item : bulkResponse.items()) {
 
 				if (item.error() != null) {
-					failedDocuments.put(item.id(), item.error().reason());
+					failedDocuments.put(item.id(), new BulkFailureException.FailureDetails(item.status(), item.error().reason()));
 				}
 			}
 			throw new BulkFailureException(

src/main/java/org/springframework/data/elasticsearch/client/elc/ReactiveElasticsearchTemplate.java

@@ -68,6 +68,7 @@
  * Elasticsearch client.
  *
  * @author Peter-Josef Meisch
+ * @author Illia Ulianov
  * @since 4.4
  */
 public class ReactiveElasticsearchTemplate extends AbstractReactiveElasticsearchTemplate {
@@ -250,12 +251,12 @@ private Flux<BulkResponseItem> doBulkOperation(List<?> queries, BulkOptions bulk
 	private Mono<BulkResponse> checkForBulkOperationFailure(BulkResponse bulkResponse) {
 
 		if (bulkResponse.errors()) {
-			Map<String, String> failedDocuments = new HashMap<>();
+			Map<String, BulkFailureException.FailureDetails> failedDocuments = new HashMap<>();
 
 			for (BulkResponseItem item : bulkResponse.items()) {
 
 				if (item.error() != null) {
-					failedDocuments.put(item.id(), item.error().reason());
+					failedDocuments.put(item.id(), new BulkFailureException.FailureDetails(item.status(), item.error().reason()));
 				}
 			}
 			BulkFailureException exception = new BulkFailureException(

src/test/java/org/springframework/data/elasticsearch/BulkFailureExceptionTest.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright 2023 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 static org.assertj.core.api.Assertions.*;
+
+import java.util.Map;
+
+import org.junit.jupiter.api.Test;
+
+/**
+ * @author Illia Ulianov
+ */
+class BulkFailureExceptionTest {
+
+	@Test // #2619
+	void shouldCreateBulkException() {
+		String documentId = "id1";
+		var failureDetails = new BulkFailureException.FailureDetails(409, "conflict");
+		var exception = new BulkFailureException("Test message", Map.of(documentId, failureDetails));
+		assertThat(exception.getFailedDocuments()).containsEntry(documentId, failureDetails);
+	}
+}

src/test/java/org/springframework/data/elasticsearch/core/ElasticsearchIntegrationTests.java

@@ -33,6 +33,7 @@
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 
+import org.assertj.core.api.InstanceOfAssertFactories;
 import org.assertj.core.api.SoftAssertions;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.DisplayName;
@@ -48,6 +49,7 @@
 import org.springframework.data.domain.PageRequest;
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.Sort;
+import org.springframework.data.elasticsearch.BulkFailureException;
 import org.springframework.data.elasticsearch.annotations.*;
 import org.springframework.data.elasticsearch.annotations.Field;
 import org.springframework.data.elasticsearch.annotations.ScriptedField;
@@ -98,6 +100,7 @@
  * @author Haibo Liu
  * @author scoobyzhang
  * @author Hamid Rahimi
+ * @author Illia Ulianov
  */
 @SpringIntegrationTest
 public abstract class ElasticsearchIntegrationTests {
@@ -3616,6 +3619,27 @@ void shouldNotErrorOnSortWithUnmappedFieldAndUnmappedTypeSettings() {
 		operations.search(query, SampleEntity.class);
 	}
 
+	@Test // #2619
+	void shouldFailWithConflictOnAttemptToSaveWithSameVersion() {
+		var entity1 = new VersionedEntity();
+		entity1.setId("id1");
+		entity1.setVersion(1L);
+		var entity2 = new VersionedEntity();
+		entity2.setId("id2");
+		entity2.setVersion(1L);
+		operations.save(entity1, entity2);
+
+		entity1.setVersion(2L);
+		assertThatThrownBy(() -> operations.save(entity1, entity2))
+			.asInstanceOf(InstanceOfAssertFactories.type(BulkFailureException.class))
+			.extracting(BulkFailureException::getFailedDocuments)
+			.asInstanceOf(InstanceOfAssertFactories.map(String.class, BulkFailureException.FailureDetails.class))
+			.containsOnlyKeys("id2")
+			.extracting(Map::values)
+			.asInstanceOf(InstanceOfAssertFactories.collection(BulkFailureException.FailureDetails.class))
+			.allMatch(failureStatus -> failureStatus.status().equals(409));
+	}
+
 	// region entities
 	@Document(indexName = "#{@indexNameProvider.indexName()}")
 	@Setting(shards = 1, replicas = 0, refreshInterval = "-1")

src/test/java/org/springframework/data/elasticsearch/core/ReactiveElasticsearchIntegrationTests.java

@@ -20,6 +20,8 @@
 import static org.springframework.data.elasticsearch.annotations.FieldType.*;
 import static org.springframework.data.elasticsearch.core.query.StringQuery.MATCH_ALL;
 
+import org.assertj.core.api.InstanceOfAssertFactories;
+import org.springframework.data.elasticsearch.BulkFailureException;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
@@ -89,6 +91,7 @@
  * @author Roman Puchkovskiy
  * @author George Popides
  * @author Sijia Liu
+ * @author Illia Ulianov
  */
 @SuppressWarnings("SpringJavaAutowiredMembersInspection")
 @SpringIntegrationTest
@@ -1192,6 +1195,27 @@ void shouldSaveDataFromFluxAndReturnSavedDataInAFlux() {
 				}) //
 				.verifyComplete();
 	}
+
+	@Test // #2619
+	void shouldFailWithConflictOnAttemptToSaveWithSameVersion() {
+		var entity1 = new VersionedEntity();
+		entity1.setId("id1");
+		entity1.setVersion(1L);
+		var entity2 = new VersionedEntity();
+		entity2.setId("id2");
+		entity2.setVersion(1L);
+		operations.saveAll(Arrays.asList(entity1, entity2), VersionedEntity.class).blockLast();
+
+		entity1.setVersion(2L);
+		assertThatThrownBy(() -> operations.saveAll(Arrays.asList(entity1, entity2), VersionedEntity.class).blockLast())
+				.asInstanceOf(InstanceOfAssertFactories.type(BulkFailureException.class))
+				.extracting(BulkFailureException::getFailedDocuments)
+				.asInstanceOf(InstanceOfAssertFactories.map(String.class, BulkFailureException.FailureDetails.class))
+				.containsOnlyKeys("id2").extracting(Map::values)
+				.asInstanceOf(InstanceOfAssertFactories.collection(BulkFailureException.FailureDetails.class))
+				.allMatch(failureStatus -> failureStatus.status().equals(409));
+	}
+
 	// endregion
 
 	// region Helper functions