Search with MoreLikeThisQuery should use Pageable.

Original Pull Request #1789
Closes #1787

(cherry picked from commit a2ca312fb2812bd34781206e47be31e9e43dac00)
(cherry picked from commit 85af54635d77b37bfa25178b571404d335a87ead)
(cherry picked from commit 2cb9e30b61e9dd1c0afcccd4499edbab93ab9ebe)

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

@@ -279,7 +279,7 @@ public abstract class AbstractElasticsearchTemplate implements ElasticsearchOper
 		Assert.notNull(query.getId(), "No document id defined for MoreLikeThisQuery");
 
 		MoreLikeThisQueryBuilder moreLikeThisQueryBuilder = requestFactory.moreLikeThisQueryBuilder(query, index);
-		return search(new NativeSearchQueryBuilder().withQuery(moreLikeThisQueryBuilder).build(), clazz, index);
+		return search(new NativeSearchQueryBuilder().withQuery(moreLikeThisQueryBuilder).withPageable(query.getPageable()).build(), clazz, index);
 	}
 
 	@Override

src/main/java/org/springframework/data/elasticsearch/core/query/MoreLikeThisQuery.java

@@ -23,6 +23,7 @@ import java.util.List;
 
 import org.springframework.data.domain.Pageable;
 import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
 
 /**
  * MoreLikeThisQuery
@@ -176,6 +177,9 @@ public class MoreLikeThisQuery {
 	}
 
 	public void setPageable(Pageable pageable) {
+
+		Assert.notNull(pageable, "pageable must not be null");
+
 		this.pageable = pageable;
 	}
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2020 the original author or authors.
+ * Copyright 2014-2021 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -21,6 +21,7 @@ import static org.assertj.core.api.Assertions.*;
 import static org.elasticsearch.index.query.QueryBuilders.*;
 import static org.springframework.data.elasticsearch.annotations.FieldType.*;
 import static org.springframework.data.elasticsearch.core.document.Document.*;
+import static org.springframework.data.elasticsearch.utils.IdGenerator.*;
 import static org.springframework.data.elasticsearch.utils.IndexBuilder.*;
 
 import lombok.AllArgsConstructor;
@@ -41,6 +42,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.UUID;
 import java.util.stream.Collectors;
+import java.util.stream.IntStream;
 
 import org.assertj.core.util.Lists;
 import org.elasticsearch.action.search.SearchRequest;
@@ -1076,6 +1078,48 @@ public abstract class ElasticsearchTemplateTests {
 		assertThat(content).contains(sampleEntity);
 	}
 
+	@Test // #1787
+	void shouldUsePageableOnMoreLikeThisQueries() {
+
+		String sampleMessage = "So we build a web site or an application and want to add search to it, "
+				+ "and then it hits us: getting search working is hard. We want our search solution to be fast,"
+				+ " we want a painless setup and a completely free search schema, we want to be able to index data simply using JSON over HTTP, "
+				+ "we want our search server to be always available, we want to be able to start with one machine and scale to hundreds, "
+				+ "we want real-time search, we want simple multi-tenancy, and we want a solution that is built for the cloud.";
+		String referenceId = nextIdAsString();
+		Collection<String> ids = IntStream.rangeClosed(1, 10).mapToObj(i -> nextIdAsString()).collect(Collectors.toList());
+		ids.add(referenceId);
+		ids.stream()
+				.map(id -> getIndexQuery(SampleEntity.builder().id(id).message(sampleMessage).version(System.currentTimeMillis()).build()))
+				.forEach(indexQuery -> operations.index(indexQuery, index));
+		indexOperations.refresh();
+
+		MoreLikeThisQuery moreLikeThisQuery = new MoreLikeThisQuery();
+		moreLikeThisQuery.setId(referenceId);
+		moreLikeThisQuery.addFields("message");
+		moreLikeThisQuery.setMinDocFreq(1);
+		moreLikeThisQuery.setPageable(PageRequest.of(0, 5));
+
+		SearchHits<SampleEntity> searchHits = operations.search(moreLikeThisQuery, SampleEntity.class, index);
+
+		assertThat(searchHits.getTotalHits()).isEqualTo(10);
+		assertThat(searchHits.getSearchHits()).hasSize(5);
+
+		Collection<String> returnedIds = searchHits.getSearchHits().stream().map(SearchHit::getId).collect(Collectors.toList());
+
+		moreLikeThisQuery.setPageable(PageRequest.of(1, 5));
+
+		searchHits = operations.search(moreLikeThisQuery, SampleEntity.class, index);
+
+		assertThat(searchHits.getTotalHits()).isEqualTo(10);
+		assertThat(searchHits.getSearchHits()).hasSize(5);
+
+		searchHits.getSearchHits().stream().map(SearchHit::getId).forEach(returnedIds::add);
+
+		assertThat(returnedIds).hasSize(10);
+		assertThat(ids).containsAll(returnedIds);
+	}
+
 	@Test // DATAES-167
 	public void shouldReturnResultsWithScanAndScrollForGivenCriteriaQuery() {