DATAES-772 - Add after-convert entity callbacks support.

Original PR: #422

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

@@ -40,6 +40,7 @@ import org.springframework.data.elasticsearch.core.convert.ElasticsearchConverte
 import org.springframework.data.elasticsearch.core.convert.MappingElasticsearchConverter;
 import org.springframework.data.elasticsearch.core.document.Document;
 import org.springframework.data.elasticsearch.core.document.SearchDocumentResponse;
+import org.springframework.data.elasticsearch.core.event.AfterConvertCallback;
 import org.springframework.data.elasticsearch.core.event.AfterSaveCallback;
 import org.springframework.data.elasticsearch.core.event.BeforeConvertCallback;
 import org.springframework.data.elasticsearch.core.mapping.ElasticsearchPersistentEntity;
@@ -236,8 +237,6 @@ public abstract class AbstractElasticsearchTemplate implements ElasticsearchOper
 
 	@Override
 	public <T> CloseableIterator<T> stream(Query query, Class<T> clazz, IndexCoordinates index) {
-
-		long scrollTimeInMillis = TimeValue.timeValueMinutes(1).millis();
 		return (CloseableIterator<T>) SearchHitSupport.unwrapSearchHits(searchForStream(query, clazz, index));
 	}
 
@@ -520,6 +519,15 @@ public abstract class AbstractElasticsearchTemplate implements ElasticsearchOper
 		queries.forEach(this::maybeCallbackAfterSaveWithQuery);
 	}
 
+	protected <T> T maybeCallbackAfterConvert(T entity, Document document, IndexCoordinates index) {
+
+		if (entityCallbacks != null) {
+			return entityCallbacks.callback(AfterConvertCallback.class, entity, document, index);
+		}
+
+		return entity;
+	}
+
 	// endregion
 
 	// region Document callbacks
@@ -528,7 +536,7 @@ public abstract class AbstractElasticsearchTemplate implements ElasticsearchOper
 		T doWith(@Nullable Document document);
 	}
 
-	protected static class ReadDocumentCallback<T> implements DocumentCallback<T> {
+	protected class ReadDocumentCallback<T> implements DocumentCallback<T> {
 		private final EntityReader<? super T, Document> reader;
 		private final Class<T> type;
 		private final IndexCoordinates index;
@@ -550,7 +558,8 @@ public abstract class AbstractElasticsearchTemplate implements ElasticsearchOper
 				return null;
 			}
 
-			return reader.read(type, document);
+			T entity = reader.read(type, document);
+			return maybeCallbackAfterConvert(entity, document, index);
 		}
 	}
 

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

@@ -172,7 +172,7 @@ public class ElasticsearchTemplate extends AbstractElasticsearchTemplate {
 	public <T> List<T> multiGet(Query query, Class<T> clazz, IndexCoordinates index) {
 
 		Assert.notNull(index, "index must not be null");
-		Assert.notEmpty(query.getIds(), "No Id define for Query");
+		Assert.notEmpty(query.getIds(), "No Ids defined for Query");
 
 		MultiGetRequestBuilder builder = requestFactory.multiGetRequestBuilder(client, query, index);
 

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

@@ -71,6 +71,7 @@ import org.springframework.data.elasticsearch.core.convert.MappingElasticsearchC
 import org.springframework.data.elasticsearch.core.document.Document;
 import org.springframework.data.elasticsearch.core.document.DocumentAdapters;
 import org.springframework.data.elasticsearch.core.document.SearchDocument;
+import org.springframework.data.elasticsearch.core.event.ReactiveAfterConvertCallback;
 import org.springframework.data.elasticsearch.core.event.ReactiveAfterSaveCallback;
 import org.springframework.data.elasticsearch.core.event.ReactiveBeforeConvertCallback;
 import org.springframework.data.elasticsearch.core.mapping.ElasticsearchPersistentEntity;
@@ -892,6 +893,16 @@ public class ReactiveElasticsearchTemplate implements ReactiveElasticsearchOpera
 
 		return Mono.just(entity);
 	}
+
+	protected <T> Mono<T> maybeCallAfterConvert(T entity, Document document, IndexCoordinates index) {
+
+		if (null != entityCallbacks) {
+			return entityCallbacks.callback(ReactiveAfterConvertCallback.class, entity, document, index);
+		}
+
+		return Mono.just(entity);
+	}
+
 	// endregion
 
 	protected interface DocumentCallback<T> {
@@ -900,7 +911,7 @@ public class ReactiveElasticsearchTemplate implements ReactiveElasticsearchOpera
 		Mono<T> doWith(@Nullable Document document);
 	}
 
-	protected static class ReadDocumentCallback<T> implements DocumentCallback<T> {
+	protected class ReadDocumentCallback<T> implements DocumentCallback<T> {
 		private final EntityReader<? super T, Document> reader;
 		private final Class<T> type;
 		private final IndexCoordinates index;
@@ -921,7 +932,7 @@ public class ReactiveElasticsearchTemplate implements ReactiveElasticsearchOpera
 			}
 
 			T entity = reader.read(type, document);
-			return Mono.just(entity);
+			return maybeCallAfterConvert(entity, document, index);
 		}
 	}
 

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

@@ -26,6 +26,7 @@ import org.springframework.data.domain.Pageable;
 import org.springframework.data.elasticsearch.core.aggregation.AggregatedPage;
 import org.springframework.data.elasticsearch.core.aggregation.impl.AggregatedPageImpl;
 import org.springframework.data.repository.util.ReactiveWrappers;
