Fix reactive native sort.

Original Pull Request #2746
Closes #2745

(cherry picked from commit 73fe086)

Author: sothawo

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

@@ -1341,14 +1341,6 @@ private <T> void prepareSearchRequest(Query query, @Nullable String routing, @Nu
 			builder.minScore((double) query.getMinScore());
 		}
 
-		if (query.getSort() != null) {
-			List<SortOptions> sortOptions = getSortOptions(query.getSort(), persistentEntity);
-
-			if (!sortOptions.isEmpty()) {
-				builder.sort(sortOptions);
-			}
-		}
-
 		addHighlight(query, builder);
 
 		query.getScriptedFields().forEach(scriptedField -> builder.scriptFields(scriptedField.getFieldName(),
@@ -1357,6 +1349,15 @@ private <T> void prepareSearchRequest(Query query, @Nullable String routing, @Nu
 		if (query instanceof NativeQuery) {
 			prepareNativeSearch((NativeQuery) query, builder);
 		}
+		// query.getSort() must be checked after prepareNativeSearch as this already might hav a sort set that must have
+		// higher priority
+		if (query.getSort() != null) {
+			List<SortOptions> sortOptions = getSortOptions(query.getSort(), persistentEntity);
+
+			if (!sortOptions.isEmpty()) {
+				builder.sort(sortOptions);
+			}
+		}
 
 		if (query.getTrackTotalHits() != null) {
 			// logic from the RHLC, choose between -1 and Integer.MAX_VALUE

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

@@ -18,6 +18,7 @@
 import static org.assertj.core.api.Assertions.*;
 import static org.springframework.data.elasticsearch.client.elc.Queries.*;
 
+import co.elastic.clients.elasticsearch._types.SortOrder;
 import co.elastic.clients.elasticsearch._types.aggregations.Aggregate;
 import co.elastic.clients.elasticsearch._types.aggregations.Buckets;
 import co.elastic.clients.elasticsearch._types.aggregations.StringTermsAggregate;
@@ -27,9 +28,12 @@
 import java.util.List;
 import java.util.concurrent.atomic.AtomicInteger;
 
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
+import org.springframework.data.domain.Pageable;
 import org.springframework.data.elasticsearch.client.elc.Aggregation;
 import org.springframework.data.elasticsearch.client.elc.ElasticsearchAggregation;
 import org.springframework.data.elasticsearch.client.elc.NativeQuery;
@@ -48,6 +52,51 @@
 @ContextConfiguration(classes = ReactiveElasticsearchELCIntegrationTests.Config.class)
 public class ReactiveElasticsearchELCIntegrationTests extends ReactiveElasticsearchIntegrationTests {
 
+	@Test // #2745
+	@DisplayName("should use sort defined in native unbounded query")
+	void shouldUseSortDefinedInNativeUnboundedQuery() {
+		var entity1 = randomEntity(null);
+		entity1.setRate(7);
+		var entity2 = randomEntity(null);
+		entity2.setRate(5);
+		var entity3 = randomEntity(null);
+		entity3.setRate(11);
+
+		operations.saveAll(List.of(entity1, entity2, entity3), SampleEntity.class).blockLast();
+
+		var query = NativeQuery.builder()
+				.withQuery(qb -> qb
+						.matchAll(m -> m))
+				.withSort(sob -> sob
+						.field(f -> f
+								.field("rate")
+								.order(SortOrder.Asc)))
+				.withPageable(Pageable.unpaged())
+				.build();
+
+		var rates = operations.search(query, SampleEntity.class)
+				.map(SearchHit::getContent)
+				.map(SampleEntity::getRate)
+				.collectList().block();
+		assertThat(rates).containsExactly(5, 7, 11);
+
+		query = NativeQuery.builder()
+				.withQuery(qb -> qb
+						.matchAll(m -> m))
+				.withSort(sob -> sob
+						.field(f -> f
+								.field("rate")
+								.order(SortOrder.Desc)))
+				.withPageable(Pageable.unpaged())
+				.build();
+
+		rates = operations.search(query, SampleEntity.class)
+				.map(SearchHit::getContent)
+				.map(SampleEntity::getRate)
+				.collectList().block();
+		assertThat(rates).containsExactly(11, 7, 5);
+	}
+
 	@Configuration
 	@Import({ ReactiveElasticsearchTemplateConfiguration.class })
 	static class Config {

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

@@ -99,7 +99,7 @@
 @SpringIntegrationTest
 public abstract class ReactiveElasticsearchIntegrationTests {
 
-	@Autowired private ReactiveElasticsearchOperations operations;
+	@Autowired protected ReactiveElasticsearchOperations operations;
 	@Autowired private IndexNameProvider indexNameProvider;
 
 	// region Setup
@@ -1205,7 +1205,7 @@ void shouldSaveDataFromFluxAndReturnSavedDataInAFlux() {
 	// endregion
 
 	// region Helper functions
-	private SampleEntity randomEntity(String message) {
+	protected SampleEntity randomEntity(@Nullable String message) {
 
 		SampleEntity entity = new SampleEntity();
 		entity.setId(UUID.randomUUID().toString());