Auto-configure the new Elasticsearch clients

This commit introduces auto-configuration for the new Elasticsearch
clients that are based upon their new Java client. The new Java
client builds on top of their existing low-level REST client,
replacing the high-level REST client which has been deprecated.
As part of introducing support for the new Elasticsearch client,
the auto-configuration for the templates (both imperative and
reactive) provided by Spring Data has also been updated to use the
new templates that build upon the new Java client.

As part of these changes, support for the high-level REST client and
the old Spring Data Elasticsearch templates has been removed. One
significant change is that the new reactive template is no longer
based on WebClient. As a result, the WebClient-specific configuration
property has been removed.

Closes gh-30647
Closes gh-28597
Closes gh-31755

Author: wilkinsona

spring-boot-project/spring-boot-actuator-autoconfigure/build.gradle

@@ -103,7 +103,6 @@ dependencies {
 	optional("org.eclipse.jetty:jetty-server") {
 		exclude group: "org.eclipse.jetty.toolchain", module: "jetty-jakarta-servlet-api"
 	}
-	optional("org.elasticsearch:elasticsearch")
 	optional("org.elasticsearch.client:elasticsearch-rest-client") {
 		exclude group: "commons-logging", module: "commons-logging"
 	}
@@ -168,9 +167,6 @@ dependencies {
 	testImplementation("org.eclipse.jetty:jetty-webapp") {
 		exclude group: "org.eclipse.jetty.toolchain", module: "jetty-jakarta-servlet-api"
 	}
-	testImplementation("org.elasticsearch.client:elasticsearch-rest-high-level-client") {
-		exclude(group: "commons-logging", module: "commons-logging")
-	}
 	testImplementation("org.hamcrest:hamcrest")
 	testImplementation("org.hsqldb:hsqldb")
 	testImplementation("org.junit.jupiter:junit-jupiter")

spring-boot-project/spring-boot-actuator-autoconfigure/src/main/java/org/springframework/boot/actuate/autoconfigure/elasticsearch/ElasticSearchReactiveHealthContributorAutoConfiguration.java renamed to spring-boot-project/spring-boot-actuator-autoconfigure/src/main/java/org/springframework/boot/actuate/autoconfigure/elasticsearch/ElasticsearchReactiveHealthContributorAutoConfiguration.java

@@ -31,7 +31,7 @@
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.autoconfigure.elasticsearch.ReactiveElasticsearchClientAutoConfiguration;
 import org.springframework.context.annotation.Bean;
-import org.springframework.data.elasticsearch.client.erhlc.ReactiveElasticsearchClient;
+import org.springframework.data.elasticsearch.client.elc.ReactiveElasticsearchClient;
 
 /**
  * {@link EnableAutoConfiguration Auto-configuration} for
@@ -45,7 +45,7 @@
 @ConditionalOnClass({ ReactiveElasticsearchClient.class, Flux.class })
 @ConditionalOnBean(ReactiveElasticsearchClient.class)
 @ConditionalOnEnabledHealthIndicator("elasticsearch")
-public class ElasticSearchReactiveHealthContributorAutoConfiguration extends
+public class ElasticsearchReactiveHealthContributorAutoConfiguration extends
 		CompositeReactiveHealthContributorConfiguration<ElasticsearchReactiveHealthIndicator, ReactiveElasticsearchClient> {
 
 	@Bean

spring-boot-project/spring-boot-actuator-autoconfigure/src/main/java/org/springframework/boot/actuate/autoconfigure/elasticsearch/ElasticSearchRestHealthContributorAutoConfiguration.java renamed to spring-boot-project/spring-boot-actuator-autoconfigure/src/main/java/org/springframework/boot/actuate/autoconfigure/elasticsearch/ElasticsearchRestHealthContributorAutoConfiguration.java

@@ -43,7 +43,7 @@
 @ConditionalOnClass(RestClient.class)
 @ConditionalOnBean(RestClient.class)
 @ConditionalOnEnabledHealthIndicator("elasticsearch")
-public class ElasticSearchRestHealthContributorAutoConfiguration
+public class ElasticsearchRestHealthContributorAutoConfiguration
 		extends CompositeHealthContributorConfiguration<ElasticsearchRestClientHealthIndicator, RestClient> {
 
 	@Bean

spring-boot-project/spring-boot-actuator-autoconfigure/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports

@@ -14,8 +14,8 @@ org.springframework.boot.actuate.autoconfigure.context.properties.ConfigurationP
 org.springframework.boot.actuate.autoconfigure.context.ShutdownEndpointAutoConfiguration
 org.springframework.boot.actuate.autoconfigure.couchbase.CouchbaseHealthContributorAutoConfiguration
 org.springframework.boot.actuate.autoconfigure.couchbase.CouchbaseReactiveHealthContributorAutoConfiguration
-org.springframework.boot.actuate.autoconfigure.elasticsearch.ElasticSearchReactiveHealthContributorAutoConfiguration
-org.springframework.boot.actuate.autoconfigure.elasticsearch.ElasticSearchRestHealthContributorAutoConfiguration
+org.springframework.boot.actuate.autoconfigure.elasticsearch.ElasticsearchReactiveHealthContributorAutoConfiguration
+org.springframework.boot.actuate.autoconfigure.elasticsearch.ElasticsearchRestHealthContributorAutoConfiguration
 org.springframework.boot.actuate.autoconfigure.endpoint.EndpointAutoConfiguration
 org.springframework.boot.actuate.autoconfigure.endpoint.jmx.JmxEndpointAutoConfiguration
 org.springframework.boot.actuate.autoconfigure.endpoint.web.WebEndpointAutoConfiguration

spring-boot-project/spring-boot-actuator-autoconfigure/src/test/java/org/springframework/boot/actuate/autoconfigure/elasticsearch/ElasticsearchReactiveHealthContributorAutoConfigurationTests.java

@@ -30,7 +30,7 @@
 import static org.assertj.core.api.Assertions.assertThat;
 
 /**
- * Tests for {@link ElasticSearchReactiveHealthContributorAutoConfiguration}.
+ * Tests for {@link ElasticsearchReactiveHealthContributorAutoConfiguration}.
  *
  * @author Aleksander Lech
  */
@@ -39,7 +39,7 @@ class ElasticsearchReactiveHealthContributorAutoConfigurationTests {
 	private final ApplicationContextRunner contextRunner = new ApplicationContextRunner()
 			.withConfiguration(AutoConfigurations.of(ElasticsearchDataAutoConfiguration.class,
 					ReactiveElasticsearchClientAutoConfiguration.class, ElasticsearchRestClientAutoConfiguration.class,
-					ElasticSearchReactiveHealthContributorAutoConfiguration.class,
+					ElasticsearchReactiveHealthContributorAutoConfiguration.class,
 					HealthContributorAutoConfiguration.class));
 
 	@Test
@@ -51,7 +51,7 @@ void runShouldCreateIndicator() {
 	@Test
 	void runWithRegularIndicatorShouldOnlyCreateReactiveIndicator() {
 		this.contextRunner
-				.withConfiguration(AutoConfigurations.of(ElasticSearchRestHealthContributorAutoConfiguration.class))
+				.withConfiguration(AutoConfigurations.of(ElasticsearchRestHealthContributorAutoConfiguration.class))
 				.run((context) -> assertThat(context).hasSingleBean(ElasticsearchReactiveHealthIndicator.class)
 						.hasBean("elasticsearchHealthContributor")
 						.doesNotHaveBean(ElasticsearchRestClientHealthIndicator.class));

spring-boot-project/spring-boot-actuator-autoconfigure/src/test/java/org/springframework/boot/actuate/autoconfigure/elasticsearch/ElasticSearchRestHealthContributorAutoConfigurationTests.java renamed to spring-boot-project/spring-boot-actuator-autoconfigure/src/test/java/org/springframework/boot/actuate/autoconfigure/elasticsearch/ElasticsearchRestHealthContributorAutoConfigurationTests.java

@@ -32,16 +32,16 @@
 import static org.assertj.core.api.Assertions.assertThat;
 
 /**
- * Tests for {@link ElasticSearchRestHealthContributorAutoConfiguration}.
+ * Tests for {@link ElasticsearchRestHealthContributorAutoConfiguration}.
  *
  * @author Filip Hrisafov
  * @author Andy Wilkinson
  */
-class ElasticSearchRestHealthContributorAutoConfigurationTests {
+class ElasticsearchRestHealthContributorAutoConfigurationTests {
 
 	private final ApplicationContextRunner contextRunner = new ApplicationContextRunner()
 			.withConfiguration(AutoConfigurations.of(ElasticsearchRestClientAutoConfiguration.class,
-					ElasticSearchRestHealthContributorAutoConfiguration.class,
+					ElasticsearchRestHealthContributorAutoConfiguration.class,
 					HealthContributorAutoConfiguration.class));
 
 	@Test
@@ -51,29 +51,19 @@ void runShouldCreateIndicator() {
 	}
 
 	@Test
-	@SuppressWarnings("deprecation")
-	void runWithoutRestHighLevelClientAndWithoutRestClientShouldNotCreateIndicator() {
-		this.contextRunner
-				.withClassLoader(
-						new FilteredClassLoader(org.elasticsearch.client.RestHighLevelClient.class, RestClient.class))
+	void runWithoutRestClientShouldNotCreateIndicator() {
+		this.contextRunner.withClassLoader(new FilteredClassLoader(RestClient.class))
 				.run((context) -> assertThat(context).doesNotHaveBean(ElasticsearchRestClientHealthIndicator.class)
 						.doesNotHaveBean("elasticsearchHealthContributor"));
 	}
 
 	@Test
-	void runWithoutRestHighLevelClientAndWithRestClientShouldCreateIndicator() {
+	void runWithRestClientShouldCreateIndicator() {
 		this.contextRunner.withUserConfiguration(CustomRestClientConfiguration.class)
 				.run((context) -> assertThat(context).hasSingleBean(ElasticsearchRestClientHealthIndicator.class)
 						.hasBean("elasticsearchHealthContributor"));
 	}
 
-	@Test
-	void runWithRestHighLevelClientAndWithRestClientShouldCreateIndicator() {
-		this.contextRunner.withUserConfiguration(CustomRestHighClientConfiguration.class)
-				.run((context) -> assertThat(context).hasSingleBean(ElasticsearchRestClientHealthIndicator.class)
-						.hasBean("elasticsearchHealthContributor"));
-	}
-
 	@Test
 	void runWhenDisabledShouldNotCreateIndicator() {
 		this.contextRunner.withPropertyValues("management.health.elasticsearch.enabled:false")
@@ -91,20 +81,4 @@ RestClient customRestClient(RestClientBuilder builder) {
 
 	}
 
-	@Configuration(proxyBeanMethods = false)
-	@SuppressWarnings("deprecation")
-	static class CustomRestHighClientConfiguration {
-
-		@Bean
-		org.elasticsearch.client.RestHighLevelClient customRestHighClient(RestClientBuilder builder) {
-			return new org.elasticsearch.client.RestHighLevelClient(builder);
-		}
-
-		@Bean
-		RestClient customClient(org.elasticsearch.client.RestHighLevelClient restHighLevelClient) {
-			return restHighLevelClient.getLowLevelClient();
-		}
-
-	}
-
 }

spring-boot-project/spring-boot-actuator/build.gradle

@@ -43,13 +43,9 @@ dependencies {
 	optional("org.eclipse.jetty:jetty-server") {
 		exclude(group: "org.eclipse.jetty.toolchain", module: "jetty-jakarta-servlet-api")
 	}
-	optional("org.elasticsearch:elasticsearch")
 	optional("org.elasticsearch.client:elasticsearch-rest-client") {
 		exclude(group: "commons-logging", module: "commons-logging")
 	}
-	optional("org.elasticsearch.client:elasticsearch-rest-high-level-client") {
-		exclude(group: "commons-logging", module: "commons-logging")
-	}
 	optional("org.flywaydb:flyway-core")
 	optional("org.hibernate.validator:hibernate-validator")
 	optional("org.influxdb:influxdb-java")

spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/elasticsearch/ElasticsearchReactiveHealthIndicator.java

@@ -16,20 +16,15 @@
 
 package org.springframework.boot.actuate.elasticsearch;
 
-import java.util.Map;
-
+import co.elastic.clients.elasticsearch._types.HealthStatus;
+import co.elastic.clients.elasticsearch.cluster.HealthResponse;
 import reactor.core.publisher.Mono;
 
 import org.springframework.boot.actuate.health.AbstractReactiveHealthIndicator;
 import org.springframework.boot.actuate.health.Health;
 import org.springframework.boot.actuate.health.HealthIndicator;
 import org.springframework.boot.actuate.health.Status;
-import org.springframework.core.ParameterizedTypeReference;
-import org.springframework.data.elasticsearch.client.erhlc.ReactiveElasticsearchClient;
-import org.springframework.http.HttpStatus;
-import org.springframework.http.HttpStatusCode;
-import org.springframework.web.reactive.function.client.ClientResponse;
-import org.springframework.web.reactive.function.client.WebClient;
+import org.springframework.data.elasticsearch.client.elc.ReactiveElasticsearchClient;
 
 /**
  * {@link HealthIndicator} for an Elasticsearch cluster using a
@@ -42,11 +37,6 @@
  */
 public class ElasticsearchReactiveHealthIndicator extends AbstractReactiveHealthIndicator {
 
-	private static final ParameterizedTypeReference<Map<String, Object>> STRING_OBJECT_MAP = new ParameterizedTypeReference<>() {
-	};
-
-	private static final String RED_STATUS = "red";
-
 	private final ReactiveElasticsearchClient client;
 
 	public ElasticsearchReactiveHealthIndicator(ReactiveElasticsearchClient client) {
@@ -56,32 +46,33 @@ public ElasticsearchReactiveHealthIndicator(ReactiveElasticsearchClient client)
 
 	@Override
 	protected Mono<Health> doHealthCheck(Health.Builder builder) {
-		return this.client.execute((webClient) -> getHealth(builder, webClient));
-	}
-
-	private Mono<Health> getHealth(Health.Builder builder, WebClient webClient) {
-		return webClient.get().uri("/_cluster/health/").exchangeToMono((response) -> doHealthCheck(builder, response));
+		return this.client.cluster().health((b) -> b).map((response) -> processResponse(builder, response));
 	}
 
-	private Mono<Health> doHealthCheck(Health.Builder builder, ClientResponse response) {
-		HttpStatusCode httpStatusCode = response.statusCode();
-		HttpStatus httpStatus = HttpStatus.resolve(httpStatusCode.value());
-		if (httpStatusCode.is2xxSuccessful()) {
-			return response.bodyToMono(STRING_OBJECT_MAP).map((body) -> getHealth(builder, body));
-		}
-		builder.down();
-		builder.withDetail("statusCode", httpStatusCode.value());
-		if (httpStatus != null) {
-			builder.withDetail("reasonPhrase", httpStatus.getReasonPhrase());
+	private Health processResponse(Health.Builder builder, HealthResponse response) {
+		if (!response.timedOut()) {
+			HealthStatus status = response.status();
+			builder.status((HealthStatus.Red == status) ? Status.OUT_OF_SERVICE : Status.UP);
+			builder.withDetail("cluster_name", response.clusterName());
+			builder.withDetail("status", response.status().jsonValue());
+			builder.withDetail("timed_out", response.timedOut());
+			builder.withDetail("number_of_nodes", response.numberOfNodes());
+			builder.withDetail("number_of_data_nodes", response.numberOfDataNodes());
+			builder.withDetail("active_primary_shards", response.activePrimaryShards());
+			builder.withDetail("active_shards", response.activeShards());
+			builder.withDetail("relocating_shards", response.relocatingShards());
+			builder.withDetail("initializing_shards", response.initializingShards());
+			builder.withDetail("unassigned_shards", response.unassignedShards());
+			builder.withDetail("delayed_unassigned_shards", response.delayedUnassignedShards());
+			builder.withDetail("number_of_pending_tasks", response.numberOfPendingTasks());
+			builder.withDetail("number_of_in_flight_fetch", response.numberOfInFlightFetch());
+			builder.withDetail("task_max_waiting_in_queue_millis",
+					response.taskMaxWaitingInQueueMillis().toEpochMilli());
+			builder.withDetail("active_shards_percent_as_number",
+					Double.parseDouble(response.activeShardsPercentAsNumber()));
+			return builder.build();
 		}
-		return response.releaseBody().thenReturn(builder.build());
-	}
-
-	private Health getHealth(Health.Builder builder, Map<String, Object> body) {
-		String status = (String) body.get("status");
-		builder.status(RED_STATUS.equals(status) ? Status.OUT_OF_SERVICE : Status.UP);
-		builder.withDetails(body);
-		return builder.build();
+		return builder.down().build();
 	}
 
 }

spring-boot-project/spring-boot-actuator/src/test/java/org/springframework/boot/actuate/elasticsearch/ElasticsearchReactiveHealthIndicatorTests.java

@@ -19,16 +19,20 @@
 import java.time.Duration;
 import java.util.Map;
 
+import co.elastic.clients.json.jackson.JacksonJsonpMapper;
+import co.elastic.clients.transport.rest_client.RestClientTransport;
 import okhttp3.mockwebserver.MockResponse;
 import okhttp3.mockwebserver.MockWebServer;
+import org.apache.http.HttpHost;
+import org.elasticsearch.client.ResponseException;
+import org.elasticsearch.client.RestClient;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
 import org.springframework.boot.actuate.health.Health;
 import org.springframework.boot.actuate.health.Status;
-import org.springframework.data.elasticsearch.client.ClientConfiguration;
-import org.springframework.data.elasticsearch.client.erhlc.ReactiveElasticsearchClient;
+import org.springframework.data.elasticsearch.client.elc.ReactiveElasticsearchClient;
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.MediaType;
@@ -50,13 +54,13 @@ class ElasticsearchReactiveHealthIndicatorTests {
 
 	private ElasticsearchReactiveHealthIndicator healthIndicator;
 
-	@SuppressWarnings("deprecation")
 	@BeforeEach
 	void setup() throws Exception {
 		this.server = new MockWebServer();
 		this.server.start();
-		ReactiveElasticsearchClient client = org.springframework.data.elasticsearch.client.erhlc.DefaultReactiveElasticsearchClient
-				.create(ClientConfiguration.create(this.server.getHostName() + ":" + this.server.getPort()));
+		ReactiveElasticsearchClient client = new ReactiveElasticsearchClient(new RestClientTransport(
+				RestClient.builder(HttpHost.create(this.server.getHostName() + ":" + this.server.getPort())).build(),
+				new JacksonJsonpMapper()));
 		this.healthIndicator = new ElasticsearchReactiveHealthIndicator(client);
 	}
 
@@ -86,20 +90,15 @@ void elasticsearchIsDown() throws Exception {
 		this.server.shutdown();
 		Health health = this.healthIndicator.health().block(TIMEOUT);
 		assertThat(health.getStatus()).isEqualTo(Status.DOWN);
-		assertThat(health.getDetails().get("error")).asString()
-				.contains("org.springframework.data.elasticsearch.client.NoReachableHostException");
+		assertThat(health.getDetails().get("error")).asString().contains("Connection refused");
 	}
 
 	@Test
 	void elasticsearchIsDownByResponseCode() {
-		// first enqueue an OK response since the HostChecker first sends a HEAD request
-		// to "/"
-		this.server.enqueue(new MockResponse().setResponseCode(HttpStatus.OK.value()));
 		this.server.enqueue(new MockResponse().setResponseCode(HttpStatus.INTERNAL_SERVER_ERROR.value()));
 		Health health = this.healthIndicator.health().block(TIMEOUT);
 		assertThat(health.getStatus()).isEqualTo(Status.DOWN);
-		assertThat(health.getDetails().get("statusCode")).asString().isEqualTo("500");
-		assertThat(health.getDetails().get("reasonPhrase")).asString().isEqualTo("Internal Server Error");
+		assertThat(health.getDetails().get("error")).asString().startsWith(ResponseException.class.getName());
 	}
 
 	@Test
@@ -116,32 +115,25 @@ private void assertHealthDetailsWithStatus(Map<String, Object> details, String s
 				entry("active_primary_shards", 0), entry("active_shards", 0), entry("relocating_shards", 0),
 				entry("initializing_shards", 0), entry("unassigned_shards", 0), entry("delayed_unassigned_shards", 0),
 				entry("number_of_pending_tasks", 0), entry("number_of_in_flight_fetch", 0),
-				entry("task_max_waiting_in_queue_millis", 0), entry("active_shards_percent_as_number", 100.0));
+				entry("task_max_waiting_in_queue_millis", 0L), entry("active_shards_percent_as_number", 100.0));
 	}
 
 	private void setupMockResponse(int responseCode, String status) {
-		// first enqueue an OK response since the HostChecker first sends a HEAD request
-		// to "/"
-		this.server.enqueue(new MockResponse());
-		MockResponse mockResponse = new MockResponse().setResponseCode(HttpStatus.valueOf(responseCode).value())
-				.setBody(createJsonResult(responseCode, status))
-				.setHeader(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE);
+		MockResponse mockResponse = new MockResponse().setBody(createJsonResult(status))
+				.setHeader(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE)
+				.setHeader("X-Elastic-Product", "Elasticsearch");
 		this.server.enqueue(mockResponse);
 	}
 
-	private String createJsonResult(int responseCode, String status) {
-		if (responseCode == 200) {
-			return String.format(
-					"{\"cluster_name\":\"elasticsearch\","
-							+ "\"status\":\"%s\",\"timed_out\":false,\"number_of_nodes\":1,"
-							+ "\"number_of_data_nodes\":1,\"active_primary_shards\":0,"
-							+ "\"active_shards\":0,\"relocating_shards\":0,\"initializing_shards\":0,"
-							+ "\"unassigned_shards\":0,\"delayed_unassigned_shards\":0,"
-							+ "\"number_of_pending_tasks\":0,\"number_of_in_flight_fetch\":0,"
-							+ "\"task_max_waiting_in_queue_millis\":0,\"active_shards_percent_as_number\":100.0}",
-					status);
-		}
-		return "{\n  \"error\": \"Server Error\",\n  \"status\": " + responseCode + "\n}";
+	private String createJsonResult(String status) {
+		return String.format(
+				"{\"cluster_name\":\"elasticsearch\"," + "\"status\":\"%s\",\"timed_out\":false,\"number_of_nodes\":1,"
+						+ "\"number_of_data_nodes\":1,\"active_primary_shards\":0,"
+						+ "\"active_shards\":0,\"relocating_shards\":0,\"initializing_shards\":0,"
+						+ "\"unassigned_shards\":0,\"delayed_unassigned_shards\":0,"
+						+ "\"number_of_pending_tasks\":0,\"number_of_in_flight_fetch\":0,"
+						+ "\"task_max_waiting_in_queue_millis\":0,\"active_shards_percent_as_number\":100.0}",
+				status);
 	}
 
 }

spring-boot-project/spring-boot-autoconfigure/build.gradle

@@ -12,6 +12,9 @@ description = "Spring Boot AutoConfigure"
 dependencies {
 	api(project(":spring-boot-project:spring-boot"))
 
+	optional("co.elastic.clients:elasticsearch-java") {
+		exclude group: "commons-logging", module: "commons-logging"
+	}
 	optional("com.fasterxml.jackson.core:jackson-databind")
 	optional("com.fasterxml.jackson.dataformat:jackson-dataformat-cbor")
 	optional("com.fasterxml.jackson.dataformat:jackson-dataformat-xml")
@@ -99,9 +102,6 @@ dependencies {
 	optional("org.elasticsearch.client:elasticsearch-rest-client-sniffer") {
 		exclude group: "commons-logging", module: "commons-logging"
 	}
-	optional("org.elasticsearch.client:elasticsearch-rest-high-level-client") {
-		exclude group: "commons-logging", module: "commons-logging"
-	}
 	optional("org.flywaydb:flyway-core")
 	optional("org.flywaydb:flyway-sqlserver")
 	optional("org.freemarker:freemarker")