Provide a configuration property for the maximum size of the in-memory buffer used by the reactive Elasticsearch REST client

spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/data/elasticsearch/ReactiveRestClientAutoConfiguration.java

@@ -29,6 +29,8 @@
 import org.springframework.data.elasticsearch.client.reactive.ReactiveElasticsearchClient;
 import org.springframework.data.elasticsearch.client.reactive.ReactiveRestClients;
 import org.springframework.http.HttpHeaders;
+import org.springframework.util.unit.DataSize;
+import org.springframework.web.reactive.function.client.ExchangeStrategies;
 import org.springframework.web.reactive.function.client.WebClient;
 
 /**
@@ -52,6 +54,7 @@ public ClientConfiguration clientConfiguration(ReactiveRestClientProperties prop
 			builder.usingSsl();
 		}
 		configureTimeouts(builder, properties);
+		configureWebClient(builder, properties);
 		return builder.build();
 	}
 
@@ -67,6 +70,19 @@ private void configureTimeouts(ClientConfiguration.TerminalClientConfigurationBu
 		});
 	}
 
+	private void configureWebClient(ClientConfiguration.TerminalClientConfigurationBuilder builder,
+			ReactiveRestClientProperties properties) {
+		PropertyMapper map = PropertyMapper.get();
+		builder.withWebClientConfigurer((webClient) -> {
+			ExchangeStrategies exchangeStrategies = ExchangeStrategies.builder()
+					.codecs((configurer) -> map.from(properties.getMaxInMemorySize()).whenNonNull()
+							.asInt(DataSize::toBytes)
+							.to((maxInMemorySize) -> configurer.defaultCodecs().maxInMemorySize(maxInMemorySize)))
+					.build();
+			return webClient.mutate().exchangeStrategies(exchangeStrategies).build();
+		});
+	}
+
 	@Bean
 	@ConditionalOnMissingBean
 	public ReactiveElasticsearchClient reactiveElasticsearchClient(ClientConfiguration clientConfiguration) {

spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/data/elasticsearch/ReactiveRestClientProperties.java

@@ -22,6 +22,7 @@
 import java.util.List;
 
 import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.util.unit.DataSize;
 
 /**
  * Configuration properties for Elasticsearch Reactive REST clients.
@@ -62,6 +63,12 @@ public class ReactiveRestClientProperties {
 	 */
 	private Duration socketTimeout;
 
+	/**
+	 * Limit on the number of bytes that can be buffered whenever the input stream needs
+	 * to be aggregated.
+	 */
+	private DataSize maxInMemorySize;
+
 	public List<String> getEndpoints() {
 		return this.endpoints;
 	}
@@ -110,4 +117,12 @@ public void setSocketTimeout(Duration socketTimeout) {
 		this.socketTimeout = socketTimeout;
 	}
 
+	public DataSize getMaxInMemorySize() {
+		return this.maxInMemorySize;
+	}
+
+	public void setMaxInMemorySize(DataSize maxInMemorySize) {
+		this.maxInMemorySize = maxInMemorySize;
+	}
+
 }

spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/data/elasticsearch/ReactiveRestClientAutoConfigurationTests.java

@@ -77,6 +77,7 @@ void restClientCanQueryElasticsearchNode() {
 		this.contextRunner.withPropertyValues(
 				"spring.data.elasticsearch.client.reactive.endpoints=" + elasticsearch.getContainerIpAddress() + ":"
 						+ elasticsearch.getFirstMappedPort(),
+				"spring.data.elasticsearch.client.reactive.max-in-memory-size=-1",
 				"spring.data.elasticsearch.client.reactive.connection-timeout=120s",
 				"spring.data.elasticsearch.client.reactive.socket-timeout=120s").run((context) -> {
 					ReactiveElasticsearchClient client = context.getBean(ReactiveElasticsearchClient.class);