+import org.springframework.data.util.CloseableIterator;
 import org.springframework.lang.Nullable;
 
 /**
@@ -33,6 +34,7 @@ import org.springframework.lang.Nullable;
  * 
  * @author Peter-Josef Meisch
  * @author Sascha Woo
+ * @author Roman Puchkovskiy
  * @since 4.0
  */
 public final class SearchHitSupport {
@@ -79,6 +81,10 @@ public final class SearchHitSupport {
 			return unwrapSearchHits(searchHits.getSearchHits());
 		}
 
+		if (result instanceof SearchHitsIterator<?>) {
+			return unwrapSearchHitsIterator((SearchHitsIterator<?>) result);
+		}
+
 		if (ReactiveWrappers.isAvailable(ReactiveWrappers.ReactiveLibrary.PROJECT_REACTOR)) {
 
 			if (result instanceof Flux) {
@@ -90,6 +96,26 @@ public final class SearchHitSupport {
 		return result;
 	}
 
+	private static CloseableIterator<?> unwrapSearchHitsIterator(SearchHitsIterator<?> iterator) {
+		
+		return new CloseableIterator<Object>() {
+			@Override
+			public boolean hasNext() {
+				return iterator.hasNext();
+			}
+
+			@Override
+			public Object next() {
+				return unwrapSearchHits(iterator.next());
+			}
+
+			@Override
+			public void close() {
+				iterator.close();
+			}
+		};
+	}
+
 	/**
 	 * Builds an {@link AggregatedPage} with the {@link SearchHit} objects from a {@link SearchHits} object.
 	 * 

src/main/java/org/springframework/data/elasticsearch/core/event/AfterConvertCallback.java

@@ -0,0 +1,42 @@
+/*
+ * Copyright 2020 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.elasticsearch.core.event;
+
+import org.springframework.data.elasticsearch.core.document.Document;
+import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates;
+import org.springframework.data.mapping.callback.EntityCallback;
+
+/**
+ * Callback being invoked after a domain object is materialized from a {@link Document} when reading results.
+ *
+ * @author Roman Puchkovskiy
+ * @since 4.0
+ * @see org.springframework.data.mapping.callback.EntityCallbacks
+ */
+@FunctionalInterface
+public interface AfterConvertCallback<T> extends EntityCallback<T> {
+
+	/**
+	 * Entity callback method invoked after a domain object is materialized from a {@link Document}. Can return either
+	 * the same or a modified instance of the domain object.
+	 *
+	 * @param entity the domain object (the result of the conversion).
+	 * @param document must not be {@literal null}.
+	 * @param indexCoordinates must not be {@literal null}.
+	 * @return the domain object that is the result of reading it from the {@link Document}.
+	 */
+	T onAfterConvert(T entity, Document document, IndexCoordinates indexCoordinates);
+}

src/main/java/org/springframework/data/elasticsearch/core/event/ReactiveAfterConvertCallback.java

@@ -0,0 +1,43 @@
+/*
+ * Copyright 2020 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.elasticsearch.core.event;
+
+import org.reactivestreams.Publisher;
+import org.springframework.data.elasticsearch.core.document.Document;
+import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates;
+import org.springframework.data.mapping.callback.EntityCallback;
+
+/**
+ * Callback being invoked after a domain object is materialized from a {@link Document} when reading results.
+ *
+ * @author Roman Puchkovskiy
+ * @since 4.0
+ * @see org.springframework.data.mapping.callback.ReactiveEntityCallbacks
+ */
+@FunctionalInterface
+public interface ReactiveAfterConvertCallback<T> extends EntityCallback<T> {
+
+	/**
+	 * Entity callback method invoked after a domain object is materialized from a {@link Document}. Can return either
+	 * the same or a modified instance of the domain object.
+	 *
+	 * @param entity the domain object (the result of the conversion).
+	 * @param document must not be {@literal null}.
+	 * @param indexCoordinates must not be {@literal null}.
+	 * @return a {@link Publisher} emitting the domain object that is the result of reading it from the {@link Document}.
+	 */
+	Publisher<T> onAfterConvert(T entity, Document document, IndexCoordinates indexCoordinates);
+}

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

@@ -0,0 +1,570 @@
+/*
+ * Copyright 2020 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.elasticsearch.core;
+
+import static java.util.Collections.*;
+import static org.assertj.core.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Iterator;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import org.apache.lucene.search.TotalHits;
+import org.elasticsearch.action.search.SearchResponse;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mock;
+import org.mockito.Spy;
+import org.springframework.data.annotation.Id;
+import org.springframework.data.domain.Page;
+import org.springframework.data.elasticsearch.core.aggregation.AggregatedPage;
+import org.springframework.data.elasticsearch.core.document.Document;
+import org.springframework.data.elasticsearch.core.event.AfterConvertCallback;
+import org.springframework.data.elasticsearch.core.event.AfterSaveCallback;
+import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates;
+import org.springframework.data.elasticsearch.core.query.BulkOptions;
+import org.springframework.data.elasticsearch.core.query.GetQuery;
+import org.springframework.data.elasticsearch.core.query.IndexQuery;
+import org.springframework.data.elasticsearch.core.query.MoreLikeThisQuery;
+import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
+import org.springframework.data.elasticsearch.core.query.Query;
+import org.springframework.data.mapping.callback.EntityCallbacks;
+import org.springframework.data.util.CloseableIterator;
+import org.springframework.lang.Nullable;
+import org.springframework.util.CollectionUtils;
+
+/**
+ * @author Roman Puchkovskiy
+ */
+abstract class AbstractElasticsearchTemplateCallbackTests {
+
+	protected AbstractElasticsearchTemplate template;
+
+	@Mock protected SearchResponse searchResponse;
+	@Mock protected org.elasticsearch.search.SearchHit searchHit;
+
+	private final IndexCoordinates index = IndexCoordinates.of("index");
+
+	@Spy private ValueCapturingAfterSaveCallback afterSaveCallback = new ValueCapturingAfterSaveCallback();
+	@Spy private ValueCapturingAfterConvertCallback afterConvertCallback = new ValueCapturingAfterConvertCallback();
+
+	protected final void initTemplate(AbstractElasticsearchTemplate template) {
+		this.template = template;
+		
+		this.template.setEntityCallbacks(EntityCallbacks.create(afterSaveCallback, afterConvertCallback));
+	}
+
+	protected final org.elasticsearch.search.SearchHits nSearchHits(int count) {
+		org.elasticsearch.search.SearchHit[] hits = new org.elasticsearch.search.SearchHit[count];
+		Arrays.fill(hits, searchHit);
+		return new org.elasticsearch.search.SearchHits(hits, new TotalHits(count, TotalHits.Relation.EQUAL_TO), 1.0f);
+	}
+
+	@Test // DATAES-771
+	void saveOneShouldInvokeAfterSaveCallbacks() {
+
+		Person entity = new Person("init", "luke");
+
+		Person saved = template.save(entity);
+
+		verify(afterSaveCallback).onAfterSave(eq(entity));
+		assertThat(saved.id).isEqualTo("after-save");
+	}
+
+	@Test // DATAES-771
+	void saveWithIndexCoordinatesShouldInvokeAfterSaveCallbacks() {
+
+		Person entity = new Person("init", "luke");
+
+		Person saved = template.save(entity, index);
+
+		verify(afterSaveCallback).onAfterSave(eq(entity));
+		assertThat(saved.id).isEqualTo("after-save");
+	}
+
+	@Test // DATAES-771
+	void saveArrayShouldInvokeAfterSaveCallbacks() {
+
+		Person entity1 = new Person("init1", "luke1");
+		Person entity2 = new Person("init2", "luke2");
+
+		Iterable<Person> saved = template.save(entity1, entity2);
+
+		verify(afterSaveCallback, times(2)).onAfterSave(any());
+		Iterator<Person> savedIterator = saved.iterator();
+		assertThat(savedIterator.next().getId()).isEqualTo("after-save");
+		assertThat(savedIterator.next().getId()).isEqualTo("after-save");
+	}
+
+	@Test // DATAES-771
+	void saveIterableShouldInvokeAfterSaveCallbacks() {
+
+		Person entity1 = new Person("init1", "luke1");
+		Person entity2 = new Person("init2", "luke2");
+
+		Iterable<Person> saved = template.save(Arrays.asList(entity1, entity2));
+
+		verify(afterSaveCallback, times(2)).onAfterSave(any());
+		Iterator<Person> savedIterator = saved.iterator();
+		assertThat(savedIterator.next().getId()).isEqualTo("after-save");
+		assertThat(savedIterator.next().getId()).isEqualTo("after-save");
+	}
+
+	@Test // DATAES-771
+	void saveIterableWithIndexCoordinatesShouldInvokeAfterSaveCallbacks() {
+
+		Person entity1 = new Person("init1", "luke1");
+		Person entity2 = new Person("init2", "luke2");
+
+		Iterable<Person> saved = template.save(Arrays.asList(entity1, entity2), index);
+
+		verify(afterSaveCallback, times(2)).onAfterSave(any());
+		Iterator<Person> savedIterator = saved.iterator();
+		assertThat(savedIterator.next().getId()).isEqualTo("after-save");
+		assertThat(savedIterator.next().getId()).isEqualTo("after-save");
+	}
+
+	@Test // DATAES-771
+	void indexShouldInvokeAfterSaveCallbacks() {
+
+		Person entity = new Person("init", "luke");
+
+		IndexQuery indexQuery = indexQueryForEntity(entity);
+		template.index(indexQuery, index);
+
+		verify(afterSaveCallback).onAfterSave(eq(entity));
+		Person savedPerson = (Person) indexQuery.getObject();
+		assertThat(savedPerson.id).isEqualTo("after-save");
+	}
+
+	private IndexQuery indexQueryForEntity(Person entity) {
+		IndexQuery indexQuery = new IndexQuery();
+		indexQuery.setObject(entity);
+		return indexQuery;
+	}
+
+	@Test // DATAES-771
+	void bulkIndexShouldInvokeAfterSaveCallbacks() {
+
+		Person entity1 = new Person("init1", "luke1");
+		Person entity2 = new Person("init2", "luke2");
+
+		IndexQuery query1 = indexQueryForEntity(entity1);
+		IndexQuery query2 = indexQueryForEntity(entity2);
+		template.bulkIndex(Arrays.asList(query1, query2), index);
+
+		verify(afterSaveCallback, times(2)).onAfterSave(any());
+		Person savedPerson1 = (Person) query1.getObject();
+		Person savedPerson2 = (Person) query2.getObject();
+		assertThat(savedPerson1.getId()).isEqualTo("after-save");
+		assertThat(savedPerson2.getId()).isEqualTo("after-save");
+	}
+
+	@Test // DATAES-771
+	void bulkIndexWithOptionsShouldInvokeAfterSaveCallbacks() {
+
+		Person entity1 = new Person("init1", "luke1");
+		Person entity2 = new Person("init2", "luke2");
+
+		IndexQuery query1 = indexQueryForEntity(entity1);
+		IndexQuery query2 = indexQueryForEntity(entity2);
+		template.bulkIndex(Arrays.asList(query1, query2), BulkOptions.defaultOptions(), index);
+
+		verify(afterSaveCallback, times(2)).onAfterSave(any());
+		Person savedPerson1 = (Person) query1.getObject();
+		Person savedPerson2 = (Person) query2.getObject();
+		assertThat(savedPerson1.getId()).isEqualTo("after-save");
+		assertThat(savedPerson2.getId()).isEqualTo("after-save");
+	}
+
+	@Test // DATAES-772
+	void getShouldInvokeAfterConvertCallback() {
+
+		Person result = template.get("init", Person.class);
+
+		verify(afterConvertCallback).onAfterConvert(eq(new Person("init", "luke")), eq(lukeDocument()), any());
+		assertThat(result.id).isEqualTo("after-convert");
+	}
+
+	private Document lukeDocument() {
+		return Document.create()
+				.append("id", "init")
+				.append("firstname", "luke");
+	}
+
+	@Test // DATAES-772
+	void getWithCoordinatesShouldInvokeAfterConvertCallback() {
+
+		Person result = template.get("init", Person.class, index);
+
+		verify(afterConvertCallback).onAfterConvert(eq(new Person("init", "luke")), eq(lukeDocument()), eq(index));
+		assertThat(result.id).isEqualTo("after-convert");
+	}
+
+	@Test // DATAES-772
+	void getViaQueryShouldInvokeAfterConvertCallback() {
+
+		@SuppressWarnings("deprecation") // we know what we test
+		Person result = template.get(new GetQuery("init"), Person.class, index);
+
+		verify(afterConvertCallback).onAfterConvert(eq(new Person("init", "luke")), eq(lukeDocument()), eq(index));
+		assertThat(result.id).isEqualTo("after-convert");
+	}
+
+	@Test // DATAES-772
+	void multiGetShouldInvokeAfterConvertCallback() {
+
+		List<Person> results = template.multiGet(queryForTwo(), Person.class, index);
+
+		verify(afterConvertCallback, times(2))
+				.onAfterConvert(eq(new Person("init", "luke")), eq(lukeDocument()), eq(index));
+		assertThat(results.get(0).id).isEqualTo("after-convert");
+		assertThat(results.get(1).id).isEqualTo("after-convert");
+	}
+
+	private Query queryForTwo() {
+		return new NativeSearchQueryBuilder().withIds(Arrays.asList("init1", "init2")).build();
+	}
+
+	@Test // DATAES-772
+	void queryForObjectShouldInvokeAfterConvertCallback() {
+
+		doReturn(nSearchHits(1)).when(searchResponse).getHits();
+
+		@SuppressWarnings("deprecation") // we know what we test
+		Person result = template.queryForObject(queryForOne(), Person.class, index);
+
+		verify(afterConvertCallback).onAfterConvert(eq(new Person("init", "luke")), eq(lukeDocument()), eq(index));
+		assertThat(result.id).isEqualTo("after-convert");
+	}
+
+	private Query queryForOne() {
+		return new NativeSearchQueryBuilder().withIds(singletonList("init")).build();
+	}
+
+	@Test // DATAES-772
+	void queryForPageShouldInvokeAfterConvertCallback() {
+
+		@SuppressWarnings("deprecation") // we know what we test
+		AggregatedPage<Person> results = template.queryForPage(queryForTwo(), Person.class, index);
+
+		verify(afterConvertCallback, times(2))
+				.onAfterConvert(eq(new Person("init", "luke")), eq(lukeDocument()), eq(index));
+		assertThat(results.getContent().get(0).id).isEqualTo("after-convert");
+		assertThat(results.getContent().get(1).id).isEqualTo("after-convert");
+	}
+
+	@Test // DATAES-772
+	void queryForPageWithMultipleQueriesAndSameEntityClassShouldInvokeAfterConvertCallback() {
+
+		@SuppressWarnings("deprecation") // we know what we test
+		List<Page<Person>> results = template.queryForPage(singletonList(queryForTwo()), Person.class, index);
+
+		verify(afterConvertCallback, times(2))
+				.onAfterConvert(eq(new Person("init", "luke")), eq(lukeDocument()), eq(index));
+		List<Person> persons = results.get(0).getContent();
+		assertThat(persons.get(0).id).isEqualTo("after-convert");
+		assertThat(persons.get(1).id).isEqualTo("after-convert");
+	}
+
+	@Test // DATAES-772
+	void queryForPageWithMultipleQueriesAndEntityClassesShouldInvokeAfterConvertCallback() {
+
+		@SuppressWarnings("deprecation") // we know what we test
+		List<AggregatedPage<?>> results = template.queryForPage(singletonList(queryForTwo()),
+				singletonList(Person.class), index);
+
+		verify(afterConvertCallback, times(2))
+				.onAfterConvert(eq(new Person("init", "luke")), eq(lukeDocument()), eq(index));
+		List<Person> persons = results.get(0).getContent().stream()
+				.map(Person.class::cast)
+				.collect(Collectors.toList());
+		assertThat(persons.get(0).id).isEqualTo("after-convert");
+		assertThat(persons.get(1).id).isEqualTo("after-convert");
+	}
+
+	@Test // DATAES-772
+	void streamShouldInvokeAfterConvertCallback() {
+
+		CloseableIterator<Person> results = template.stream(queryForTwo(), Person.class, index);
+
+		verify(afterConvertCallback, times(2))
+				.onAfterConvert(eq(new Person("init", "luke")), eq(lukeDocument()), eq(index));
+		assertThat(results.next().id).isEqualTo("after-convert");
+		assertThat(results.next().id).isEqualTo("after-convert");
+	}
+
+	@Test // DATAES-772
+	void searchScrollContinueShouldInvokeAfterConvertCallback() {
+
+		CloseableIterator<Person> results = template.stream(queryForTwo(), Person.class, index);
+
+		skipItemsFromScrollStart(results);
+		assertThat(results.next().id).isEqualTo("after-convert");
+		assertThat(results.next().id).isEqualTo("after-convert");
+
+		verify(afterConvertCallback, times(4))
+				.onAfterConvert(eq(new Person("init", "luke")), eq(lukeDocument()), eq(index));
+	}
+
+	private void skipItemsFromScrollStart(CloseableIterator<Person> results) {
+		results.next();
+		results.next();
+	}
+
+	@Test // DATAES-772
+	void queryForListShouldInvokeAfterConvertCallback() {
+
+		@SuppressWarnings("deprecation") // we know what we test
+		List<Person> results = template.queryForList(queryForTwo(), Person.class, index);
+
+		verify(afterConvertCallback, times(2))
+				.onAfterConvert(eq(new Person("init", "luke")), eq(lukeDocument()), eq(index));
+		assertThat(results.get(0).id).isEqualTo("after-convert");
+		assertThat(results.get(1).id).isEqualTo("after-convert");
+	}
+
+	@Test // DATAES-772
+	void queryForListWithMultipleQueriesAndSameEntityClassShouldInvokeAfterConvertCallback() {
+
+		@SuppressWarnings("deprecation") // we know what we test
+		List<List<Person>> results = template.queryForList(singletonList(queryForTwo()), Person.class, index);
+
+		verify(afterConvertCallback, times(2))
+				.onAfterConvert(eq(new Person("init", "luke")), eq(lukeDocument()), eq(index));
+		List<Person> persons = results.get(0);
+		assertThat(persons.get(0).id).isEqualTo("after-convert");
+		assertThat(persons.get(1).id).isEqualTo("after-convert");
+	}
+
+	@Test // DATAES-772
+	void queryForListWithMultipleQueriesAndEntityClassesShouldInvokeAfterConvertCallback() {
+
+		@SuppressWarnings("deprecation") // we know what we test
+		List<List<?>> results = template.queryForList(singletonList(queryForTwo()), singletonList(Person.class), index);
+
+		verify(afterConvertCallback, times(2))
+				.onAfterConvert(eq(new Person("init", "luke")), eq(lukeDocument()), eq(index));
+		List<Person> persons = results.get(0).stream()
+				.map(Person.class::cast)
+				.collect(Collectors.toList());
+		assertThat(persons.get(0).id).isEqualTo("after-convert");
+		assertThat(persons.get(1).id).isEqualTo("after-convert");
+	}
+
+	@Test // DATAES-772
+	void moreLikeThisShouldInvokeAfterConvertCallback() {
+
+		@SuppressWarnings("deprecation") // we know what we test
+		AggregatedPage<Person> results = template.moreLikeThis(moreLikeThisQuery(), Person.class, index);
+
+		verify(afterConvertCallback, times(2))
+				.onAfterConvert(eq(new Person("init", "luke")), eq(lukeDocument()), eq(index));
+		assertThat(results.getContent().get(0).id).isEqualTo("after-convert");
+		assertThat(results.getContent().get(1).id).isEqualTo("after-convert");
+	}
+
+	private MoreLikeThisQuery moreLikeThisQuery() {
+		MoreLikeThisQuery query = new MoreLikeThisQuery();
+		query.setId("init");
+		query.addFields("id");
+		return query;
+	}
+
+	@Test // DATAES-772
+	void searchOneShouldInvokeAfterConvertCallback() {
+
+		doReturn(nSearchHits(1)).when(searchResponse).getHits();
+
+		SearchHit<Person> result = template.searchOne(queryForOne(), Person.class);
+
+		verify(afterConvertCallback).onAfterConvert(eq(new Person("init", "luke")), eq(lukeDocument()), any());
+		assertThat(result.getContent().id).isEqualTo("after-convert");
+	}
+
+	@Test // DATAES-772
+	void searchOneWithIndexCoordinatesShouldInvokeAfterConvertCallback() {
+
+		doReturn(nSearchHits(1)).when(searchResponse).getHits();
+
+		SearchHit<Person> result = template.searchOne(queryForOne(), Person.class, index);
+
+		verify(afterConvertCallback)
+				.onAfterConvert(eq(new Person("init", "luke")), eq(lukeDocument()), eq(index));
+		assertThat(result.getContent().id).isEqualTo("after-convert");
+	}
+
+	@Test // DATAES-772
+	void multiSearchShouldInvokeAfterConvertCallback() {
+
+		List<SearchHits<Person>> results = template.multiSearch(singletonList(queryForTwo()), Person.class, index);
+
+		verify(afterConvertCallback, times(2))
+				.onAfterConvert(eq(new Person("init", "luke")), eq(lukeDocument()), eq(index));
+		List<SearchHit<Person>> hits = results.get(0).getSearchHits();
+		assertThat(hits.get(0).getContent().id).isEqualTo("after-convert");
+		assertThat(hits.get(1).getContent().id).isEqualTo("after-convert");
+	}
+
+	@Test // DATAES-772
+	void multiSearchWithMultipleEntityClassesShouldInvokeAfterConvertCallback() {
+
+		List<SearchHits<?>> results = template.multiSearch(singletonList(queryForTwo()),
+				singletonList(Person.class), index);
+
+		verify(afterConvertCallback, times(2))
+				.onAfterConvert(eq(new Person("init", "luke")), eq(lukeDocument()), eq(index));
+		List<? extends SearchHit<?>> hits = results.get(0).getSearchHits();
+		assertThat(((Person) hits.get(0).getContent()).id).isEqualTo("after-convert");
+		assertThat(((Person) hits.get(1).getContent()).id).isEqualTo("after-convert");
+	}
+
+	@Test // DATAES-772
+	void searchShouldInvokeAfterConvertCallback() {
+
+		SearchHits<Person> results = template.search(queryForTwo(), Person.class);
+
+		verify(afterConvertCallback, times(2))
+				.onAfterConvert(eq(new Person("init", "luke")), eq(lukeDocument()), any());
+		List<SearchHit<Person>> hits = results.getSearchHits();
+		assertThat(hits.get(0).getContent().id).isEqualTo("after-convert");
+		assertThat(hits.get(1).getContent().id).isEqualTo("after-convert");
+	}
+
+	@Test // DATAES-772
+	void searchWithIndexCoordinatesShouldInvokeAfterConvertCallback() {
+
+		SearchHits<Person> results = template.search(queryForTwo(), Person.class, index);
+
+		verify(afterConvertCallback, times(2))
+				.onAfterConvert(eq(new Person("init", "luke")), eq(lukeDocument()), eq(index));
+		List<SearchHit<Person>> hits = results.getSearchHits();
+		assertThat(hits.get(0).getContent().id).isEqualTo("after-convert");
+		assertThat(hits.get(1).getContent().id).isEqualTo("after-convert");
+	}
+
+	@Test // DATAES-772
+	void searchViaMoreLikeThisShouldInvokeAfterConvertCallback() {
+
+		SearchHits<Person> results = template.search(moreLikeThisQuery(), Person.class);
+
+		verify(afterConvertCallback, times(2))
+				.onAfterConvert(eq(new Person("init", "luke")), eq(lukeDocument()), any());
+		List<SearchHit<Person>> hits = results.getSearchHits();
+		assertThat(hits.get(0).getContent().id).isEqualTo("after-convert");
+		assertThat(hits.get(1).getContent().id).isEqualTo("after-convert");
+	}
+
+	@Test // DATAES-772
+	void searchViaMoreLikeThisWithIndexCoordinatesShouldInvokeAfterConvertCallback() {
+
+		SearchHits<Person> results = template.search(moreLikeThisQuery(), Person.class, index);
+
+		verify(afterConvertCallback, times(2))
+				.onAfterConvert(eq(new Person("init", "luke")), eq(lukeDocument()), eq(index));
+		List<SearchHit<Person>> hits = results.getSearchHits();
+		assertThat(hits.get(0).getContent().id).isEqualTo("after-convert");
+		assertThat(hits.get(1).getContent().id).isEqualTo("after-convert");
+	}
+
+	@Test // DATAES-772
+	void searchForStreamShouldInvokeAfterConvertCallback() {
+
+		SearchHitsIterator<Person> results = template.searchForStream(queryForTwo(), Person.class);
+
+		verify(afterConvertCallback, times(2))
+				.onAfterConvert(eq(new Person("init", "luke")), eq(lukeDocument()), any());
+		assertThat(results.next().getContent().id).isEqualTo("after-convert");
+		assertThat(results.next().getContent().id).isEqualTo("after-convert");
+	}
+
+	@Test // DATAES-772
+	void searchForStreamWithIndexCoordinatesShouldInvokeAfterConvertCallback() {
+
+		SearchHitsIterator<Person> results = template.searchForStream(queryForTwo(), Person.class, index);
+
+		verify(afterConvertCallback, times(2))
+				.onAfterConvert(eq(new Person("init", "luke")), eq(lukeDocument()), eq(index));
+		assertThat(results.next().getContent().id).isEqualTo("after-convert");
+		assertThat(results.next().getContent().id).isEqualTo("after-convert");
+	}
+
+	@Data
+	@AllArgsConstructor
+	@NoArgsConstructor
+	static class Person {
+
+		@Id String id;
+		String firstname;
+	}
+
+	static class ValueCapturingEntityCallback<T> {
+
+		private final List<T> values = new ArrayList<>(1);
+
+		protected void capture(T value) {
+			values.add(value);
+		}
+
+		public List<T> getValues() {
+			return values;
+		}
+
+		@Nullable
+		public T getValue() {
+			return CollectionUtils.lastElement(values);
+		}
+
+	}
+
+	static class ValueCapturingAfterSaveCallback extends ValueCapturingEntityCallback<Person>
+			implements AfterSaveCallback<Person> {
+
+		@Override
+		public Person onAfterSave(Person entity) {
+
+			capture(entity);
+			return new Person() {
+				{
+					id = "after-save";
+					firstname = entity.firstname;
+				}
+			};
+		}
+	}
+
+	static class ValueCapturingAfterConvertCallback extends ValueCapturingEntityCallback<Person>
+			implements AfterConvertCallback<Person> {
+
+		@Override
+		public Person onAfterConvert(Person entity, Document document, IndexCoordinates indexCoordinates) {
+
+			capture(entity);
+			return new Person() {
+				{
+					id = "after-convert";
+					firstname = entity.firstname;
+				}
+			};
+		}
+	}
+}

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

@@ -15,59 +15,55 @@
  */
 package org.springframework.data.elasticsearch.core;
 
-import static org.assertj.core.api.Assertions.*;
 import static org.mockito.Mockito.*;
 
-import lombok.AllArgsConstructor;
+import java.util.HashMap;
-import lombok.Data;
-import lombok.NoArgsConstructor;
-
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Iterator;
-import java.util.List;
 
 import org.elasticsearch.action.bulk.BulkItemResponse;
 import org.elasticsearch.action.bulk.BulkRequest;
 import org.elasticsearch.action.bulk.BulkResponse;
+import org.elasticsearch.action.get.GetRequest;
+import org.elasticsearch.action.get.GetResponse;
+import org.elasticsearch.action.get.MultiGetItemResponse;
+import org.elasticsearch.action.get.MultiGetRequest;
+import org.elasticsearch.action.get.MultiGetResponse;
 import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.action.index.IndexResponse;
+import org.elasticsearch.action.search.MultiSearchRequest;
+import org.elasticsearch.action.search.MultiSearchResponse;
+import org.elasticsearch.action.search.SearchRequest;
+import org.elasticsearch.action.search.SearchScrollRequest;
 import org.elasticsearch.client.RequestOptions;
 import org.elasticsearch.client.RestHighLevelClient;
+import org.elasticsearch.common.bytes.BytesArray;
 import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.mockito.Mock;
 import org.mockito.junit.jupiter.MockitoExtension;
 import org.mockito.junit.jupiter.MockitoSettings;
 import org.mockito.quality.Strictness;
-import org.springframework.data.annotation.Id;
-import org.springframework.data.elasticsearch.core.event.AfterSaveCallback;
-import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates;
-import org.springframework.data.elasticsearch.core.query.BulkOptions;
-import org.springframework.data.elasticsearch.core.query.IndexQuery;
-import org.springframework.data.mapping.callback.EntityCallbacks;
-import org.springframework.lang.Nullable;
-import org.springframework.util.CollectionUtils;
 
 /**
  * @author Roman Puchkovskiy
  */
 @ExtendWith(MockitoExtension.class)
 @MockitoSettings(strictness = Strictness.LENIENT)
-public class ElasticsearchRestTemplateCallbackTests {
+class ElasticsearchRestTemplateCallbackTests extends AbstractElasticsearchTemplateCallbackTests {
-
-	private ElasticsearchRestTemplate template;
 
 	@Mock private RestHighLevelClient client;
 
 	@Mock private IndexResponse indexResponse;
 	@Mock private BulkResponse bulkResponse;
 	@Mock private BulkItemResponse bulkItemResponse;
+	@Mock private GetResponse getResponse;
+	@Mock private MultiGetResponse multiGetResponse;
+	@Mock private MultiGetItemResponse multiGetItemResponse;
+	@Mock private MultiSearchResponse.Item multiSearchResponseItem;
 
+	@SuppressWarnings("deprecation") // we know what we test
 	@BeforeEach
 	public void setUp() throws Exception {
-		template = new ElasticsearchRestTemplate(client);
+		initTemplate(new ElasticsearchRestTemplate(client));
 
 		doReturn(indexResponse).when(client).index(any(IndexRequest.class), any(RequestOptions.class));
 		doReturn("response-id").when(indexResponse).getId();
@@ -75,198 +71,39 @@ public class ElasticsearchRestTemplateCallbackTests {
 		doReturn(bulkResponse).when(client).bulk(any(BulkRequest.class), any(RequestOptions.class));
 		doReturn(new BulkItemResponse[] { bulkItemResponse, bulkItemResponse }).when(bulkResponse).getItems();
 		doReturn("response-id").when(bulkItemResponse).getId();
-	}
 
-	@Test // DATAES-771
+		doReturn(getResponse).when(client).get(any(GetRequest.class), any(RequestOptions.class));
-	void saveOneShouldInvokeAfterSaveCallbacks() {
 
-		ValueCapturingAfterSaveCallback afterSaveCallback = spy(new ValueCapturingAfterSaveCallback());
+		doReturn(true).when(getResponse).isExists();
-
+		doReturn(false).when(getResponse).isSourceEmpty();
-		template.setEntityCallbacks(EntityCallbacks.create(afterSaveCallback));
+		doReturn(new HashMap<String, Object>() {
-
-		Person entity = new Person("init", "luke");
-
-		Person saved = template.save(entity);
-
-		verify(afterSaveCallback).onAfterSave(eq(entity));
-		assertThat(saved.id).isEqualTo("after-save");
-	}
-
-	@Test // DATAES-771
-	void saveWithIndexCoordinatesShouldInvokeAfterSaveCallbacks() {
-
-		ValueCapturingAfterSaveCallback afterSaveCallback = spy(new ValueCapturingAfterSaveCallback());
-
-		template.setEntityCallbacks(EntityCallbacks.create(afterSaveCallback));
-
-		Person entity = new Person("init", "luke");
-
-		Person saved = template.save(entity, IndexCoordinates.of("index"));
-
-		verify(afterSaveCallback).onAfterSave(eq(entity));
-		assertThat(saved.id).isEqualTo("after-save");
-	}
-
-	@Test // DATAES-771
-	void saveArrayShouldInvokeAfterSaveCallbacks() {
-
-		ValueCapturingAfterSaveCallback afterSaveCallback = spy(new ValueCapturingAfterSaveCallback());
-
-		template.setEntityCallbacks(EntityCallbacks.create(afterSaveCallback));
-
-		Person entity1 = new Person("init1", "luke1");
-		Person entity2 = new Person("init2", "luke2");
-
-		Iterable<Person> saved = template.save(entity1, entity2);
-
-		verify(afterSaveCallback, times(2)).onAfterSave(any());
-		Iterator<Person> savedIterator = saved.iterator();
-		assertThat(savedIterator.next().getId()).isEqualTo("after-save");
-		assertThat(savedIterator.next().getId()).isEqualTo("after-save");
-	}
-
-	@Test // DATAES-771
-	void saveIterableShouldInvokeAfterSaveCallbacks() {
-
-		ValueCapturingAfterSaveCallback afterSaveCallback = spy(new ValueCapturingAfterSaveCallback());
-
-		template.setEntityCallbacks(EntityCallbacks.create(afterSaveCallback));
-
-		Person entity1 = new Person("init1", "luke1");
-		Person entity2 = new Person("init2", "luke2");
-
-		Iterable<Person> saved = template.save(Arrays.asList(entity1, entity2));
-
-		verify(afterSaveCallback, times(2)).onAfterSave(any());
-		Iterator<Person> savedIterator = saved.iterator();
-		assertThat(savedIterator.next().getId()).isEqualTo("after-save");
-		assertThat(savedIterator.next().getId()).isEqualTo("after-save");
-	}
-
-	@Test // DATAES-771
-	void saveIterableWithIndexCoordinatesShouldInvokeAfterSaveCallbacks() {
-
-		ValueCapturingAfterSaveCallback afterSaveCallback = spy(new ValueCapturingAfterSaveCallback());
-
-		template.setEntityCallbacks(EntityCallbacks.create(afterSaveCallback));
-
-		Person entity1 = new Person("init1", "luke1");
-		Person entity2 = new Person("init2", "luke2");
-
-		Iterable<Person> saved = template.save(Arrays.asList(entity1, entity2), IndexCoordinates.of("index"));
-
-		verify(afterSaveCallback, times(2)).onAfterSave(any());
-		Iterator<Person> savedIterator = saved.iterator();
-		assertThat(savedIterator.next().getId()).isEqualTo("after-save");
-		assertThat(savedIterator.next().getId()).isEqualTo("after-save");
-	}
-
-	@Test // DATAES-771
-	void indexShouldInvokeAfterSaveCallbacks() {
-
-		ValueCapturingAfterSaveCallback afterSaveCallback = spy(new ValueCapturingAfterSaveCallback());
-
-		template.setEntityCallbacks(EntityCallbacks.create(afterSaveCallback));
-
-		Person entity = new Person("init", "luke");
-
-		IndexQuery indexQuery = indexQueryForEntity(entity);
-		template.index(indexQuery, IndexCoordinates.of("index"));
-
-		verify(afterSaveCallback).onAfterSave(eq(entity));
-		Person newPerson = (Person) indexQuery.getObject();
-		assertThat(newPerson.id).isEqualTo("after-save");
-	}
-
-	private IndexQuery indexQueryForEntity(Person entity) {
-		IndexQuery indexQuery = new IndexQuery();
-		indexQuery.setObject(entity);
-		return indexQuery;
-	}
-
-	@Test // DATAES-771
-	void bulkIndexShouldInvokeAfterSaveCallbacks() {
-
-		ValueCapturingAfterSaveCallback afterSaveCallback = spy(new ValueCapturingAfterSaveCallback());
-
-		template.setEntityCallbacks(EntityCallbacks.create(afterSaveCallback));
-
-		Person entity1 = new Person("init1", "luke1");
-		Person entity2 = new Person("init2", "luke2");
-
-		IndexQuery query1 = indexQueryForEntity(entity1);
-		IndexQuery query2 = indexQueryForEntity(entity2);
-		template.bulkIndex(Arrays.asList(query1, query2), IndexCoordinates.of("index"));
-
-		verify(afterSaveCallback, times(2)).onAfterSave(any());
-		Person savedPerson1 = (Person) query1.getObject();
-		Person savedPerson2 = (Person) query2.getObject();
-		assertThat(savedPerson1.getId()).isEqualTo("after-save");
-		assertThat(savedPerson2.getId()).isEqualTo("after-save");
-	}
-
-	@Test // DATAES-771
-	void bulkIndexWithOptionsShouldInvokeAfterSaveCallbacks() {
-
-		ValueCapturingAfterSaveCallback afterSaveCallback = spy(new ValueCapturingAfterSaveCallback());
-
-		template.setEntityCallbacks(EntityCallbacks.create(afterSaveCallback));
-
-		Person entity1 = new Person("init1", "luke1");
-		Person entity2 = new Person("init2", "luke2");
-
-		IndexQuery query1 = indexQueryForEntity(entity1);
-		IndexQuery query2 = indexQueryForEntity(entity2);
-		template.bulkIndex(Arrays.asList(query1, query2), BulkOptions.defaultOptions(), IndexCoordinates.of("index"));
-
-		verify(afterSaveCallback, times(2)).onAfterSave(any());
-		Person savedPerson1 = (Person) query1.getObject();
-		Person savedPerson2 = (Person) query2.getObject();
-		assertThat(savedPerson1.getId()).isEqualTo("after-save");
-		assertThat(savedPerson2.getId()).isEqualTo("after-save");
-	}
-
-	@Data
-	@AllArgsConstructor
-	@NoArgsConstructor
-	static class Person {
-
-		@Id String id;
-		String firstname;
-	}
-
-	static class ValueCapturingEntityCallback<T> {
-
-		private final List<T> values = new ArrayList<>(1);
-
-		protected void capture(T value) {
-			values.add(value);
-		}
-
-		public List<T> getValues() {
-			return values;
-		}
-
-		@Nullable
-		public T getValue() {
-			return CollectionUtils.lastElement(values);
-		}
-
-	}
-
-	static class ValueCapturingAfterSaveCallback extends ValueCapturingEntityCallback<Person>
-			implements AfterSaveCallback<Person> {
-
-		@Override
-		public Person onAfterSave(Person entity) {
-
-			capture(entity);
-			return new Person() {
 				{
-					id = "after-save";
+					put("id", "init");
-					firstname = entity.firstname;
+					put("firstname", "luke");
 				}
-			};
+		}).when(getResponse).getSourceAsMap();
-		}
+
+		doReturn(multiGetResponse).when(client).mget(any(MultiGetRequest.class), any(RequestOptions.class));
+		doReturn(new MultiGetItemResponse[]{multiGetItemResponse, multiGetItemResponse})
+				.when(multiGetResponse).getResponses();
+		doReturn(getResponse).when(multiGetItemResponse).getResponse();
+
+		doReturn(searchResponse).when(client).search(any(SearchRequest.class), any(RequestOptions.class));
+		doReturn(nSearchHits(2)).when(searchResponse).getHits();
+		doReturn("scroll-id").when(searchResponse).getScrollId();
+		doReturn(new BytesArray(new byte[8])).when(searchHit).getSourceRef();
+		doReturn(new HashMap<String, Object>() {
+				{
+					put("id", "init");
+					put("firstname", "luke");
+				}
+		}).when(searchHit).getSourceAsMap();
+
+		MultiSearchResponse multiSearchResponse = new MultiSearchResponse(
+				new MultiSearchResponse.Item[]{ multiSearchResponseItem }, 1L);
+		doReturn(multiSearchResponse).when(client).multiSearch(any(MultiSearchRequest.class), any());
+		doReturn(searchResponse).when(multiSearchResponseItem).getResponse();
+
+		doReturn(searchResponse).when(client).scroll(any(SearchScrollRequest.class), any(RequestOptions.class));
 	}
 }

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

@@ -2561,7 +2561,6 @@ public abstract class ElasticsearchTemplateTests {
 	public void shouldSortResultsGivenSortCriteriaWithScanAndScroll() {
 
 		// given
-		List<IndexQuery> indexQueries = new ArrayList<>();
 		// first document
 		String documentId = randomNumeric(5);
 		SampleEntity sampleEntity1 = SampleEntity.builder().id(documentId).message("abc").rate(10)
@@ -2577,7 +2576,7 @@ public abstract class ElasticsearchTemplateTests {
 		SampleEntity sampleEntity3 = SampleEntity.builder().id(documentId3).message("xyz").rate(10)
 				.version(System.currentTimeMillis()).build();
 
-		indexQueries = getIndexQueries(Arrays.asList(sampleEntity1, sampleEntity2, sampleEntity3));
+		List<IndexQuery> indexQueries = getIndexQueries(Arrays.asList(sampleEntity1, sampleEntity2, sampleEntity3));
 
 		operations.bulkIndex(indexQueries, index);
 		indexOperations.refresh();
@@ -2609,7 +2608,6 @@ public abstract class ElasticsearchTemplateTests {
 	public void shouldSortResultsGivenSortCriteriaFromPageableWithScanAndScroll() {
 
 		// given
-		List<IndexQuery> indexQueries = new ArrayList<>();
 		// first document
 		String documentId = randomNumeric(5);
 		SampleEntity sampleEntity1 = SampleEntity.builder().id(documentId).message("abc").rate(10)
@@ -2625,7 +2623,7 @@ public abstract class ElasticsearchTemplateTests {
 		SampleEntity sampleEntity3 = SampleEntity.builder().id(documentId3).message("xyz").rate(10)
 				.version(System.currentTimeMillis()).build();
 
-		indexQueries = getIndexQueries(Arrays.asList(sampleEntity1, sampleEntity2, sampleEntity3));
+		List<IndexQuery> indexQueries = getIndexQueries(Arrays.asList(sampleEntity1, sampleEntity2, sampleEntity3));
 
 		operations.bulkIndex(indexQueries, index);
 		indexOperations.refresh();

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

@@ -15,49 +15,43 @@
  */
 package org.springframework.data.elasticsearch.core;
 
-import static org.assertj.core.api.Assertions.*;
 import static org.mockito.Mockito.*;
 
-import lombok.AllArgsConstructor;
+import java.util.HashMap;
-import lombok.Data;
-import lombok.NoArgsConstructor;
-
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Iterator;
-import java.util.List;
 
 import org.elasticsearch.action.ActionFuture;
 import org.elasticsearch.action.bulk.BulkItemResponse;
 import org.elasticsearch.action.bulk.BulkRequestBuilder;
 import org.elasticsearch.action.bulk.BulkResponse;
+import org.elasticsearch.action.get.GetRequestBuilder;
+import org.elasticsearch.action.get.GetResponse;
+import org.elasticsearch.action.get.MultiGetItemResponse;
+import org.elasticsearch.action.get.MultiGetRequestBuilder;
+import org.elasticsearch.action.get.MultiGetResponse;
 import org.elasticsearch.action.index.IndexRequestBuilder;
 import org.elasticsearch.action.index.IndexResponse;
+import org.elasticsearch.action.search.MultiSearchRequest;
+import org.elasticsearch.action.search.MultiSearchResponse;
+import org.elasticsearch.action.search.SearchRequestBuilder;
+import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.action.search.SearchScrollRequestBuilder;
+import org.elasticsearch.action.search.SearchType;
 import org.elasticsearch.client.Client;
+import org.elasticsearch.common.bytes.BytesArray;
+import org.elasticsearch.common.unit.TimeValue;
 import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.mockito.Mock;
 import org.mockito.junit.jupiter.MockitoExtension;
 import org.mockito.junit.jupiter.MockitoSettings;
 import org.mockito.quality.Strictness;
-import org.springframework.data.annotation.Id;
-import org.springframework.data.elasticsearch.core.event.AfterSaveCallback;
-import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates;
-import org.springframework.data.elasticsearch.core.query.BulkOptions;
-import org.springframework.data.elasticsearch.core.query.IndexQuery;
-import org.springframework.data.mapping.callback.EntityCallbacks;
-import org.springframework.lang.Nullable;
-import org.springframework.util.CollectionUtils;
 
 /**
  * @author Roman Puchkovskiy
  */
 @ExtendWith(MockitoExtension.class)
 @MockitoSettings(strictness = Strictness.LENIENT)
-public class ElasticsearchTransportTemplateCallbackTests {
+class ElasticsearchTransportTemplateCallbackTests extends AbstractElasticsearchTemplateCallbackTests {
-
-	private ElasticsearchTemplate template;
 
 	@Mock private Client client;
 
@@ -68,10 +62,23 @@ public class ElasticsearchTransportTemplateCallbackTests {
 	@Mock private ActionFuture<BulkResponse> bulkResponseActionFuture;
 	@Mock private BulkResponse bulkResponse;
 	@Mock private BulkItemResponse bulkItemResponse;
+	@Mock private GetRequestBuilder getRequestBuilder;
+	@Mock private ActionFuture<GetResponse> getResponseActionFuture;
+	@Mock private GetResponse getResponse;
+	@Mock private MultiGetRequestBuilder multiGetRequestBuilder;
+	@Mock private ActionFuture<MultiGetResponse> multiGetResponseActionFuture;
+	@Mock private MultiGetResponse multiGetResponse;
+	@Mock private MultiGetItemResponse multiGetItemResponse;
+	@Mock private SearchRequestBuilder searchRequestBuilder;
+	@Mock private ActionFuture<SearchResponse> searchResponseActionFuture;
+	@Mock private ActionFuture<MultiSearchResponse> multiSearchResponseActionFuture;
+	@Mock private MultiSearchResponse.Item multiSearchResponseItem;
+	@Mock private SearchScrollRequestBuilder searchScrollRequestBuilder;
 
 	@BeforeEach
+	@SuppressWarnings("deprecation") // we know what we are testing
 	public void setUp() {
-		template = new ElasticsearchTemplate(client);
+		initTemplate(new ElasticsearchTemplate(client));
 
 		when(client.prepareIndex(anyString(), anyString(), anyString())).thenReturn(indexRequestBuilder);
 		doReturn(indexResponseActionFuture).when(indexRequestBuilder).execute();
@@ -83,198 +90,53 @@ public class ElasticsearchTransportTemplateCallbackTests {
 		when(bulkResponseActionFuture.actionGet()).thenReturn(bulkResponse);
 		doReturn(new BulkItemResponse[] { bulkItemResponse, bulkItemResponse }).when(bulkResponse).getItems();
 		doReturn("response-id").when(bulkItemResponse).getId();
-	}
 
-	@Test // DATAES-771
+		when(client.prepareGet(anyString(), any(), any())).thenReturn(getRequestBuilder);
-	void saveOneShouldInvokeAfterSaveCallbacks() {
+		doReturn(getResponseActionFuture).when(getRequestBuilder).execute();
+		when(getResponseActionFuture.actionGet()).thenReturn(getResponse);
 
-		ValueCapturingAfterSaveCallback afterSaveCallback = spy(new ValueCapturingAfterSaveCallback());
+		doReturn(true).when(getResponse).isExists();
-
+		doReturn(false).when(getResponse).isSourceEmpty();
-		template.setEntityCallbacks(EntityCallbacks.create(afterSaveCallback));
+		doReturn(new HashMap<String, Object>() {
-
-		Person entity = new Person("init", "luke");
-
-		Person saved = template.save(entity);
-
-		verify(afterSaveCallback).onAfterSave(eq(entity));
-		assertThat(saved.id).isEqualTo("after-save");
-	}
-
-	@Test // DATAES-771
-	void saveWithIndexCoordinatesShouldInvokeAfterSaveCallbacks() {
-
-		ValueCapturingAfterSaveCallback afterSaveCallback = spy(new ValueCapturingAfterSaveCallback());
-
-		template.setEntityCallbacks(EntityCallbacks.create(afterSaveCallback));
-
-		Person entity = new Person("init", "luke");
-
-		Person saved = template.save(entity, IndexCoordinates.of("index"));
-
-		verify(afterSaveCallback).onAfterSave(eq(entity));
-		assertThat(saved.id).isEqualTo("after-save");
-	}
-
-	@Test // DATAES-771
-	void saveArrayShouldInvokeAfterSaveCallbacks() {
-
-		ValueCapturingAfterSaveCallback afterSaveCallback = spy(new ValueCapturingAfterSaveCallback());
-
-		template.setEntityCallbacks(EntityCallbacks.create(afterSaveCallback));
-
-		Person entity1 = new Person("init1", "luke1");
-		Person entity2 = new Person("init2", "luke2");
-
-		Iterable<Person> saved = template.save(entity1, entity2);
-
-		verify(afterSaveCallback, times(2)).onAfterSave(any());
-		Iterator<Person> savedIterator = saved.iterator();
-		assertThat(savedIterator.next().getId()).isEqualTo("after-save");
-		assertThat(savedIterator.next().getId()).isEqualTo("after-save");
-	}
-
-	@Test // DATAES-771
-	void saveIterableShouldInvokeAfterSaveCallbacks() {
-
-		ValueCapturingAfterSaveCallback afterSaveCallback = spy(new ValueCapturingAfterSaveCallback());
-
-		template.setEntityCallbacks(EntityCallbacks.create(afterSaveCallback));
-
-		Person entity1 = new Person("init1", "luke1");
-		Person entity2 = new Person("init2", "luke2");
-
-		Iterable<Person> saved = template.save(Arrays.asList(entity1, entity2));
-
-		verify(afterSaveCallback, times(2)).onAfterSave(any());
-		Iterator<Person> savedIterator = saved.iterator();
-		assertThat(savedIterator.next().getId()).isEqualTo("after-save");
-		assertThat(savedIterator.next().getId()).isEqualTo("after-save");
-	}
-
-	@Test // DATAES-771
-	void saveIterableWithIndexCoordinatesShouldInvokeAfterSaveCallbacks() {
-
-		ValueCapturingAfterSaveCallback afterSaveCallback = spy(new ValueCapturingAfterSaveCallback());
-
-		template.setEntityCallbacks(EntityCallbacks.create(afterSaveCallback));
-
-		Person entity1 = new Person("init1", "luke1");
-		Person entity2 = new Person("init2", "luke2");
-
-		Iterable<Person> saved = template.save(Arrays.asList(entity1, entity2), IndexCoordinates.of("index"));
-
-		verify(afterSaveCallback, times(2)).onAfterSave(any());
-		Iterator<Person> savedIterator = saved.iterator();
-		assertThat(savedIterator.next().getId()).isEqualTo("after-save");
-		assertThat(savedIterator.next().getId()).isEqualTo("after-save");
-	}
-
-	@Test // DATAES-771
-	void indexShouldInvokeAfterSaveCallbacks() {
-
-		ValueCapturingAfterSaveCallback afterSaveCallback = spy(new ValueCapturingAfterSaveCallback());
-
-		template.setEntityCallbacks(EntityCallbacks.create(afterSaveCallback));
-
-		Person entity = new Person("init", "luke");
-
-		IndexQuery indexQuery = indexQueryForEntity(entity);
-		template.index(indexQuery, IndexCoordinates.of("index"));
-
-		verify(afterSaveCallback).onAfterSave(eq(entity));
-		Person savedPerson = (Person) indexQuery.getObject();
-		assertThat(savedPerson.id).isEqualTo("after-save");
-	}
-
-	private IndexQuery indexQueryForEntity(Person entity) {
-		IndexQuery indexQuery = new IndexQuery();
-		indexQuery.setObject(entity);
-		return indexQuery;
-	}
-
-	@Test // DATAES-771
-	void bulkIndexShouldInvokeAfterSaveCallbacks() {
-
-		ValueCapturingAfterSaveCallback afterSaveCallback = spy(new ValueCapturingAfterSaveCallback());
-
-		template.setEntityCallbacks(EntityCallbacks.create(afterSaveCallback));
-
-		Person entity1 = new Person("init1", "luke1");
-		Person entity2 = new Person("init2", "luke2");
-
-		IndexQuery query1 = indexQueryForEntity(entity1);
-		IndexQuery query2 = indexQueryForEntity(entity2);
-		template.bulkIndex(Arrays.asList(query1, query2), IndexCoordinates.of("index"));
-
-		verify(afterSaveCallback, times(2)).onAfterSave(any());
-		Person savedPerson1 = (Person) query1.getObject();
-		Person savedPerson2 = (Person) query2.getObject();
-		assertThat(savedPerson1.getId()).isEqualTo("after-save");
-		assertThat(savedPerson2.getId()).isEqualTo("after-save");
-	}
-
-	@Test // DATAES-771
-	void bulkIndexWithOptionsShouldInvokeAfterSaveCallbacks() {
-
-		ValueCapturingAfterSaveCallback afterSaveCallback = spy(new ValueCapturingAfterSaveCallback());
-
-		template.setEntityCallbacks(EntityCallbacks.create(afterSaveCallback));
-
-		Person entity1 = new Person("init1", "luke1");
-		Person entity2 = new Person("init2", "luke2");
-
-		IndexQuery query1 = indexQueryForEntity(entity1);
-		IndexQuery query2 = indexQueryForEntity(entity2);
-		template.bulkIndex(Arrays.asList(query1, query2), BulkOptions.defaultOptions(), IndexCoordinates.of("index"));
-
-		verify(afterSaveCallback, times(2)).onAfterSave(any());
-		Person savedPerson1 = (Person) query1.getObject();
-		Person savedPerson2 = (Person) query2.getObject();
-		assertThat(savedPerson1.getId()).isEqualTo("after-save");
-		assertThat(savedPerson2.getId()).isEqualTo("after-save");
-	}
-
-	@Data
-	@AllArgsConstructor
-	@NoArgsConstructor
-	static class Person {
-
-		@Id String id;
-		String firstname;
-	}
-
-	static class ValueCapturingEntityCallback<T> {
-
-		private final List<T> values = new ArrayList<>(1);
-
-		protected void capture(T value) {
-			values.add(value);
-		}
-
-		public List<T> getValues() {
-			return values;
-		}
-
-		@Nullable
-		public T getValue() {
-			return CollectionUtils.lastElement(values);
-		}
-
-	}
-
-	static class ValueCapturingAfterSaveCallback extends ValueCapturingEntityCallback<Person>
-			implements AfterSaveCallback<Person> {
-
-		@Override
-		public Person onAfterSave(Person entity) {
-
-			capture(entity);
-			return new Person() {
 				{
-					id = "after-save";
+					put("id", "init");
-					firstname = entity.firstname;
+					put("firstname", "luke");
 				}
-			};
+		}).when(getResponse).getSourceAsMap();
-		}
+
+		when(client.prepareMultiGet()).thenReturn(multiGetRequestBuilder);
+		doReturn(multiGetResponseActionFuture).when(multiGetRequestBuilder).execute();
+		when(multiGetResponseActionFuture.actionGet()).thenReturn(multiGetResponse);
+		doReturn(new MultiGetItemResponse[]{ multiGetItemResponse, multiGetItemResponse })
+				.when(multiGetResponse).getResponses();
+		doReturn(getResponse).when(multiGetItemResponse).getResponse();
+
+		when(client.prepareSearch(anyVararg())).thenReturn(searchRequestBuilder);
+		doReturn(searchRequestBuilder).when(searchRequestBuilder).setSearchType(any(SearchType.class));
+		doReturn(searchRequestBuilder).when(searchRequestBuilder).setVersion(anyBoolean());
+		doReturn(searchRequestBuilder).when(searchRequestBuilder).setTrackScores(anyBoolean());
+		doReturn(searchRequestBuilder).when(searchRequestBuilder).setScroll(any(TimeValue.class));
+		doReturn(searchResponseActionFuture).when(searchRequestBuilder).execute();
+		when(searchResponseActionFuture.actionGet()).thenReturn(searchResponse);
+		when(searchResponseActionFuture.actionGet(anyString())).thenReturn(searchResponse);
+		doReturn(nSearchHits(2)).when(searchResponse).getHits();
+		doReturn("scroll-id").when(searchResponse).getScrollId();
+		doReturn(new BytesArray(new byte[8])).when(searchHit).getSourceRef();
+		doReturn(new HashMap<String, Object>() {
+				{
+					put("id", "init");
+					put("firstname", "luke");
+				}
+		}).when(searchHit).getSourceAsMap();
+
+		when(client.multiSearch(any(MultiSearchRequest.class))).thenReturn(multiSearchResponseActionFuture);
+		MultiSearchResponse multiSearchResponse = new MultiSearchResponse(
+				new MultiSearchResponse.Item[]{ multiSearchResponseItem }, 1L);
+		when(multiSearchResponseActionFuture.actionGet()).thenReturn(multiSearchResponse);
+		doReturn(searchResponse).when(multiSearchResponseItem).getResponse();
+
+		when(client.prepareSearchScroll(anyString())).thenReturn(searchScrollRequestBuilder);
+		doReturn(searchScrollRequestBuilder).when(searchScrollRequestBuilder).setScroll(any(TimeValue.class));
+		doReturn(searchResponseActionFuture).when(searchScrollRequestBuilder).execute();
 	}
 }

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

@@ -21,11 +21,13 @@ import static org.mockito.Mockito.*;
 import lombok.AllArgsConstructor;
 import lombok.Data;
 import lombok.NoArgsConstructor;
+import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.HashMap;
 import java.util.List;
 import java.util.stream.Collectors;
 
@@ -33,8 +35,13 @@ import org.elasticsearch.action.DocWriteResponse;
 import org.elasticsearch.action.bulk.BulkItemResponse;
 import org.elasticsearch.action.bulk.BulkRequest;
 import org.elasticsearch.action.bulk.BulkResponse;
+import org.elasticsearch.action.get.GetRequest;
+import org.elasticsearch.action.get.MultiGetRequest;
 import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.action.index.IndexResponse;
+import org.elasticsearch.action.search.SearchRequest;
+import org.elasticsearch.common.bytes.BytesArray;
+import org.elasticsearch.index.get.GetResult;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
@@ -43,9 +50,14 @@ import org.mockito.junit.jupiter.MockitoExtension;
 import org.mockito.junit.jupiter.MockitoSettings;
 import org.mockito.quality.Strictness;
 import org.springframework.data.annotation.Id;
+import org.springframework.data.domain.PageRequest;
 import org.springframework.data.elasticsearch.client.reactive.ReactiveElasticsearchClient;
+import org.springframework.data.elasticsearch.core.document.Document;
+import org.springframework.data.elasticsearch.core.event.ReactiveAfterConvertCallback;
 import org.springframework.data.elasticsearch.core.event.ReactiveAfterSaveCallback;
 import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates;
+import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
+import org.springframework.data.elasticsearch.core.query.Query;
 import org.springframework.data.mapping.callback.ReactiveEntityCallbacks;
 import org.springframework.lang.Nullable;
 import org.springframework.util.CollectionUtils;
@@ -65,6 +77,10 @@ public class ReactiveElasticsearchTemplateCallbackTests {
 	@Mock private BulkResponse bulkResponse;
 	@Mock private BulkItemResponse bulkItemResponse;
 	@Mock private DocWriteResponse docWriteResponse;
+	@Mock private GetResult getResult;
+	@Mock private org.elasticsearch.search.SearchHit searchHit;
+
+	private final IndexCoordinates index = IndexCoordinates.of("index");
 
 	@BeforeEach
 	public void setUp() {
@@ -77,6 +93,30 @@ public class ReactiveElasticsearchTemplateCallbackTests {
 		doReturn(new BulkItemResponse[] { bulkItemResponse, bulkItemResponse }).when(bulkResponse).getItems();
 		doReturn(docWriteResponse).when(bulkItemResponse).getResponse();
 		doReturn("response-id").when(docWriteResponse).getId();
+
+		when(client.multiGet(any(MultiGetRequest.class))).thenReturn(Flux.just(getResult, getResult));
+
+		doReturn(true).when(getResult).isExists();
+		doReturn(false).when(getResult).isSourceEmpty();
+		doReturn(new HashMap<String, Object>() {
+				{
+					put("id", "init");
+					put("firstname", "luke");
+				}
+		}).when(getResult).getSource();
+
+		doReturn(Mono.just(getResult)).when(client).get(any(GetRequest.class));
+
+		when(client.search(any(SearchRequest.class))).thenReturn(Flux.just(searchHit, searchHit));
+		doReturn(new BytesArray(new byte[8])).when(searchHit).getSourceRef();
+		doReturn(new HashMap<String, Object>() {
+				{
+					put("id", "init");
+					put("firstname", "luke");
+				}
+		}).when(searchHit).getSourceAsMap();
+
+		when(client.scroll(any(SearchRequest.class))).thenReturn(Flux.just(searchHit, searchHit));
 	}
 
 	@Test // DATAES-771
@@ -175,6 +215,254 @@ public class ReactiveElasticsearchTemplateCallbackTests {
 		assertThat(saved.get(1).getId()).isEqualTo("after-save");
 	}
 
+	@Test // DATAES-772
+	void multiGetShouldInvokeAfterConvertCallbacks() {
+
+		ValueCapturingAfterConvertCallback afterConvertCallback = spy(new ValueCapturingAfterConvertCallback());
+
+		template.setEntityCallbacks(ReactiveEntityCallbacks.create(afterConvertCallback));
+
+		List<Person> results = template.multiGet(pagedQueryForTwo(), Person.class, index)
+				.timeout(Duration.ofSeconds(1))
+				.toStream().collect(Collectors.toList());
+
+		verify(afterConvertCallback, times(2))
+				.onAfterConvert(eq(new Person("init", "luke")), eq(lukeDocument()), eq(index));
+		assertThat(results.get(0).id).isEqualTo("after-convert");
+		assertThat(results.get(1).id).isEqualTo("after-convert");
+	}
+
+	@Test // DATAES-772
+	void findByIdShouldInvokeAfterConvertCallbacks() {
+
+		ValueCapturingAfterConvertCallback afterConvertCallback = spy(new ValueCapturingAfterConvertCallback());
+
+		template.setEntityCallbacks(ReactiveEntityCallbacks.create(afterConvertCallback));
+
+		@SuppressWarnings("deprecation") // we know what we test
+		Person result = template.findById("init", Person.class).block(Duration.ofSeconds(1));
+
+		verify(afterConvertCallback).onAfterConvert(eq(new Person("init", "luke")), eq(lukeDocument()), any());
+		assertThat(result.id).isEqualTo("after-convert");
+	}
+
+	@Test // DATAES-772
+	void findByIdWithIndexCoordinatesShouldInvokeAfterConvertCallbacks() {
+
+		ValueCapturingAfterConvertCallback afterConvertCallback = spy(new ValueCapturingAfterConvertCallback());
+
+		template.setEntityCallbacks(ReactiveEntityCallbacks.create(afterConvertCallback));
+
+		@SuppressWarnings("deprecation") // we know what we test
+		Person result = template.findById("init", Person.class, index).block(Duration.ofSeconds(1));
+
+		verify(afterConvertCallback).onAfterConvert(eq(new Person("init", "luke")), eq(lukeDocument()), eq(index));
+		assertThat(result.id).isEqualTo("after-convert");
+	}
+
+	@Test // DATAES-772
+	void getShouldInvokeAfterConvertCallbacks() {
+
+		ValueCapturingAfterConvertCallback afterConvertCallback = spy(new ValueCapturingAfterConvertCallback());
+
+		template.setEntityCallbacks(ReactiveEntityCallbacks.create(afterConvertCallback));
+
+		Person result = template.get("init", Person.class).block(Duration.ofSeconds(1));
+
+		verify(afterConvertCallback).onAfterConvert(eq(new Person("init", "luke")), eq(lukeDocument()), any());
+		assertThat(result.id).isEqualTo("after-convert");
+	}
+
+	@Test // DATAES-772
+	void getWithIndexCoordinatesShouldInvokeAfterConvertCallbacks() {
+
+		ValueCapturingAfterConvertCallback afterConvertCallback = spy(new ValueCapturingAfterConvertCallback());
+
+		template.setEntityCallbacks(ReactiveEntityCallbacks.create(afterConvertCallback));
+
+		Person result = template.get("init", Person.class, index).block(Duration.ofSeconds(1));
+
+		verify(afterConvertCallback).onAfterConvert(eq(new Person("init", "luke")), eq(lukeDocument()), eq(index));
+		assertThat(result.id).isEqualTo("after-convert");
+	}
+
+	@Test // DATAES-772
+	void findUsingPageableShouldInvokeAfterConvertCallbacks() {
+
+		ValueCapturingAfterConvertCallback afterConvertCallback = spy(new ValueCapturingAfterConvertCallback());
+
+		template.setEntityCallbacks(ReactiveEntityCallbacks.create(afterConvertCallback));
+
+		@SuppressWarnings("deprecation") // we know what we test
+		List<Person> results = template.find(pagedQueryForTwo(), Person.class)
+				.timeout(Duration.ofSeconds(1)).toStream()
+				.collect(Collectors.toList());
+
+		verify(afterConvertCallback, times(2))
+				.onAfterConvert(eq(new Person("init", "luke")), eq(lukeDocument()), any());
+		assertThat(results.get(0).id).isEqualTo("after-convert");
+		assertThat(results.get(1).id).isEqualTo("after-convert");
+	}
+	
+	private Query pagedQueryForTwo() {
+		return new NativeSearchQueryBuilder()
+				.withIds(Arrays.asList("init1", "init2"))
+				.withPageable(PageRequest.of(0, 10))
+				.build();
+	}
+
+	private Document lukeDocument() {
+		return Document.create()
+				.append("id", "init")
+				.append("firstname", "luke");
+	}
+
+	@Test // DATAES-772
+	void findUsingScrollShouldInvokeAfterConvertCallbacks() {
+
+		ValueCapturingAfterConvertCallback afterConvertCallback = spy(new ValueCapturingAfterConvertCallback());
+
+		template.setEntityCallbacks(ReactiveEntityCallbacks.create(afterConvertCallback));
+
+		@SuppressWarnings("deprecation") // we know what we test
+		List<Person> results = template.find(scrollingQueryForTwo(), Person.class)
+				.timeout(Duration.ofSeconds(1))
+				.toStream().collect(Collectors.toList());
+
+		verify(afterConvertCallback, times(2))
+				.onAfterConvert(eq(new Person("init", "luke")), eq(lukeDocument()), any());
+		assertThat(results.get(0).id).isEqualTo("after-convert");
+		assertThat(results.get(1).id).isEqualTo("after-convert");
+	}
+
+	private Query scrollingQueryForTwo() {
+		return new NativeSearchQueryBuilder()
+				.withIds(Arrays.asList("init1", "init2"))
+				.build();
+	}
+
+	@Test // DATAES-772
+	void findWithIndexCoordinatesShouldInvokeAfterConvertCallbacks() {
+
+		ValueCapturingAfterConvertCallback afterConvertCallback = spy(new ValueCapturingAfterConvertCallback());
+
+		template.setEntityCallbacks(ReactiveEntityCallbacks.create(afterConvertCallback));
+
+		@SuppressWarnings("deprecation") // we know what we test
+		List<Person> results = template.find(pagedQueryForTwo(), Person.class, index)
+				.timeout(Duration.ofSeconds(1)).toStream()
+				.collect(Collectors.toList());
+
+		verify(afterConvertCallback, times(2))
+				.onAfterConvert(eq(new Person("init", "luke")), eq(lukeDocument()), eq(index));
+		assertThat(results.get(0).id).isEqualTo("after-convert");
+		assertThat(results.get(1).id).isEqualTo("after-convert");
+	}
+
+	@Test // DATAES-772
+	void findWithReturnTypeShouldInvokeAfterConvertCallbacks() {
+
+		ValueCapturingAfterConvertCallback afterConvertCallback = spy(new ValueCapturingAfterConvertCallback());
+
+		template.setEntityCallbacks(ReactiveEntityCallbacks.create(afterConvertCallback));
+
+		@SuppressWarnings("deprecation") // we know what we test
+		List<Person> results = template.find(pagedQueryForTwo(), Person.class, Person.class)
+				.timeout(Duration.ofSeconds(1)).toStream()
+				.collect(Collectors.toList());
+
+		verify(afterConvertCallback, times(2))
+				.onAfterConvert(eq(new Person("init", "luke")), eq(lukeDocument()), any());
+		assertThat(results.get(0).id).isEqualTo("after-convert");
+		assertThat(results.get(1).id).isEqualTo("after-convert");
+	}
+
+	@Test // DATAES-772
+	void findWithReturnTypeAndIndexCoordinatesShouldInvokeAfterConvertCallbacks() {
+
+		ValueCapturingAfterConvertCallback afterConvertCallback = spy(new ValueCapturingAfterConvertCallback());
+
+		template.setEntityCallbacks(ReactiveEntityCallbacks.create(afterConvertCallback));
+
+		@SuppressWarnings("deprecation") // we know what we test
+		List<Person> results = template.find(pagedQueryForTwo(), Person.class, Person.class, index)
+				.timeout(Duration.ofSeconds(1)).toStream()
+				.collect(Collectors.toList());
+
+		verify(afterConvertCallback, times(2))
+				.onAfterConvert(eq(new Person("init", "luke")), eq(lukeDocument()), eq(index));
+		assertThat(results.get(0).id).isEqualTo("after-convert");
+		assertThat(results.get(1).id).isEqualTo("after-convert");
+	}
+
+	@Test // DATAES-772
+	void searchShouldInvokeAfterConvertCallbacks() {
+
+		ValueCapturingAfterConvertCallback afterConvertCallback = spy(new ValueCapturingAfterConvertCallback());
+
+		template.setEntityCallbacks(ReactiveEntityCallbacks.create(afterConvertCallback));
+
+		List<SearchHit<Person>> results = template.search(pagedQueryForTwo(), Person.class)
+				.timeout(Duration.ofSeconds(1)).toStream()
+				.collect(Collectors.toList());
+
+		verify(afterConvertCallback, times(2))
+				.onAfterConvert(eq(new Person("init", "luke")), eq(lukeDocument()), any());
+		assertThat(results.get(0).getContent().id).isEqualTo("after-convert");
+		assertThat(results.get(1).getContent().id).isEqualTo("after-convert");
+	}
+
+	@Test // DATAES-772
+	void searchWithIndexCoordinatesShouldInvokeAfterConvertCallbacks() {
+
+		ValueCapturingAfterConvertCallback afterConvertCallback = spy(new ValueCapturingAfterConvertCallback());
+
+		template.setEntityCallbacks(ReactiveEntityCallbacks.create(afterConvertCallback));
+
+		List<SearchHit<Person>> results = template.search(pagedQueryForTwo(), Person.class, index)
+				.timeout(Duration.ofSeconds(1)).toStream()
+				.collect(Collectors.toList());
+
+		verify(afterConvertCallback, times(2))
+				.onAfterConvert(eq(new Person("init", "luke")), eq(lukeDocument()), eq(index));
+		assertThat(results.get(0).getContent().id).isEqualTo("after-convert");
+		assertThat(results.get(1).getContent().id).isEqualTo("after-convert");
+	}
+
+	@Test // DATAES-772
+	void searchWithResultTypeShouldInvokeAfterConvertCallbacks() {
+
+		ValueCapturingAfterConvertCallback afterConvertCallback = spy(new ValueCapturingAfterConvertCallback());
+
+		template.setEntityCallbacks(ReactiveEntityCallbacks.create(afterConvertCallback));
+
+		List<SearchHit<Person>> results = template.search(pagedQueryForTwo(), Person.class, Person.class)
+				.timeout(Duration.ofSeconds(1)).toStream()
+				.collect(Collectors.toList());
+
+		verify(afterConvertCallback, times(2))
+				.onAfterConvert(eq(new Person("init", "luke")), eq(lukeDocument()), any());
+		assertThat(results.get(0).getContent().id).isEqualTo("after-convert");
+		assertThat(results.get(1).getContent().id).isEqualTo("after-convert");
+	}
+
+	@Test // DATAES-772
+	void searchWithResultTypeAndIndexCoordinatesShouldInvokeAfterConvertCallbacks() {
+
+		ValueCapturingAfterConvertCallback afterConvertCallback = spy(new ValueCapturingAfterConvertCallback());
+
+		template.setEntityCallbacks(ReactiveEntityCallbacks.create(afterConvertCallback));
+
+		List<SearchHit<Person>> results = template.search(pagedQueryForTwo(), Person.class, Person.class, index)
+				.timeout(Duration.ofSeconds(1)).toStream()
+				.collect(Collectors.toList());
+
+		verify(afterConvertCallback, times(2))
+				.onAfterConvert(eq(new Person("init", "luke")), eq(lukeDocument()), eq(index));
+		assertThat(results.get(0).getContent().id).isEqualTo("after-convert");
+		assertThat(results.get(1).getContent().id).isEqualTo("after-convert");
+	}
+
 	@Data
 	@AllArgsConstructor
 	@NoArgsConstructor
@@ -221,4 +509,23 @@ public class ReactiveElasticsearchTemplateCallbackTests {
 			});
 		}
 	}
+
+	static class ValueCapturingAfterConvertCallback extends ValueCapturingEntityCallback<Person>
+			implements ReactiveAfterConvertCallback<Person> {
+
+		@Override
+		public Mono<Person> onAfterConvert(Person entity, Document document, IndexCoordinates index) {
+
+			return Mono.defer(() -> {
+				capture(entity);
+				Person newPerson = new Person() {
+					{
+						id = "after-convert";
+						firstname = entity.firstname;
+					}
+				};
+				return Mono.just(newPerson);
+			});
+		}
+	}
 }

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

@@ -0,0 +1,89 @@
+/*
+ * Copyright 2020 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.elasticsearch.core;
+
+import static java.util.Collections.*;
+import static org.assertj.core.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+import java.util.Arrays;
+import java.util.Iterator;
+
+import org.elasticsearch.search.aggregations.Aggregations;
+import org.junit.jupiter.api.Test;
+import org.springframework.data.util.CloseableIterator;
+
+/**
+ * @author Roman Puchkovskiy
+ */
+class SearchHitSupportTest {
+	@Test // DATAES-772
+	void unwrapsSearchHitsIteratorToCloseableIteratorOfEntities() {
+		TestStringSearchHitsIterator searchHitsIterator = new TestStringSearchHitsIterator();
+
+		@SuppressWarnings("unchecked") CloseableIterator<String> unwrappedIterator =
+				(CloseableIterator<String>) SearchHitSupport.unwrapSearchHits(searchHitsIterator);
+
+		assertThat(unwrappedIterator.next()).isEqualTo("one");
+		assertThat(unwrappedIterator.next()).isEqualTo("two");
+		assertThat(unwrappedIterator.hasNext()).isFalse();
+
+		unwrappedIterator.close();
+
+		assertThat(searchHitsIterator.closed).isTrue();
+	}
+
+	private static class TestStringSearchHitsIterator implements SearchHitsIterator<String> {
+		private final Iterator<String> iterator = Arrays.asList("one", "two").iterator();
+		private boolean closed = false;
+
+		@Override
+		public Aggregations getAggregations() {
+			return mock(Aggregations.class);
+		}
+
+		@Override
+		public float getMaxScore() {
+			return 0;
+		}
+
+		@Override
+		public long getTotalHits() {
+			return 2;
+		}
+
+		@Override
+		public TotalHitsRelation getTotalHitsRelation() {
+			return mock(TotalHitsRelation.class);
+		}
+
+		@Override
+		public void close() {
+			closed = true;
+		}
+
+		@Override
+		public boolean hasNext() {
+			return iterator.hasNext();
+		}
+
+		@Override
+		public SearchHit<String> next() {
+			String nextString = iterator.next();
+			return new SearchHit<>("id", 1.0f, new Object[0], emptyMap(), nextString);
+		}
+	}
+}