Upgrade to Elasticsearch 8.4.2.

Original Pull Request #2301
Closes #2284

pom.xml

@@ -21,15 +21,15 @@
 		<springdata.commons>3.0.0-SNAPSHOT</springdata.commons>
 
 		<!-- version of the RestHighLevelClient	-->
-		<elasticsearch-rhlc>7.17.5</elasticsearch-rhlc>
+		<elasticsearch-rhlc>7.17.6</elasticsearch-rhlc>
 		<!-- version of the new ElasticsearchClient	-->
-		<elasticsearch-java>8.3.3</elasticsearch-java>
+		<elasticsearch-java>8.4.2</elasticsearch-java>
 
 		<log4j>2.18.0</log4j>
 		<netty>4.1.65.Final</netty>
 
 		<blockhound-junit>1.0.6.RELEASE</blockhound-junit>
-		<hoverfly>0.14.2</hoverfly>
+		<hoverfly>0.14.3</hoverfly>
 		<jsonassert>1.5.1</jsonassert>
 		<testcontainers>1.17.3</testcontainers>
 		<wiremock>2.33.2</wiremock>

src/main/asciidoc/preface.adoc

@@ -37,7 +37,7 @@ built and tested.
 [cols="^,^,^,^,^",options="header"]
 |===
 | Spring Data Release Train | Spring Data Elasticsearch | Elasticsearch | Spring Framework | Spring Boot
-| 2022.0 (Turing) | 5.0.x | 8.3.3 | 6.0.x | 3.0.x?
+| 2022.0 (Turing) | 5.0.x | 8.4.2 | 6.0.x | 3.0.x?
 | 2021.2 (Raj) | 4.4.x | 7.17.3 | 5.3.x | 2.7.x
 | 2021.1 (Q) | 4.3.x | 7.15.2 | 5.3.x | 2.6.x
 | 2021.0 (Pascal) | 4.2.xfootnote:oom[Out of maintenance] | 7.12.0 | 5.3.x | 2.5.x

src/main/asciidoc/reference/elasticsearch-migration-guide-4.4-5.0.adoc

@@ -151,4 +151,4 @@ The old deprecated `RestHighLevelClient` can still be used, but you will need to
 ----
 ====
 
-Make sure to specify the version 7.17.5 explicitly, otherwise maven will resolve to 8.3.3, and this does not exist.
+Make sure to specify the version 7.17.6 explicitly, otherwise maven will resolve to 8.4.2, and this does not exist.

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

@@ -29,6 +29,7 @@ import java.net.InetSocketAddress;
 import java.time.Duration;
 import java.util.Arrays;
 import java.util.List;
