DATAES-547 - Polishing.

Add test and directly use SearchHit to pass on the index name.
Fix minor flaw in Exception translation for non existing indices along the way.

Original Pull Request: #257
This commit is contained in:
Christoph Strobl 2019-04-09 13:58:52 +02:00
parent 082988955b
commit 16bf8450f0
6 changed files with 177 additions and 67 deletions

View File

@ -1,31 +0,0 @@
/*
* Copyright 2019 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.core;
import lombok.Value;
/**
* DeleteEntry
*
* @author Lorenzo Spinelli
*/
@Value
public class DeleteEntry {
private final String id;
private final String indexName;
}

View File

@ -17,13 +17,17 @@
package org.springframework.data.elasticsearch.core; package org.springframework.data.elasticsearch.core;
import java.net.ConnectException; import java.net.ConnectException;
import java.util.List;
import org.elasticsearch.ElasticsearchException; import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.ElasticsearchStatusException;
import org.springframework.dao.DataAccessException; import org.springframework.dao.DataAccessException;
import org.springframework.dao.DataAccessResourceFailureException; import org.springframework.dao.DataAccessResourceFailureException;
import org.springframework.dao.support.PersistenceExceptionTranslator; import org.springframework.dao.support.PersistenceExceptionTranslator;
import org.springframework.data.elasticsearch.NoSuchIndexException; import org.springframework.data.elasticsearch.NoSuchIndexException;
import org.springframework.util.CollectionUtils; import org.springframework.util.CollectionUtils;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;
/** /**
* @author Christoph Strobl * @author Christoph Strobl
@ -39,7 +43,8 @@ public class ElasticsearchExceptionTranslator implements PersistenceExceptionTra
ElasticsearchException elasticsearchException = (ElasticsearchException) ex; ElasticsearchException elasticsearchException = (ElasticsearchException) ex;
if (!indexAvailable(elasticsearchException)) { if (!indexAvailable(elasticsearchException)) {
return new NoSuchIndexException(elasticsearchException.getMetadata("es.index").toString(), ex); return new NoSuchIndexException(ObjectUtils.nullSafeToString(elasticsearchException.getMetadata("es.index")),
ex);
} }
} }
@ -51,6 +56,13 @@ public class ElasticsearchExceptionTranslator implements PersistenceExceptionTra
} }
private boolean indexAvailable(ElasticsearchException ex) { private boolean indexAvailable(ElasticsearchException ex) {
return !CollectionUtils.contains(ex.getMetadata("es.index_uuid").iterator(), "_na_");
List<String> metadata = ex.getMetadata("es.index_uuid");
if (metadata == null) {
if (ex instanceof ElasticsearchStatusException) {
return StringUtils.hasText(ObjectUtils.nullSafeToString(((ElasticsearchStatusException) ex).getIndex()));
}
}
return !CollectionUtils.contains(metadata.iterator(), "_na_");
} }
} }

View File

@ -869,35 +869,27 @@ public class ElasticsearchRestTemplate
SearchQuery searchQuery = new NativeSearchQueryBuilder().withQuery(deleteQuery.getQuery()).withIndices(indexName) SearchQuery searchQuery = new NativeSearchQueryBuilder().withQuery(deleteQuery.getQuery()).withIndices(indexName)
.withTypes(typeName).withPageable(PageRequest.of(0, pageSize)).build(); .withTypes(typeName).withPageable(PageRequest.of(0, pageSize)).build();
SearchResultMapper deleteentryResultMapper = new SearchResultMapperAdapter() { SearchResultMapper deleteEntryResultMapper = new SearchResultMapperAdapter() {
@Override @Override
public <T> AggregatedPage<T> mapResults(SearchResponse response, Class<T> clazz, Pageable pageable) { public <T> AggregatedPage<T> mapResults(SearchResponse response, Class<T> clazz, Pageable pageable) {
List<DeleteEntry> result = new ArrayList<>(); return new AggregatedPageImpl<>((List<T>) Arrays.asList(response.getHits().getHits()), response.getScrollId());
for (SearchHit searchHit : response.getHits().getHits()) {
String id = searchHit.getId();
String indexName = searchHit.getIndex();
result.add(new DeleteEntry(id, indexName));
}
if (result.size() > 0) {
return new AggregatedPageImpl<>((List<T>) result, response.getScrollId());
}
return new AggregatedPageImpl<>(Collections.emptyList(), response.getScrollId());
} }
}; };
Page<DeleteEntry> scrolledResult = startScroll(scrollTimeInMillis, searchQuery, DeleteEntry.class, Page<SearchHit> scrolledResult = startScroll(scrollTimeInMillis, searchQuery, SearchHit.class,
deleteentryResultMapper); deleteEntryResultMapper);
BulkRequest request = new BulkRequest(); BulkRequest request = new BulkRequest();
List<DeleteEntry> documentsToDelete = new ArrayList<>(); List<SearchHit> documentsToDelete = new ArrayList<>();
do { do {
documentsToDelete.addAll(scrolledResult.getContent()); documentsToDelete.addAll(scrolledResult.getContent());
scrolledResult = continueScroll(((ScrolledPage<T>) scrolledResult).getScrollId(), scrollTimeInMillis, scrolledResult = continueScroll(((ScrolledPage<T>) scrolledResult).getScrollId(), scrollTimeInMillis,
DeleteEntry.class, deleteentryResultMapper); SearchHit.class, deleteEntryResultMapper);
} while (scrolledResult.getContent().size() != 0); } while (scrolledResult.getContent().size() != 0);
for (DeleteEntry entry : documentsToDelete) { for (SearchHit entry : documentsToDelete) {
request.add(new DeleteRequest(entry.getIndexName(), typeName, entry.getId())); request.add(new DeleteRequest(entry.getIndex(), typeName, entry.getId()));
} }
if (request.numberOfActions() > 0) { if (request.numberOfActions() > 0) {

View File

@ -24,7 +24,7 @@ import java.io.BufferedReader;
import java.io.IOException; import java.io.IOException;
import java.io.InputStreamReader; import java.io.InputStreamReader;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.Collections; import java.util.Arrays;
import java.util.HashMap; import java.util.HashMap;
import java.util.Iterator; import java.util.Iterator;
import java.util.LinkedList; import java.util.LinkedList;
@ -757,35 +757,26 @@ public class ElasticsearchTemplate implements ElasticsearchOperations, EsClient<
.withTypes(typeName).withPageable(PageRequest.of(0, pageSize)).build(); .withTypes(typeName).withPageable(PageRequest.of(0, pageSize)).build();
SearchResultMapper deleteEntryResultMapper = new SearchResultMapperAdapter() { SearchResultMapper deleteEntryResultMapper = new SearchResultMapperAdapter() {
@Override @Override
public <T> AggregatedPage<T> mapResults(SearchResponse response, Class<T> clazz, Pageable pageable) { public <T> AggregatedPage<T> mapResults(SearchResponse response, Class<T> clazz, Pageable pageable) {
List<DeleteEntry> result = new ArrayList<>(); return new AggregatedPageImpl<>((List<T>) Arrays.asList(response.getHits().getHits()), response.getScrollId());
for (SearchHit searchHit : response.getHits().getHits()) {
String id = searchHit.getId();
String indexName = searchHit.getIndex();
result.add(new DeleteEntry(id, indexName));
}
if (result.size() > 0) {
return new AggregatedPageImpl<T>((List<T>) result, response.getScrollId());
}
return new AggregatedPageImpl<T>(Collections.emptyList(), response.getScrollId());
} }
}; };
Page<DeleteEntry> scrolledResult = startScroll(scrollTimeInMillis, searchQuery, DeleteEntry.class, Page<SearchHit> scrolledResult = startScroll(scrollTimeInMillis, searchQuery, SearchHit.class,
deleteEntryResultMapper); deleteEntryResultMapper);
BulkRequestBuilder bulkRequestBuilder = client.prepareBulk(); BulkRequestBuilder bulkRequestBuilder = client.prepareBulk();
List<DeleteEntry> documentsToDelete = new ArrayList<>(); List<SearchHit> documentsToDelete = new ArrayList<>();
do { do {
documentsToDelete.addAll(scrolledResult.getContent()); documentsToDelete.addAll(scrolledResult.getContent());
scrolledResult = continueScroll(((ScrolledPage<T>) scrolledResult).getScrollId(), scrollTimeInMillis, scrolledResult = continueScroll(((ScrolledPage<T>) scrolledResult).getScrollId(), scrollTimeInMillis,
DeleteEntry.class, deleteEntryResultMapper); SearchHit.class, deleteEntryResultMapper);
} while (scrolledResult.getContent().size() != 0); } while (scrolledResult.getContent().size() != 0);
for (DeleteEntry entry : documentsToDelete) { for (SearchHit entry : documentsToDelete) {
bulkRequestBuilder.add(client.prepareDelete(entry.getIndexName(), typeName, entry.getId())); bulkRequestBuilder.add(client.prepareDelete(entry.getIndex(), typeName, entry.getId()));
} }
if (bulkRequestBuilder.numberOfActions() > 0) { if (bulkRequestBuilder.numberOfActions() > 0) {

View File

@ -66,12 +66,14 @@ import org.springframework.data.elasticsearch.entities.GTEVersionEntity;
import org.springframework.data.elasticsearch.entities.HetroEntity1; import org.springframework.data.elasticsearch.entities.HetroEntity1;
import org.springframework.data.elasticsearch.entities.HetroEntity2; import org.springframework.data.elasticsearch.entities.HetroEntity2;
import org.springframework.data.elasticsearch.entities.SampleEntity; import org.springframework.data.elasticsearch.entities.SampleEntity;
import org.springframework.data.elasticsearch.entities.SampleEntityUUIDKeyed;
import org.springframework.data.elasticsearch.entities.SampleMappingEntity; import org.springframework.data.elasticsearch.entities.SampleMappingEntity;
import org.springframework.data.elasticsearch.entities.UseServerConfigurationEntity; import org.springframework.data.elasticsearch.entities.UseServerConfigurationEntity;
import org.springframework.data.util.CloseableIterator; import org.springframework.data.util.CloseableIterator;
/** /**
* Base for testing rest/transport templates * Base for testing rest/transport templates
*
* @author Rizwan Idrees * @author Rizwan Idrees
* @author Mohsin Husen * @author Mohsin Husen
* @author Franck Marchand * @author Franck Marchand
@ -101,9 +103,15 @@ public class ElasticsearchTemplateTests {
@Before @Before
public void before() { public void before() {
elasticsearchTemplate.deleteIndex(SampleEntity.class); elasticsearchTemplate.deleteIndex(SampleEntity.class);
elasticsearchTemplate.createIndex(SampleEntity.class); elasticsearchTemplate.createIndex(SampleEntity.class);
elasticsearchTemplate.putMapping(SampleEntity.class); elasticsearchTemplate.putMapping(SampleEntity.class);
elasticsearchTemplate.deleteIndex(SampleEntityUUIDKeyed.class);
elasticsearchTemplate.createIndex(SampleEntityUUIDKeyed.class);
elasticsearchTemplate.putMapping(SampleEntityUUIDKeyed.class);
elasticsearchTemplate.deleteIndex(INDEX_1_NAME); elasticsearchTemplate.deleteIndex(INDEX_1_NAME);
elasticsearchTemplate.deleteIndex(INDEX_2_NAME); elasticsearchTemplate.deleteIndex(INDEX_2_NAME);
elasticsearchTemplate.deleteIndex(UseServerConfigurationEntity.class); elasticsearchTemplate.deleteIndex(UseServerConfigurationEntity.class);
@ -405,6 +413,86 @@ public class ElasticsearchTemplateTests {
assertThat(sampleEntities.getTotalElements(), equalTo(0L)); assertThat(sampleEntities.getTotalElements(), equalTo(0L));
} }
@Test // DATAES-547
public void shouldDeleteAcrossIndex() {
// given
SampleEntity sampleEntity = SampleEntity.builder() //
.message("foo") //
.version(System.currentTimeMillis()) //
.build();
IndexQuery idxQuery1 = new IndexQueryBuilder().withIndexName(INDEX_1_NAME).withId(randomNumeric(5))
.withObject(sampleEntity).build();
elasticsearchTemplate.index(idxQuery1);
elasticsearchTemplate.refresh(INDEX_1_NAME);
IndexQuery idxQuery2 = new IndexQueryBuilder().withIndexName(INDEX_2_NAME).withId(randomNumeric(5))
.withObject(sampleEntity).build();
elasticsearchTemplate.index(idxQuery2);
elasticsearchTemplate.refresh(INDEX_2_NAME);
// when
DeleteQuery deleteQuery = new DeleteQuery();
deleteQuery.setQuery(termQuery("message", "foo"));
deleteQuery.setType("test-type");
deleteQuery.setIndex("test-index-*");
elasticsearchTemplate.delete(deleteQuery);
elasticsearchTemplate.refresh(INDEX_1_NAME);
elasticsearchTemplate.refresh(INDEX_2_NAME);
// then
SearchQuery searchQuery = new NativeSearchQueryBuilder().withQuery(termQuery("message", "foo"))
.withIndices(INDEX_1_NAME, INDEX_2_NAME) //
.build();
assertThat(elasticsearchTemplate.count(searchQuery), equalTo(0L));
}
@Test // DATAES-547
public void shouldDeleteAcrossIndexWhenNoMatchingDataPresent() {
// given
SampleEntity sampleEntity = SampleEntity.builder() //
.message("positive") //
.version(System.currentTimeMillis()) //
.build();
IndexQuery idxQuery1 = new IndexQueryBuilder().withIndexName(INDEX_1_NAME).withId(randomNumeric(5))
.withObject(sampleEntity).build();
elasticsearchTemplate.index(idxQuery1);
elasticsearchTemplate.refresh(INDEX_1_NAME);
IndexQuery idxQuery2 = new IndexQueryBuilder().withIndexName(INDEX_2_NAME).withId(randomNumeric(5))
.withObject(sampleEntity).build();
elasticsearchTemplate.index(idxQuery2);
elasticsearchTemplate.refresh(INDEX_2_NAME);
// when
DeleteQuery deleteQuery = new DeleteQuery();
deleteQuery.setQuery(termQuery("message", "negative"));
deleteQuery.setType("test-type");
deleteQuery.setIndex("test-index-*");
elasticsearchTemplate.delete(deleteQuery);
elasticsearchTemplate.refresh(INDEX_1_NAME);
elasticsearchTemplate.refresh(INDEX_2_NAME);
// then
SearchQuery searchQuery = new NativeSearchQueryBuilder().withQuery(termQuery("message", "positive"))
.withIndices(INDEX_1_NAME, INDEX_2_NAME) //
.build();
assertThat(elasticsearchTemplate.count(searchQuery), equalTo(2L));
}
@Test @Test
public void shouldFilterSearchResultsForGivenFilter() { public void shouldFilterSearchResultsForGivenFilter() {
// given // given

View File

@ -52,6 +52,8 @@ import org.springframework.data.elasticsearch.core.query.Criteria;
import org.springframework.data.elasticsearch.core.query.CriteriaQuery; import org.springframework.data.elasticsearch.core.query.CriteriaQuery;
import org.springframework.data.elasticsearch.core.query.IndexQuery; import org.springframework.data.elasticsearch.core.query.IndexQuery;
import org.springframework.data.elasticsearch.core.query.IndexQueryBuilder; import org.springframework.data.elasticsearch.core.query.IndexQueryBuilder;
import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
import org.springframework.data.elasticsearch.core.query.SearchQuery;
import org.springframework.data.elasticsearch.core.query.StringQuery; import org.springframework.data.elasticsearch.core.query.StringQuery;
import org.springframework.data.elasticsearch.entities.SampleEntity; import org.springframework.data.elasticsearch.entities.SampleEntity;
import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.ContextConfiguration;
@ -519,6 +521,62 @@ public class ReactiveElasticsearchTemplateTests {
.verifyComplete(); .verifyComplete();
} }
@Test // DATAES-547
@ElasticsearchVersion(asOf = "6.5.0")
public void shouldDeleteAcrossIndex() {
String indexPrefix = "rx-template-test-index";
String thisIndex = indexPrefix + "-this";
String thatIndex = indexPrefix + "-that";
template.save(randomEntity("test"), thisIndex) //
.then(template.save(randomEntity("test"), thatIndex)) //
.then() //
.as(StepVerifier::create)//
.verifyComplete();
restTemplate.refresh(thisIndex);
restTemplate.refresh(thatIndex);
SearchQuery searchQuery = new NativeSearchQueryBuilder() //
.withQuery(termQuery("message", "test")) //
.withIndices(indexPrefix + "*") //
.build();
template.deleteBy(searchQuery, SampleEntity.class) //
.as(StepVerifier::create) //
.expectNext(2L) //
.verifyComplete();
}
@Test // DATAES-547
@ElasticsearchVersion(asOf = "6.5.0")
public void shouldDeleteAcrossIndexWhenNoMatchingDataPresent() {
String indexPrefix = "rx-template-test-index";
String thisIndex = indexPrefix + "-this";
String thatIndex = indexPrefix + "-that";
template.save(randomEntity("positive"), thisIndex) //
.then(template.save(randomEntity("positive"), thatIndex)) //
.then() //
.as(StepVerifier::create)//
.verifyComplete();
restTemplate.refresh(thisIndex);
restTemplate.refresh(thatIndex);
SearchQuery searchQuery = new NativeSearchQueryBuilder() //
.withQuery(termQuery("message", "negative")) //
.withIndices(indexPrefix + "*") //
.build();
template.deleteBy(searchQuery, SampleEntity.class) //
.as(StepVerifier::create) //
.expectNext(0L) //
.verifyComplete();
}
@Test // DATAES-504 @Test // DATAES-504
@ElasticsearchVersion(asOf = "6.5.0") @ElasticsearchVersion(asOf = "6.5.0")
public void deleteByQueryShouldReturnNumberOfDeletedDocuments() { public void deleteByQueryShouldReturnNumberOfDeletedDocuments() {