Skip to content

Add more implementations using the new client. #2136

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Apr 13, 2022
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
Original file line number Diff line number Diff line change
@@ -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;
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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) -> {
Expand All @@ -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);
Expand Down
Original file line number Diff line number Diff line change
@@ -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;
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -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;
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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);
}

Expand All @@ -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");
}

Expand Down
Loading