Fix inner hits metadata mapping.

Closes spring-projects#2521

Author: JKatzwinkel

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

@@ -45,6 +45,7 @@
  * @author Roman Puchkovskiy
  * @author Matt Gilene
  * @author Sascha Woo
+ * @author Jakob Hoeper
  * @since 4.0
  */
 public class SearchHitMapping<T> {
@@ -154,7 +155,8 @@ private Map<String, List<String>> getHighlightsAndRemapFieldNames(SearchDocument
 		}
 
 		return highlightFields.entrySet().stream().collect(Collectors.toMap(entry -> {
-			ElasticsearchPersistentProperty property = persistentEntity.getPersistentPropertyWithFieldName(entry.getKey());
+			ElasticsearchPersistentProperty property = persistentEntity
+					.getPersistentPropertyWithFieldName(entry.getKey());
 			return property != null ? property.getName() : entry.getKey();
 		}, Map.Entry::getValue));
 	}
@@ -199,9 +201,10 @@ private SearchHits<?> mapInnerDocuments(SearchHits<SearchDocument> searchHits, C
 		}
 
 		try {
+			ElasticsearchPersistentEntity<?> persistentEntityForType = mappingContext.getPersistentEntity(type);
 			NestedMetaData nestedMetaData = searchHits.getSearchHit(0).getContent().getNestedMetaData();
 			ElasticsearchPersistentEntityWithNestedMetaData persistentEntityWithNestedMetaData = getPersistentEntity(
-					mappingContext.getPersistentEntity(type), nestedMetaData);
+					persistentEntityForType, nestedMetaData);
 
 			if (persistentEntityWithNestedMetaData.entity != null) {
 				List<SearchHit<Object>> convertedSearchHits = new ArrayList<>();
@@ -219,7 +222,8 @@ private SearchHits<?> mapInnerDocuments(SearchHits<SearchDocument> searchHits, C
 							searchDocument.getSortValues(), //
 							searchDocument.getHighlightFields(), //
 							searchHit.getInnerHits(), //
-							persistentEntityWithNestedMetaData.nestedMetaData, //
+							getPersistentEntity(persistentEntityForType, //
+									searchHit.getContent().getNestedMetaData()).nestedMetaData, //
 							searchHit.getExplanation(), //
 							searchHit.getMatchedQueries(), //
 							targetObject));

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

@@ -42,6 +42,7 @@
  * Testing the querying and parsing of inner_hits.
  *
  * @author Peter-Josef Meisch
+ * @author Jakob Hoeper
  */
 @SpringIntegrationTest
 public abstract class InnerHitsIntegrationTests {
@@ -58,8 +59,9 @@ void setUp() {
 		indexOps.createWithMapping();
 
 		Inhabitant john = new Inhabitant("John", "Smith");
-		Inhabitant carla = new Inhabitant("Carla", "Miller");
-		House cornerHouse = new House("Round the corner", "7", Arrays.asList(john, carla));
+		Inhabitant carla1 = new Inhabitant("Carla", "Miller");
+		Inhabitant carla2 = new Inhabitant("Carla", "Nguyen");
+		House cornerHouse = new House("Round the corner", "7", Arrays.asList(john, carla1, carla2));
 		City metropole = new City("Metropole", Arrays.asList(cornerHouse));
 
 		Inhabitant jack = new Inhabitant("Jack", "Wayne");
@@ -76,7 +78,7 @@ void cleanup() {
 		operations.indexOps(IndexCoordinates.of(indexNameProvider.getPrefix() + "*")).delete();
 	}
 
-	@Test
+	@Test // #2521
 	void shouldReturnInnerHits() {
 
 		Query query = buildQueryForInnerHits("inner_hit_name", "hou-ses.in-habi-tants", "hou-ses.in-habi-tants.first-name",
@@ -91,7 +93,7 @@ void shouldReturnInnerHits() {
 		softly.assertThat(searchHit.getInnerHits()).hasSize(1);
 
 		SearchHits<?> innerHits = searchHit.getInnerHits("inner_hit_name");
-		softly.assertThat(innerHits).hasSize(1);
+		softly.assertThat(innerHits).hasSize(2);
 
 		SearchHit<?> innerHit = innerHits.getSearchHit(0);
 		Object content = innerHit.getContent();
@@ -106,6 +108,10 @@ void shouldReturnInnerHits() {
 		softly.assertThat(nestedMetaData.getChild().getField()).isEqualTo("inhabitants");
 		softly.assertThat(nestedMetaData.getChild().getOffset()).isEqualTo(1);
 
+		innerHit = innerHits.getSearchHit(1);
+		softly.assertThat(((Inhabitant) innerHit.getContent()).getLastName()).isEqualTo("Nguyen");
+		softly.assertThat(innerHit.getNestedMetaData().getChild().getOffset()).isEqualTo(2);
+
 		softly.assertAll();
 	}