+import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
@@ -250,23 +251,26 @@ public final class ElasticsearchClients {
 		TransportOptions.Builder transportOptionsBuilder = transportOptions != null ? transportOptions.toBuilder()
 				: new RestClientOptions(RequestOptions.DEFAULT).toBuilder();
 
+		ContentType jsonContentType = Version.VERSION == null ? ContentType.APPLICATION_JSON
+				: ContentType.create("application/vnd.elasticsearch+json",
+						new BasicNameValuePair("compatible-with", String.valueOf(Version.VERSION.major())));
+
+		Consumer<String> setHeaderIfNotPresent = header -> {
+			if (transportOptionsBuilder.build().headers().stream() //
+					.noneMatch((h) -> h.getKey().equalsIgnoreCase(header))) {
 				// need to add the compatibility header, this is only done automatically when not passing in custom options.
 				// code copied from RestClientTransport as it is not available outside the package
-		ContentType jsonContentType = null;
+				transportOptionsBuilder.addHeader(header, jsonContentType.toString());
-		if (Version.VERSION == null) {
-			jsonContentType = ContentType.APPLICATION_JSON;
-		} else {
-			jsonContentType = ContentType.create("application/vnd.elasticsearch+json",
-					new BasicNameValuePair("compatible-with", String.valueOf(Version.VERSION.major())));
 			}
-		transportOptionsBuilder.addHeader("Accept", jsonContentType.toString());
+		};
+
+		setHeaderIfNotPresent.accept("Content-Type");
+		setHeaderIfNotPresent.accept("Accept");
 
 		TransportOptions transportOptionsWithHeader = transportOptionsBuilder
 				.addHeader(X_SPRING_DATA_ELASTICSEARCH_CLIENT, clientType).build();
 
-		ElasticsearchTransport transport = new RestClientTransport(restClient, new JacksonJsonpMapper(),
+		return new RestClientTransport(restClient, new JacksonJsonpMapper(), transportOptionsWithHeader);
-				transportOptionsWithHeader);
-		return transport;
 	}
 
 	private static List<String> formattedHosts(List<InetSocketAddress> hosts, boolean useSsl) {
@@ -333,7 +337,7 @@ public final class ElasticsearchClients {
 							+ ((header.getName().equals("Authorization")) ? ": *****" : ": " + header.getValue()))
 					.collect(Collectors.joining(", ", "[", "]"));
 
-			// no way of logging the body, in this callback, it is not read yset, later there is no callback possibility in
+			// no way of logging the body, in this callback, it is not read yet, later there is no callback possibility in
 			// RestClient or RestClientTransport
 			ClientLogger.logRawResponse(logId, response.getStatusLine().getStatusCode(), headers);
 		}

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

@@ -808,7 +808,7 @@ class RequestConverter {
 				.maxDocs(reindexRequest.getMaxDocs()).waitForCompletion(waitForCompletion) //
 				.refresh(reindexRequest.getRefresh()) //
 				.requireAlias(reindexRequest.getRequireAlias()) //
-				.requestsPerSecond(reindexRequest.getRequestsPerSecond()) //
+				.requestsPerSecond(toFloat(reindexRequest.getRequestsPerSecond())) //
 				.slices(slices(reindexRequest.getSlices()));
 
 		return builder.build();
@@ -948,8 +948,7 @@ class RequestConverter {
 					.script(getScript(updateQuery.getScriptData())) //
 					.maxDocs(updateQuery.getMaxDocs() != null ? Long.valueOf(updateQuery.getMaxDocs()) : null) //
 					.pipeline(updateQuery.getPipeline()) //
-					.requestsPerSecond(
+					.requestsPerSecond(updateQuery.getRequestsPerSecond()) //
-							updateQuery.getRequestsPerSecond() != null ? updateQuery.getRequestsPerSecond().longValue() : null) //
 					.slices(slices(updateQuery.getSlices() != null ? Long.valueOf(updateQuery.getSlices()) : null));
 
 			if (updateQuery.getAbortOnVersionConflict() != null) {
@@ -1107,7 +1106,7 @@ class RequestConverter {
 							query.getRescorerQueries().forEach(rescorerQuery -> bb.rescore(getRescore(rescorerQuery)));
 
 							if (!query.getRuntimeFields().isEmpty()) {
-								Map<String, List<RuntimeField>> runtimeMappings = new HashMap<>();
+								Map<String, RuntimeField> runtimeMappings = new HashMap<>();
 								query.getRuntimeFields().forEach(runtimeField -> {
 									RuntimeField esRuntimeField = RuntimeField.of(rt -> {
 										RuntimeField.Builder builder = rt
@@ -1119,7 +1118,7 @@ class RequestConverter {
 										}
 										return builder;
 									});
-									runtimeMappings.put(runtimeField.getName(), Collections.singletonList(esRuntimeField));
+									runtimeMappings.put(runtimeField.getName(), esRuntimeField);
 								});
 								bb.runtimeMappings(runtimeMappings);
 							}
@@ -1251,12 +1250,11 @@ class RequestConverter {
 
 		if (!query.getRuntimeFields().isEmpty()) {
 
-			Map<String, List<RuntimeField>> runtimeMappings = new HashMap<>();
+			Map<String, RuntimeField> runtimeMappings = new HashMap<>();
 			query.getRuntimeFields()
-					.forEach(runtimeField -> runtimeMappings.put(runtimeField.getName(),
+					.forEach(runtimeField -> runtimeMappings.put(runtimeField.getName(), RuntimeField.of(rt -> rt //
-							Collections.singletonList(RuntimeField.of(rt -> rt //
 							.type(RuntimeFieldType._DESERIALIZER.parse(runtimeField.getType())) //
-									.script(s -> s.inline(is -> is.source(runtimeField.getScript())))))));
+							.script(s -> s.inline(is -> is.source(runtimeField.getScript()))))));
 			builder.runtimeMappings(runtimeMappings);
 		}
 

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

@@ -90,7 +90,7 @@ class ResponseConverter {
 				.withNumberOfPendingTasks(healthResponse.numberOfPendingTasks()) //
 				.withRelocatingShards(healthResponse.relocatingShards()) //
 				.withStatus(healthResponse.status().toString()) //
-				.withTaskMaxWaitingTimeMillis(healthResponse.taskMaxWaitingInQueueMillis().toEpochMilli()) //
+				.withTaskMaxWaitingTimeMillis(healthResponse.taskMaxWaitingInQueueMillis()) //
 				.withTimedOut(healthResponse.timedOut()) //
 				.withUnassignedShards(healthResponse.unassignedShards()) //
 				.build(); //
@@ -266,7 +266,7 @@ class ResponseConverter {
 
 		// noinspection ConstantConditions
 		return ReindexResponse.builder() //
-				.withTook(timeToLong(reindexResponse.took())) //
+				.withTook(reindexResponse.took()) //
 				.withTimedOut(reindexResponse.timedOut()) //
 				.withTotal(reindexResponse.total()) //
 				.withCreated(reindexResponse.created()) //
@@ -277,9 +277,10 @@ class ResponseConverter {
 				.withNoops(reindexResponse.noops()) //
 				.withBulkRetries(reindexResponse.retries().bulk()) //
 				.withSearchRetries(reindexResponse.retries().search()) //
-				.withThrottledMillis(reindexResponse.throttledMillis().toEpochMilli()) //
+				.withThrottledMillis(reindexResponse.throttledMillis()) //
 				.withRequestsPerSecond(reindexResponse.requestsPerSecond()) //
-				.withThrottledUntilMillis(reindexResponse.throttledUntilMillis().toEpochMilli()).withFailures(failures) //
+				.withThrottledUntilMillis(reindexResponse.throttledUntilMillis()) //
+			.withFailures(failures) //
 				.build();
 	}
 

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

@@ -319,4 +319,15 @@ final class TypeUtils {
 		}
 	}
 
+	/**
+	 * Converts a Long to a Float, returning null if the input is null.
+	 *
+	 * @param value the long value
+	 * @return a FLoat with the given value
+	 * @since 5.0
+	 */
+	@Nullable
+	static Float toFloat(@Nullable Long value) {
+		return value != null ? Float.valueOf(value) : null;
+	}
 }

src/test/java/org/springframework/data/elasticsearch/client/elc/DevTests.java

@@ -39,7 +39,6 @@ import reactor.core.publisher.Mono;
 import java.io.IOException;
 import java.time.LocalDateTime;
 import java.time.format.DateTimeFormatter;
-import java.util.Collections;
 import java.util.Objects;
 import java.util.function.Function;
 
@@ -108,7 +107,7 @@ public class DevTests {
 
 		client.search(sr -> sr //
 				.index(index) //
-				.runtimeMappings("priceWithTax", Collections.singletonList(runtimeField)), //
+				.runtimeMappings("priceWithTax", runtimeField), //
 				Person.class); //
 	}
 

src/test/java/org/springframework/data/elasticsearch/repositories/cdi/ElasticsearchOperationsProducer.java

@@ -68,6 +68,12 @@ class ElasticsearchOperationsProducer {
 			configurationBuilder.withBasicAuth(user, password);
 		}
 
+		String proxy = System.getenv("DATAES_ELASTICSEARCH_PROXY");
+
+		if (hasText(proxy)) {
+			configurationBuilder.withProxy(proxy);
+		}
+
 		ClientConfiguration clientConfiguration = configurationBuilder //
 				.build();
 

src/test/java/org/springframework/data/elasticsearch/repository/query/keywords/QueryKeywordsIntegrationTests.java

@@ -23,6 +23,7 @@ import java.util.List;
 import java.util.stream.Collectors;
 
 import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.DisplayName;
 import org.junit.jupiter.api.Order;
 import org.junit.jupiter.api.Test;
@@ -265,6 +266,7 @@ abstract class QueryKeywordsIntegrationTests {
 		assertThat(products).isEmpty();
 	}
 
+	@Disabled("issue #2300, Elasticsearch bug https://github.com/elastic/elasticsearch/issues/89760")
 	@Test // #1909
 	@DisplayName("should find by property exists")
 	void shouldFindByPropertyExists() {
@@ -274,6 +276,7 @@ abstract class QueryKeywordsIntegrationTests {
 		assertThat(searchHits.getTotalHits()).isEqualTo(6);
 	}
 
+	@Disabled("issue #2300, Elasticsearch bug https://github.com/elastic/elasticsearch/issues/89760")
 	@Test // #1909
 	@DisplayName("should find by property is not null")
 	void shouldFindByPropertyIsNotNull() {

src/test/resources/testcontainers-elasticsearch.properties

@@ -15,7 +15,7 @@
 #
 #
 sde.testcontainers.image-name=docker.elastic.co/elasticsearch/elasticsearch
-sde.testcontainers.image-version=8.3.3
+sde.testcontainers.image-version=8.4.2
 #
 #
 # needed as we do a DELETE /* at the end of the tests, will be required from 8.0 on, produces a warning since 7.13