Release version 5.4.6 (2024.1.6).

See #3095

.mvn/jvm.config

@@ -8,3 +8,7 @@
 --add-exports jdk.compiler/com.sun.tools.javac.util=ALL-UNNAMED
 --add-opens jdk.compiler/com.sun.tools.javac.code=ALL-UNNAMED
 --add-opens jdk.compiler/com.sun.tools.javac.comp=ALL-UNNAMED
+--add-opens=java.base/java.util=ALL-UNNAMED
+--add-opens=java.base/java.lang.reflect=ALL-UNNAMED
+--add-opens=java.base/java.text=ALL-UNNAMED
+--add-opens=java.desktop/java.awt.font=ALL-UNNAMED

ci/pipeline.properties

@@ -1,6 +1,6 @@
 # Java versions
-java.main.tag=17.0.13_11-jdk-focal
-java.next.tag=23.0.1_11-jdk-noble
+java.main.tag=17.0.15_6-jdk-focal
+java.next.tag=24.0.1_9-jdk-noble
 
 # Docker container images - standard
 docker.java.main.image=library/eclipse-temurin:${java.main.tag}

pom.xml

@@ -5,12 +5,12 @@
 
 	<groupId>org.springframework.data</groupId>
 	<artifactId>spring-data-elasticsearch</artifactId>
-	<version>5.4.5</version>
+	<version>5.4.6</version>
 
 	<parent>
 		<groupId>org.springframework.data.build</groupId>
 		<artifactId>spring-data-parent</artifactId>
-		<version>3.4.5</version>
+		<version>3.4.6</version>
 	</parent>
 
 	<name>Spring Data Elasticsearch</name>
@@ -18,7 +18,7 @@
 	<url>https://github.com/spring-projects/spring-data-elasticsearch</url>
 
 	<properties>
-		<springdata.commons>3.4.5</springdata.commons>
+		<springdata.commons>3.4.6</springdata.commons>
 
 		<!-- version of the ElasticsearchClient	-->
 		<elasticsearch-java>8.15.5</elasticsearch-java>

src/main/java/org/springframework/data/elasticsearch/client/elc/RequestConverter.java

