Skip to content

Commit f08c34e

Browse files
authored
Improve multiget return.
Original Pull Request #1710 Closes #1678
1 parent dd3d01e commit f08c34e

21 files changed

+574
-361
lines changed

Diff for: src/main/asciidoc/reference/elasticsearch-migration-guide-4.1-4.2.adoc

+7
Original file line numberDiff line numberDiff line change
@@ -53,3 +53,10 @@ Previously the reactive code initialized this to `IMMEDIATE`, now reactive and n
5353
==== delete methods that take a Query
5454

5555
The reactive methods previously returned a `Mono<Long>` with the number of deleted documents, the non reactive versions were void. They now return a `Mono<ByQueryResponse>` which contains much more detailed information about the deleted documents and errors that might have occurred.
56+
57+
==== multiget methods
58+
59+
The implementations of _multiget_ previousl only returned the found entities in a `List<T>` for non-reactive implementations and in a `Flux<T>` for reactive implementations. If the request contained ids that were not found, the information that these are missing was not available. The user needed to compare the returned ids to the requested ones to find
60+
which ones were missing.
61+
62+
Now the `multiget` methods return a `MultiGetItem` for every requested id. This contains information about failures (like non existing indices) and the information if the item existed (then it is contained in the `MultiGetItem) or not.

Diff for: src/main/java/org/springframework/data/elasticsearch/client/reactive/DefaultReactiveElasticsearchClient.java

+3-8
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@
2222
import io.netty.handler.ssl.JdkSslContext;
2323
import io.netty.handler.timeout.ReadTimeoutHandler;
2424
import io.netty.handler.timeout.WriteTimeoutHandler;
25+
import org.elasticsearch.action.get.MultiGetItemResponse;
2526
import reactor.core.publisher.Flux;
2627
import reactor.core.publisher.Mono;
2728
import reactor.netty.http.client.HttpClient;
@@ -330,18 +331,12 @@ public Mono<GetResult> get(HttpHeaders headers, GetRequest getRequest) {
330331
.next();
331332
}
332333

333-
/*
334-
* (non-Javadoc)
335-
* @see org.springframework.data.elasticsearch.client.reactive.ReactiveElasticsearchClient#multiGet(org.springframework.http.HttpHeaders, org.elasticsearch.action.get.MultiGetRequest)
336-
*/
337334
@Override
338-
public Flux<GetResult> multiGet(HttpHeaders headers, MultiGetRequest multiGetRequest) {
335+
public Flux<MultiGetItemResponse> multiGet(HttpHeaders headers, MultiGetRequest multiGetRequest) {
339336

340337
return sendRequest(multiGetRequest, requestCreator.multiGet(), MultiGetResponse.class, headers)
341338
.map(MultiGetResponse::getResponses) //
342-
.flatMap(Flux::fromArray) //
343-
.filter(it -> !it.isFailed() && it.getResponse().isExists()) //
344-
.map(it -> DefaultReactiveElasticsearchClient.getResponseToGetResult(it.getResponse()));
339+
.flatMap(Flux::fromArray); //
345340
}
346341

347342
/*

Diff for: src/main/java/org/springframework/data/elasticsearch/client/reactive/ReactiveElasticsearchClient.java

+93-82
Large diffs are not rendered by default.

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

+1-1
Original file line numberDiff line numberDiff line change
@@ -250,7 +250,7 @@ public <T> T get(String id, Class<T> clazz) {
250250
}
251251

252252
@Override
253-
public <T> List<T> multiGet(Query query, Class<T> clazz) {
253+
public <T> List<MultiGetItem<T>> multiGet(Query query, Class<T> clazz) {
254254
return multiGet(query, clazz, getIndexCoordinatesFor(clazz));
255255
}
256256

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

+4-4
Original file line numberDiff line numberDiff line change
@@ -120,20 +120,20 @@ public interface DocumentOperations {
120120
*
121121
* @param query the query defining the ids of the objects to get
122122
* @param clazz the type of the object to be returned
123-
* @return list of objects, contains null values for ids that are not found
123+
* @return list of {@link MultiGetItem}s
124124
* @since 4.1
125125
*/
126-
<T> List<T> multiGet(Query query, Class<T> clazz);
126+
<T> List<MultiGetItem<T>> multiGet(Query query, Class<T> clazz);
127127

128128
/**
129129
* Execute a multiGet against elasticsearch for the given ids.
130130
*
131131
* @param query the query defining the ids of the objects to get
132132
* @param clazz the type of the object to be returned
133133
* @param index the index(es) from which the objects are read.
134-
* @return list of objects, contains null values for ids that are not found
134+
* @return list of {@link MultiGetItem}s
135135
*/
136-
<T> List<T> multiGet(Query query, Class<T> clazz, IndexCoordinates index);
136+
<T> List<MultiGetItem<T>> multiGet(Query query, Class<T> clazz, IndexCoordinates index);
137137

138138
/**
139139
* Check if an entity with given {@literal id} exists.

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

+5-2
Original file line numberDiff line numberDiff line change
@@ -168,7 +168,7 @@ public <T> T get(String id, Class<T> clazz, IndexCoordinates index) {
168168
}
169169

170170
@Override
171-
public <T> List<T> multiGet(Query query, Class<T> clazz, IndexCoordinates index) {
171+
public <T> List<MultiGetItem<T>> multiGet(Query query, Class<T> clazz, IndexCoordinates index) {
172172

173173
Assert.notNull(index, "index must not be null");
174174
Assert.notEmpty(query.getIds(), "No Id defined for Query");
@@ -177,7 +177,10 @@ public <T> List<T> multiGet(Query query, Class<T> clazz, IndexCoordinates index)
177177
MultiGetResponse result = execute(client -> client.mget(request, RequestOptions.DEFAULT));
178178

179179
DocumentCallback<T> callback = new ReadDocumentCallback<>(elasticsearchConverter, clazz, index);
180-
return DocumentAdapters.from(result).stream().map(callback::doWith).collect(Collectors.toList());
180+
return DocumentAdapters.from(result).stream() //
181+
.map(multiGetItem -> MultiGetItem.of( //
182+
multiGetItem.isFailed() ? null : callback.doWith(multiGetItem.getItem()), multiGetItem.getFailure())) //
183+
.collect(Collectors.toList());
181184
}
182185

183186
@Override

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

+6-4
Original file line numberDiff line numberDiff line change
@@ -42,7 +42,6 @@
4242
import org.slf4j.Logger;
4343
import org.slf4j.LoggerFactory;
4444
import org.springframework.data.elasticsearch.core.convert.ElasticsearchConverter;
45-
import org.springframework.data.elasticsearch.core.document.Document;
4645
import org.springframework.data.elasticsearch.core.document.DocumentAdapters;
4746
import org.springframework.data.elasticsearch.core.document.SearchDocumentResponse;
4847
import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates;
@@ -190,16 +189,19 @@ public <T> T get(String id, Class<T> clazz, IndexCoordinates index) {
190189
}
191190

192191
@Override
193-
public <T> List<T> multiGet(Query query, Class<T> clazz, IndexCoordinates index) {
192+
public <T> List<MultiGetItem<T>> multiGet(Query query, Class<T> clazz, IndexCoordinates index) {
194193

195194
Assert.notNull(index, "index must not be null");
196195
Assert.notEmpty(query.getIds(), "No Ids defined for Query");
197196

198197
MultiGetRequestBuilder builder = requestFactory.multiGetRequestBuilder(client, query, clazz, index);
199198

200199
DocumentCallback<T> callback = new ReadDocumentCallback<>(elasticsearchConverter, clazz, index);
201-
List<Document> documents = DocumentAdapters.from(builder.execute().actionGet());
202-
return documents.stream().map(callback::doWith).collect(Collectors.toList());
200+
201+
return DocumentAdapters.from(builder.execute().actionGet()).stream() //
202+
.map(multiGetItem -> MultiGetItem.of(multiGetItem.isFailed() ? null : callback.doWith(multiGetItem.getItem()),
203+
multiGetItem.getFailure()))
204+
.collect(Collectors.toList());
203205
}
204206

205207
@Override
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,96 @@
1+
/*
2+
* Copyright 2021 the original author or authors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* https://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
package org.springframework.data.elasticsearch.core;
17+
18+
import org.springframework.lang.Nullable;
19+
20+
/**
21+
* Response object for items returned from multiget requests, encapsulating the returned data and potential error
22+
* information.
23+
*
24+
* @param <T> the entity type
25+
* @author Peter-Josef Meisch
26+
* @since 4.2
27+
*/
28+
public class MultiGetItem<T> {
29+
@Nullable private final T item;
30+
@Nullable private final Failure failure;
31+
32+
private MultiGetItem(@Nullable T item, @Nullable Failure failure) {
33+
this.item = item;
34+
this.failure = failure;
35+
}
36+
37+
public static <T> MultiGetItem<T> of(@Nullable T item, @Nullable Failure failure) {
38+
return new MultiGetItem<>(item, failure);
39+
}
40+
41+
public boolean hasItem() {
42+
return item != null;
43+
}
44+
45+
@Nullable
46+
public T getItem() {
47+
return item;
48+
}
49+
50+
public boolean isFailed() {
51+
return failure != null;
52+
}
53+
54+
@Nullable
55+
public Failure getFailure() {
56+
return failure;
57+
}
58+
59+
public static class Failure {
60+
@Nullable private final String index;
61+
@Nullable private final String type;
62+
@Nullable private final String id;
63+
@Nullable private final Exception exception;
64+
65+
private Failure(@Nullable String index, @Nullable String type, @Nullable String id, @Nullable Exception exception) {
66+
this.index = index;
67+
this.type = type;
68+
this.id = id;
69+
this.exception = exception;
70+
}
71+
72+
public static Failure of(String index, String type, String id, Exception exception) {
73+
return new Failure(index, type, id, exception);
74+
}
75+
76+
@Nullable
77+
public String getIndex() {
78+
return index;
79+
}
80+
81+
@Nullable
82+
public String getType() {
83+
return type;
84+
}
85+
86+
@Nullable
87+
public String getId() {
88+
return id;
89+
}
90+
91+
@Nullable
92+
public Exception getException() {
93+
return exception;
94+
}
95+
}
96+
}

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

+4-4
Original file line numberDiff line numberDiff line change
@@ -147,21 +147,21 @@ default <T> Flux<T> saveAll(Iterable<T> entities, IndexCoordinates index) {
147147
*
148148
* @param query the query defining the ids of the objects to get
149149
* @param clazz the type of the object to be returned, used to determine the index
150-
* @return flux with list of nullable objects
150+
* @return flux with list of {@link MultiGetItem}s that contain the entities
151151
* @since 4.1
152152
*/
153-
<T> Flux<T> multiGet(Query query, Class<T> clazz);
153+
<T> Flux<MultiGetItem<T>> multiGet(Query query, Class<T> clazz);
154154

155155
/**
156156
* Execute a multiGet against elasticsearch for the given ids.
157157
*
158158
* @param query the query defining the ids of the objects to get
159159
* @param clazz the type of the object to be returned
160160
* @param index the index(es) from which the objects are read.
161-
* @return flux with list of nullable objects
161+
* @return flux with list of {@link MultiGetItem}s that contain the entities
162162
* @since 4.0
163163
*/
164-
<T> Flux<T> multiGet(Query query, Class<T> clazz, IndexCoordinates index);
164+
<T> Flux<MultiGetItem<T>> multiGet(Query query, Class<T> clazz, IndexCoordinates index);
165165

166166
/**
167167
* Bulk update all objects. Will do update.

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

+9-4
Original file line numberDiff line numberDiff line change
@@ -71,10 +71,10 @@
7171
import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates;
7272
import org.springframework.data.elasticsearch.core.mapping.SimpleElasticsearchMappingContext;
7373
import org.springframework.data.elasticsearch.core.query.BulkOptions;
74+
import org.springframework.data.elasticsearch.core.query.ByQueryResponse;
7475
import org.springframework.data.elasticsearch.core.query.IndexQuery;
7576
import org.springframework.data.elasticsearch.core.query.Query;
7677
import org.springframework.data.elasticsearch.core.query.SeqNoPrimaryTerm;
77-
import org.springframework.data.elasticsearch.core.query.ByQueryResponse;
7878
import org.springframework.data.elasticsearch.core.query.UpdateQuery;
7979
import org.springframework.data.elasticsearch.core.query.UpdateResponse;
8080
import org.springframework.data.elasticsearch.core.routing.DefaultRoutingResolver;
@@ -298,12 +298,12 @@ private <T> T updateIndexedObject(T entity, IndexedObjectInformation indexedObje
298298
}
299299

300300
@Override
301-
public <T> Flux<T> multiGet(Query query, Class<T> clazz) {
301+
public <T> Flux<MultiGetItem<T>> multiGet(Query query, Class<T> clazz) {
302302
return multiGet(query, clazz, getIndexCoordinatesFor(clazz));
303303
}
304304

305305
@Override
306-
public <T> Flux<T> multiGet(Query query, Class<T> clazz, IndexCoordinates index) {
306+
public <T> Flux<MultiGetItem<T>> multiGet(Query query, Class<T> clazz, IndexCoordinates index) {
307307

308308
Assert.notNull(index, "Index must not be null");
309309
Assert.notNull(clazz, "Class must not be null");
@@ -314,7 +314,12 @@ public <T> Flux<T> multiGet(Query query, Class<T> clazz, IndexCoordinates index)
314314

315315
MultiGetRequest request = requestFactory.multiGetRequest(query, clazz, index);
316316
return Flux.from(execute(client -> client.multiGet(request))) //
317-
.concatMap(result -> callback.toEntity(DocumentAdapters.from(result)));
317+
.map(DocumentAdapters::from) //
318+
.flatMap(multiGetItem -> multiGetItem.isFailed() //
319+
? Mono.just(MultiGetItem.of(null, multiGetItem.getFailure())) //
320+
: callback.toEntity(multiGetItem.getItem())
321+
.map((T item) -> MultiGetItem.of(item, multiGetItem.getFailure())) //
322+
);
318323
}
319324

320325
@Override

0 commit comments

Comments
 (0)