WIP: Updated Elasticsearch Version to 7.1.0 - DATAES-585

pom.xml

@@ -19,7 +19,7 @@
 
 	<properties>
 		<commonslang>2.6</commonslang>
-		<elasticsearch>6.7.2</elasticsearch>
+		<elasticsearch>7.1.0</elasticsearch>
 		<log4j>2.9.1</log4j>
 		<springdata.commons>2.2.0.BUILD-SNAPSHOT</springdata.commons>
 		<java-module-name>spring.data.elasticsearch</java-module-name>

src/main/java/org/springframework/data/elasticsearch/client/NodeClientFactoryBean.java

@@ -58,7 +58,7 @@ public static class TestNode extends Node {
 
 		public TestNode(Settings preparedSettings, Collection<Class<? extends Plugin>> classpathPlugins) {
 
-			super(InternalSettingsPreparer.prepareEnvironment(preparedSettings, null), classpathPlugins, false);
+			super(InternalSettingsPreparer.prepareEnvironment(preparedSettings, null, null, null), classpathPlugins, false);
 		}
 
 		protected void registerDerivedNodeNameWithLogger(String nodeName) {

src/main/java/org/springframework/data/elasticsearch/core/DefaultResultMapper.java

@@ -99,7 +99,7 @@ private static EntityMapper initEntityMapper(
 	@Override
 	public <T> AggregatedPage<T> mapResults(SearchResponse response, Class<T> clazz, Pageable pageable) {
 
-		long totalHits = response.getHits().getTotalHits();
+		long totalHits = response.getHits().getTotalHits().value;
 		float maxScore = response.getHits().getMaxScore();
 
 		List<T> results = new ArrayList<>();

src/main/java/org/springframework/data/elasticsearch/core/ElasticsearchTemplate.java

@@ -98,6 +98,7 @@
 import org.springframework.data.elasticsearch.core.mapping.ElasticsearchPersistentProperty;
 import org.springframework.data.elasticsearch.core.mapping.SimpleElasticsearchMappingContext;
 import org.springframework.data.elasticsearch.core.query.*;
+import org.springframework.data.elasticsearch.utils.ElasticsearchCloseableIterator;
 import org.springframework.data.util.CloseableIterator;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
@@ -447,61 +448,7 @@ public <T> CloseableIterator<T> stream(SearchQuery query, final Class<T> clazz,
 
 	private <T> CloseableIterator<T> doStream(final long scrollTimeInMillis, final ScrolledPage<T> page,
 			final Class<T> clazz, final SearchResultMapper mapper) {
-		return new CloseableIterator<T>() {
-
-			/** As we couldn't retrieve single result with scroll, store current hits. */
-			private volatile Iterator<T> currentHits = page.iterator();
-
-			/** The scroll id. */
-			private volatile String scrollId = page.getScrollId();
-
-			/** If stream is finished (ie: cluster returns no results. */
-			private volatile boolean finished = !currentHits.hasNext();
-
-			@Override
-			public void close() {
-				try {
-					// Clear scroll on cluster only in case of error (cause elasticsearch auto clear scroll when it's done)
-					if (!finished && scrollId != null && currentHits != null && currentHits.hasNext()) {
-						clearScroll(scrollId);
-					}
-				} finally {
-					currentHits = null;
-					scrollId = null;
-				}
-			}
-
-			@Override
-			public boolean hasNext() {
-				// Test if stream is finished
-				if (finished) {
-					return false;
-				}
-				// Test if it remains hits
-				if (currentHits == null || !currentHits.hasNext()) {
-					// Do a new request
-					final ScrolledPage<T> scroll = continueScroll(scrollId, scrollTimeInMillis, clazz, mapper);
-					// Save hits and scroll id
-					currentHits = scroll.iterator();
-					finished = !currentHits.hasNext();
-					scrollId = scroll.getScrollId();
-				}
-				return currentHits.hasNext();
-			}
-
-			@Override
-			public T next() {
-				if (hasNext()) {
-					return currentHits.next();
-				}
-				throw new NoSuchElementException();
-			}
-
-			@Override
-			public void remove() {
-				throw new UnsupportedOperationException("remove");
-			}
-		};
+		return new ElasticsearchCloseableIterator<>(this, page, scrollTimeInMillis, clazz, mapper);
 	}
 
 	@Override
@@ -546,7 +493,7 @@ private long doCount(SearchRequestBuilder countRequestBuilder, QueryBuilder elas
 		if (elasticsearchQuery != null) {
 			countRequestBuilder.setQuery(elasticsearchQuery);
 		}
-		return countRequestBuilder.execute().actionGet().getHits().getTotalHits();
+		return countRequestBuilder.execute().actionGet().getHits().getTotalHits().value;
 	}
 
 	private long doCount(SearchRequestBuilder searchRequestBuilder, QueryBuilder elasticsearchQuery,
@@ -559,7 +506,7 @@ private long doCount(SearchRequestBuilder searchRequestBuilder, QueryBuilder ela
 		if (elasticsearchFilter != null) {
 			searchRequestBuilder.setPostFilter(elasticsearchFilter);
 		}
-		return searchRequestBuilder.execute().actionGet().getHits().getTotalHits();
+		return searchRequestBuilder.execute().actionGet().getHits().getTotalHits().value;
 	}
 
 	private <T> SearchRequestBuilder prepareCount(Query query, Class<T> clazz) {
@@ -1161,9 +1108,10 @@ private IndexRequestBuilder prepareIndex(IndexQuery query) {
 				indexRequestBuilder.setVersionType(versionType);
 			}
 
-			if (query.getParentId() != null) {
-				indexRequestBuilder.setParent(query.getParentId());
-			}
+			// TODO find alternative
+			//if (query.getParentId() != null) {
+			//	indexRequestBuilder.setParent(query.getParentId());
+			//}
 
 			return indexRequestBuilder;
 		} catch (IOException e) {

src/main/java/org/springframework/data/elasticsearch/core/FacetedPageImpl.java

@@ -24,8 +24,8 @@
 import org.elasticsearch.search.aggregations.bucket.histogram.Histogram;
 import org.elasticsearch.search.aggregations.bucket.range.Range;
 import org.elasticsearch.search.aggregations.bucket.terms.Terms;
-import org.elasticsearch.search.aggregations.metrics.stats.extended.ExtendedStats;
-import org.elasticsearch.search.aggregations.metrics.sum.Sum;
+import org.elasticsearch.search.aggregations.metrics.ExtendedStats;
+import org.elasticsearch.search.aggregations.metrics.Sum;
 import org.joda.time.DateTime;
 import org.springframework.data.domain.PageImpl;
 import org.springframework.data.domain.Pageable;

src/main/java/org/springframework/data/elasticsearch/core/ReactiveElasticsearchTemplate.java

@@ -167,13 +167,14 @@ private Mono<IndexResponse> doIndex(Object value, AdaptibleEntity<?> entity, @Nu
 				}
 			}
 
-			if (entity.hasParent()) {
+			// TODO find alternative
+			/*if (entity.hasParent()) {
 
 				Object parentId = entity.getParentId();
 				if (parentId != null) {
 					request.parent(converter.convertId(parentId));
 				}
-			}
+			}*/
 
 			request = prepareIndexRequest(value, request);
 			return doIndex(request);

src/main/java/org/springframework/data/elasticsearch/utils/ElasticsearchCloseableIterator.java

@@ -0,0 +1,79 @@
+package org.springframework.data.elasticsearch.utils;
+
+import org.springframework.data.elasticsearch.core.*;
+import org.springframework.data.util.CloseableIterator;
+
+import java.util.Iterator;
+import java.util.NoSuchElementException;
+
+public class ElasticsearchCloseableIterator<T> implements CloseableIterator<T> {
+
+    private ElasticsearchOperations elasticsearchTemplate;
+    private final ScrolledPage<T> page;
+    private final long scrollTimeInMillis;
+    private final Class<T> clazz;
+    private final SearchResultMapper mapper;
+    /** As we couldn't retrieve single result with scroll, store current hits. */
+    private volatile Iterator<T> currentHits;
+
+    /** The scroll id. */
+    private volatile String scrollId;
+
+    /** If stream is finished (ie: cluster returns no results. */
+    private volatile boolean finished;
+
+    public ElasticsearchCloseableIterator(ElasticsearchOperations elasticsearchTemplate, ScrolledPage<T> page, long scrollTimeInMillis, Class<T> clazz, SearchResultMapper mapper) {
+        this.elasticsearchTemplate = elasticsearchTemplate;
+        this.page = page;
+        this.scrollTimeInMillis = scrollTimeInMillis;
+        this.clazz = clazz;
+        this.mapper = mapper;
+        currentHits = page.iterator();
+        scrollId = page.getScrollId();
+        finished = !currentHits.hasNext();
+    }
+
+    @Override
+    public void close() {
+        try {
+            // Clear scroll on cluster only in case of error (cause elasticsearch auto clear scroll when it's done)
+            if (!finished && scrollId != null && currentHits != null && currentHits.hasNext()) {
+                elasticsearchTemplate.clearScroll(scrollId);
+            }
+        } finally {
+            currentHits = null;
+            scrollId = null;
+        }
+    }
+
+    @Override
+    public boolean hasNext() {
+        // Test if stream is finished
+        if (finished) {
+            return false;
+        }
+        // Test if it remains hits
+        if (currentHits == null || !currentHits.hasNext()) {
+            // Do a new request
+            final ScrolledPage<T> scroll = elasticsearchTemplate.continueScroll(scrollId, scrollTimeInMillis, clazz, mapper);
+            // Save hits and scroll id
+            currentHits = scroll.iterator();
+            finished = !currentHits.hasNext();
+            scrollId = scroll.getScrollId();
+        }
+        return currentHits.hasNext();
+    }
+
+    @Override
+    public T next() {
+        if (hasNext()) {
+            return currentHits.next();
+        }
+        throw new NoSuchElementException();
+    }
+
+    @Override
+    public void remove() {
+        throw new UnsupportedOperationException("remove");
+    }
+}

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

@@ -94,7 +94,7 @@ public static boolean isEmptyIndex(String indexName) {
 			return 0L == client
 					.search(new SearchRequest(indexName)
 							.source(SearchSourceBuilder.searchSource().query(QueryBuilders.matchAllQuery())), RequestOptions.DEFAULT)
-					.getHits().getTotalHits();
+					.getHits().getTotalHits().value;
 		}
 	}
 

src/test/java/org/springframework/data/elasticsearch/client/reactive/ReactiveElasticsearchClientTests.java

@@ -135,8 +135,6 @@ public void infoShouldReturnClusterInformation() {
 
 		client.info().as(StepVerifier::create) //
 				.consumeNextWith(it -> {
-
-					assertThat(it.isAvailable()).isTrue();
 					assertThat(it.getVersion()).isGreaterThanOrEqualTo(Version.CURRENT);
 				}) //
 				.verifyComplete();

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

@@ -102,7 +102,7 @@ public void shouldMapAggregationsToPage() {
 		// Given
 		SearchHit[] hits = { createCarHit("Ford", "Grat"), createCarHit("BMW", "Arrow") };
 		SearchHits searchHits = mock(SearchHits.class);
-		when(searchHits.getTotalHits()).thenReturn(2L);
+		when(searchHits.getTotalHits().value).thenReturn(2L);
 		when(searchHits.iterator()).thenReturn(new ArrayIterator(hits));
 		when(response.getHits()).thenReturn(searchHits);
 
@@ -123,7 +123,7 @@ public void shouldMapSearchRequestToPage() {
 		// Given
 		SearchHit[] hits = { createCarHit("Ford", "Grat"), createCarHit("BMW", "Arrow") };
 		SearchHits searchHits = mock(SearchHits.class);
-		when(searchHits.getTotalHits()).thenReturn(2L);
+		when(searchHits.getTotalHits().value).thenReturn(2L);
 		when(searchHits.iterator()).thenReturn(new ArrayIterator(hits));
 		when(response.getHits()).thenReturn(searchHits);
 
@@ -141,7 +141,7 @@ public void shouldMapPartialSearchRequestToObject() {
 		// Given
 		SearchHit[] hits = { createCarPartialHit("Ford", "Grat"), createCarPartialHit("BMW", "Arrow") };
 		SearchHits searchHits = mock(SearchHits.class);
-		when(searchHits.getTotalHits()).thenReturn(2L);
+		when(searchHits.getTotalHits().value).thenReturn(2L);
 		when(searchHits.iterator()).thenReturn(new ArrayIterator(hits));
 		when(response.getHits()).thenReturn(searchHits);
 
@@ -234,7 +234,7 @@ public void setsVersionFromSearchResponse() {
 		when(hit2.getVersion()).thenReturn(5678L);
 
 		SearchHits searchHits = mock(SearchHits.class);
-		when(searchHits.getTotalHits()).thenReturn(2L);
+		when(searchHits.getTotalHits().value).thenReturn(2L);
 		when(searchHits.iterator()).thenReturn(Arrays.asList(hit1, hit2).iterator());
 
 		SearchResponse searchResponse = mock(SearchResponse.class);