@@ -1287,11 +1287,15 @@ class RequestConverter extends AbstractQueryProcessor {
 									.timeout(timeStringMs(query.getTimeout())) //
 							;
 
-							if (query.getPageable().isPaged()) {
-								bb //
-										.from((int) query.getPageable().getOffset()) //
-										.size(query.getPageable().getPageSize());
-							}
+							var offset = query.getPageable().isPaged() ? query.getPageable().getOffset() : 0;
+							var pageSize = query.getPageable().isPaged() ? query.getPageable().getPageSize()
+									: INDEX_MAX_RESULT_WINDOW;
+							// if we have both a page size and a max results, we take the min, this is necessary for
+							// searchForStream to work correctly (#3098) as there the page size defines what is
+							// returned in a single request, and the max result determines the total number of
+							// documents returned
+							var size = query.isLimiting() ? Math.min(pageSize, query.getMaxResults()) : pageSize;
+							bb.from((int) offset).size(size);
 
 							if (!isEmpty(query.getFields())) {
 								bb.fields(fb -> {
@@ -1304,10 +1308,6 @@ class RequestConverter extends AbstractQueryProcessor {
 								bb.storedFields(query.getStoredFields());
 							}
 
-							if (query.isLimiting()) {
-								bb.size(query.getMaxResults());
-							}
-
 							if (query.getMinScore() > 0) {
 								bb.minScore((double) query.getMinScore());
 							}
@@ -1460,13 +1460,14 @@ class RequestConverter extends AbstractQueryProcessor {
 			builder.seqNoPrimaryTerm(true);
 		}
 
-		if (query.getPageable().isPaged()) {
-			builder //
-					.from((int) query.getPageable().getOffset()) //
-					.size(query.getPageable().getPageSize());
-		} else {
-			builder.from(0).size(INDEX_MAX_RESULT_WINDOW);
-		}
+		var offset = query.getPageable().isPaged() ? query.getPageable().getOffset() : 0;
+		var pageSize = query.getPageable().isPaged() ? query.getPageable().getPageSize() : INDEX_MAX_RESULT_WINDOW;
+		// if we have both a page size and a max results, we take the min, this is necessary for
+		// searchForStream to work correctly (#3098) as there the page size defines what is
+		// returned in a single request, and the max result determines the total number of
+		// documents returned
+		var size = query.isLimiting() ? Math.min(pageSize, query.getMaxResults()) : pageSize;
+		builder.from((int) offset).size(size);
 
 		if (!isEmpty(query.getFields())) {
 			var fieldAndFormats = query.getFields().stream().map(field -> FieldAndFormat.of(b -> b.field(field))).toList();
@@ -1481,10 +1482,6 @@ class RequestConverter extends AbstractQueryProcessor {
 			addIndicesOptions(builder, query.getIndicesOptions());
 		}
 
-		if (query.isLimiting()) {
-			builder.size(query.getMaxResults());
-		}
-
 		if (query.getMinScore() > 0) {
 			builder.minScore((double) query.getMinScore());
 		}

src/main/java/org/springframework/data/elasticsearch/client/elc/ResponseConverter.java

@@ -500,6 +500,10 @@ class ResponseConverter {
 			builder.withDeleted(response.deleted());
 		}
 
+		if(response.updated() != null) {
+			builder.withUpdated(response.updated());
+		}
+
 		if (response.batches() != null) {
 			builder.withBatches(Math.toIntExact(response.batches()));
 		}

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

@@ -233,6 +233,7 @@ abstract public class AbstractReactiveElasticsearchTemplate
 					.subscribe(new Subscriber<>() {
 						@Nullable private Subscription subscription = null;
 						private final AtomicBoolean upstreamComplete = new AtomicBoolean(false);
+						private final AtomicBoolean onNextHasBeenCalled = new AtomicBoolean(false);
 
 						@Override
 						public void onSubscribe(Subscription subscription) {
@@ -242,6 +243,7 @@ abstract public class AbstractReactiveElasticsearchTemplate
 
 						@Override
 						public void onNext(List<T> entityList) {
+							onNextHasBeenCalled.set(true);
 							saveAll(entityList, index)
 									.map(sink::tryEmitNext)
 									.doOnComplete(() -> {
@@ -267,6 +269,10 @@ abstract public class AbstractReactiveElasticsearchTemplate
 						@Override
 						public void onComplete() {
 							upstreamComplete.set(true);
+							if (!onNextHasBeenCalled.get()) {
+								// this happens when an empty flux is saved
+								sink.tryEmitComplete();
+							}
 						}
 					});
 			return sink.asFlux();

src/main/resources/notice.txt

@@ -1,4 +1,4 @@
-Spring Data Elasticsearch 5.4.5 (2024.1.5)
+Spring Data Elasticsearch 5.4.6 (2024.1.6)
 Copyright (c) [2013-2022] Pivotal Software, Inc.
 
 This product is licensed to you under the Apache License, Version 2.0 (the "License").
@@ -28,5 +28,6 @@ conditions of the subcomponent's license, as noted in the LICENSE file.
 
 
 
+
 
 

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

@@ -1636,7 +1636,9 @@ public abstract class ElasticsearchIntegrationTests {
 				.withParams(Collections.singletonMap("newMessage", messageAfterUpdate)).withAbortOnVersionConflict(true)
 				.build();
 
-		operations.updateByQuery(updateQuery, IndexCoordinates.of(indexNameProvider.indexName()));
+		var byQueryResponse = operations.updateByQuery(updateQuery, IndexCoordinates.of(indexNameProvider.indexName()));
+
+		assertThat(byQueryResponse.getUpdated()).isEqualTo(1);
 
 		SampleEntity indexedEntity = operations.get(documentId, SampleEntity.class,
 				IndexCoordinates.of(indexNameProvider.indexName()));

src/test/java/org/springframework/data/elasticsearch/repository/support/SimpleReactiveElasticsearchRepositoryIntegrationTests.java

@@ -105,6 +105,26 @@ abstract class SimpleReactiveElasticsearchRepositoryIntegrationTests {
 		return operations.exists(id, IndexCoordinates.of(indexNameProvider.indexName()));
 	}
 
+	@Test // #3093
+	@DisplayName("should save all from empty collection")
+	void shouldSaveAllFromEmptyCollection() {
+
+		repository.saveAll(Collections.emptyList())
+				.as(StepVerifier::create)
+				.expectNextCount(0)
+				.verifyComplete();
+	}
+
+	@Test // #3093
+	@DisplayName("should save all from empty flux")
+	void shouldSaveAllFromEmptyFlux() {
+
+		repository.saveAll(Flux.empty())
+				.as(StepVerifier::create)
+				.expectNextCount(0)
+				.verifyComplete();
+	}
+
 	@Test // DATAES-519
 	void saveShouldComputeMultipleEntities() {