Add more implementations using the new client.

Original Pull Request #2136
See #1973

Author: sothawo

Diff for: src/main/java/org/springframework/data/elasticsearch/NoSuchIndexException.java

@@ -25,6 +25,14 @@ public class NoSuchIndexException extends NonTransientDataAccessResourceExceptio
 
 	private final String index;
 
+	/**
+	 * @since 4.4
+	 */
+	public NoSuchIndexException(String index) {
+		super(String.format("Index %s not found.", index));
+		this.index = index;
+	}
+
 	public NoSuchIndexException(String index, Throwable cause) {
 		super(String.format("Index %s not found.", index), cause);
 		this.index = index;

Diff for: src/main/java/org/springframework/data/elasticsearch/client/elc/Aggregation.java

@@ -0,0 +1,44 @@
+/*
+ * Copyright 2022 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.client.elc;
+
+import co.elastic.clients.elasticsearch._types.aggregations.Aggregate;
+
+/**
+ * Class to combine an Elasticsearch {@link co.elastic.clients.elasticsearch._types.aggregations.Aggregate} with its
+ * name. Necessary as the Elasticsearch Aggregate does not know i"s name.
+ *
+ * @author Peter-Josef Meisch
+ * @since 4.4
+ */
+public class Aggregation {
+
+	private final String name;
+	private final Aggregate aggregate;
+
+	public Aggregation(String name, Aggregate aggregate) {
+		this.name = name;
+		this.aggregate = aggregate;
+	}
+
+	public String getName() {
+		return name;
+	}
+
+	public Aggregate getAggregate() {
+		return aggregate;
+	}
+}

Diff for: src/main/java/org/springframework/data/elasticsearch/client/elc/DocumentAdapters.java

@@ -28,6 +28,7 @@
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.function.Function;
 import java.util.stream.Collectors;
 
 import org.apache.commons.logging.Log;
@@ -77,18 +78,12 @@ public static SearchDocument from(Hit<?> hit, JsonpMapper jsonpMapper) {
 
 		NestedMetaData nestedMetaData = from(hit.nested());
 
-		// todo #1973 explanation
 		Explanation explanation = from(hit.explanation());
 
 		// todo #1973 matchedQueries
 		List<String> matchedQueries = null;
-		// todo #1973 documentFields
-		Map<String, List<Object>> documentFields = Collections.emptyMap();
 
-		Document document;
-		Object source = hit.source();
-		if (source == null) {
-			// Elasticsearch provides raw JsonData, so we build the fields into a JSON string
+		Function<Map<String, JsonData>, EntityAsMap> fromFields = fields -> {
 			StringBuilder sb = new StringBuilder("{");
 			final boolean[] firstField = { true };
 			hit.fields().forEach((key, jsonData) -> {
@@ -100,7 +95,25 @@ public static SearchDocument from(Hit<?> hit, JsonpMapper jsonpMapper) {
 				firstField[0] = false;
 			});
 			sb.append('}');
-			document = Document.parse(sb.toString());
+			return new EntityAsMap().fromJson(sb.toString());
+		};
+
+		EntityAsMap hitFieldsAsMap = fromFields.apply(hit.fields());
+
+		Map<String, List<Object>> documentFields = new LinkedHashMap<>();
+		hitFieldsAsMap.entrySet().forEach(entry -> {
+			if (entry.getValue() instanceof List) {
+				// noinspection unchecked
+				documentFields.put(entry.getKey(), (List<Object>) entry.getValue());
+			} else {
+				documentFields.put(entry.getKey(), Collections.singletonList(entry.getValue()));
+			}
+		});
+
+		Document document;
+		Object source = hit.source();
+		if (source == null) {
+			document = Document.from(hitFieldsAsMap);
 		} else {
 			if (source instanceof EntityAsMap) {
 				document = Document.from((EntityAsMap) source);

Diff for: src/main/java/org/springframework/data/elasticsearch/client/elc/ElasticsearchAggregation.java

@@ -0,0 +1,37 @@
+/*
+ * Copyright 2022 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.client.elc;
+
+import org.springframework.data.elasticsearch.core.AggregationContainer;
+
+/**
+ * {@link AggregationContainer} for a {@link Aggregation} that holds Elasticsearch data.
+ * @author Peter-Josef Meisch
+ * @since 4.4
+ */
+public class ElasticsearchAggregation implements AggregationContainer<Aggregation> {
+
+	private final Aggregation aggregation;
+
+	public ElasticsearchAggregation(Aggregation aggregation) {
+		this.aggregation = aggregation;
+	}
+
+	@Override
+	public Aggregation aggregation() {
+		return aggregation;
+	}
+}

Diff for: src/main/java/org/springframework/data/elasticsearch/client/elc/ElasticsearchAggregations.java

@@ -17,26 +17,46 @@
 
 import co.elastic.clients.elasticsearch._types.aggregations.Aggregate;
 
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Map;
 
 import org.springframework.data.elasticsearch.core.AggregationsContainer;
+import org.springframework.util.Assert;
 
 /**
  * AggregationsContainer implementation for the Elasticsearch aggregations.
  *
  * @author Peter-Josef Meisch
  * @since 4.4
  */
-public class ElasticsearchAggregations implements AggregationsContainer<Map<String, Aggregate>> {
+public class ElasticsearchAggregations implements AggregationsContainer<List<ElasticsearchAggregation>> {
 
-	private final Map<String, Aggregate> aggregations;
+	private final List<ElasticsearchAggregation> aggregations;
+
+	public ElasticsearchAggregations(List<ElasticsearchAggregation> aggregations) {
+
+		Assert.notNull(aggregations, "aggregations must not be null");
 
-	public ElasticsearchAggregations(Map<String, Aggregate> aggregations) {
 		this.aggregations = aggregations;
 	}
 
+	/**
+	 * convenience constructor taking a map as it is returned from the new Elasticsearch client.
+	 *
+	 * @param aggregationsMap aggregate map
+	 */
+	public ElasticsearchAggregations(Map<String, Aggregate> aggregationsMap) {
+
+		Assert.notNull(aggregationsMap, "aggregationsMap must not be null");
+
+		aggregations = new ArrayList<>(aggregationsMap.size());
+		aggregationsMap
+				.forEach((name, aggregate) -> aggregations.add(new ElasticsearchAggregation(new Aggregation(name, aggregate))));
+	}
+
 	@Override
-	public Map<String, Aggregate> aggregations() {
+	public List<ElasticsearchAggregation> aggregations() {
 		return aggregations;
 	}
 }

Diff for: src/main/java/org/springframework/data/elasticsearch/client/elc/ElasticsearchExceptionTranslator.java

@@ -20,13 +20,18 @@
 import co.elastic.clients.json.JsonpMapper;
 
 import java.io.IOException;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 
+import org.elasticsearch.client.ResponseException;
 import org.springframework.dao.DataAccessException;
 import org.springframework.dao.DataAccessResourceFailureException;
+import org.springframework.dao.DataIntegrityViolationException;
 import org.springframework.dao.OptimisticLockingFailureException;
 import org.springframework.dao.support.PersistenceExceptionTranslator;
-import org.springframework.data.elasticsearch.RestStatusException;
+import org.springframework.data.elasticsearch.NoSuchIndexException;
 import org.springframework.data.elasticsearch.UncategorizedElasticsearchException;
+import org.springframework.http.HttpStatus;
 
 /**
  * Simple {@link PersistenceExceptionTranslator} for Elasticsearch. Convert the given runtime exception to an
@@ -67,14 +72,28 @@ public DataAccessException translateExceptionIfPossible(RuntimeException ex) {
 			return new OptimisticLockingFailureException("Cannot index a document due to seq_no+primary_term conflict", ex);
 		}
 
-		// todo #1973 index unavailable?
-
 		if (ex instanceof ElasticsearchException) {
 			ElasticsearchException elasticsearchException = (ElasticsearchException) ex;
 
 			ErrorResponse response = elasticsearchException.response();
+
+			if (response.status() == HttpStatus.NOT_FOUND.value()
+					&& "index_not_found_exception".equals(response.error().type())) {
+
+				Pattern pattern = Pattern.compile(".*no such index \\[(.*)\\]");
+				String index = "";
+				Matcher matcher = pattern.matcher(response.error().reason());
+				if (matcher.matches()) {
+					index = matcher.group(1);
+				}
+				return new NoSuchIndexException(index);
+			}
 			String body = JsonUtils.toJson(response, jsonpMapper);
 
+			if (response.error().type().contains("validation_exception")) {
+				return new DataIntegrityViolationException(response.error().reason());
+			}
+
 			return new UncategorizedElasticsearchException(ex.getMessage(), response.status(), body, ex);
 		}
 
@@ -86,20 +105,22 @@ public DataAccessException translateExceptionIfPossible(RuntimeException ex) {
 		return null;
 	}
 
-	private boolean isSeqNoConflict(Exception exception) {
+	private boolean isSeqNoConflict(Throwable exception) {
 		// todo #1973 check if this works
 		Integer status = null;
 		String message = null;
 
-		if (exception instanceof RestStatusException) {
 
-			RestStatusException statusException = (RestStatusException) exception;
-			status = statusException.getStatus();
-			message = statusException.getMessage();
+		if (exception instanceof ResponseException) {
+			ResponseException responseException = (ResponseException) exception;
+			status = responseException.getResponse().getStatusLine().getStatusCode();
+			message = responseException.getMessage();
+		} else if (exception.getCause() != null) {
+			return isSeqNoConflict(exception.getCause());
 		}
 
 		if (status != null && message != null) {
-			return status == 409 && message.contains("type=version_conflict_engine_exception")
+			return status == 409 && message.contains("type\":\"version_conflict_engine_exception")
 					&& message.contains("version conflict, required seqNo");
 		}