DATAES-530 - Add converter based EntityMapper.

ElasticsearchEntityMapper provides an alternative to the Jackson based EntityMapper implementation.

Original Pull Request: #237

src/main/asciidoc/index.adoc

@@ -29,6 +29,7 @@ include::{spring-data-commons-docs}/repositories.adoc[]
 
 :leveloffset: +1
 include::reference/elasticsearch-clients.adoc[]
+include::reference/elasticsearch-object-mapping.adoc[]
 include::reference/data-elasticsearch.adoc[]
 include::reference/reactive-elasticsearch-operations.adoc[]
 include::reference/reactive-elasticsearch-repositories.adoc[]

src/main/asciidoc/reference/elasticsearch-object-mapping.adoc

@@ -0,0 +1,287 @@
+[[elasticsearch.mapping]]
+= Elasticsearch Object Mapping
+
+Spring Data Elasticsearch allows to choose between two mapping implementations abstracted via the `EntityMapper` interface:
+
+* <<elasticsearch.mapping.jackson2>>
+* <<elasticsearch.mapping.meta-model>>
+
+[[elasticsearch.mapping.jackson2]]
+== Jackson Object Mapping
+
+The Jackson2 based approach (used by default) utilizes a customized `ObjectMapper` instance with spring data specific modules.
+Extensions to the actual mapping need to be customized via Jackson annotations like `@JsonInclude`.
+
+.Jackson2 Object Mapping Configuration
+====
+[source,java]
+----
+@Configuration
+public class Config extends AbstractElasticsearchConfiguration { <1>
+
+  @Override
+  public RestHighLevelClient elasticsearchClient() {
+    return RestClients.create(ClientConfiguration.create("localhost:9200")).rest()
+  }
+}
+----
+<1> `AbstractElasticsearchConfiguration` already defines a Jackson2 based `entityMapper` via `ElasticsearchConfigurationSupport`.
+====
+
+WARNING: `CustomConversions`, `@ReadingConverter` & `@WritingConverter` cannot be applied when using the Jackson based `EntityMapper`.
+
+[[elasticsearch.mapping.meta-model]]
+== Meta Model Object Mapping
+
+The Metamodel based approach uses domain type information for reading/writing from/to Elasticsearch.
+This allows to register `Converter` instances for specific domain type mapping.
+
+.Meta Model Object Mapping Configuration
+====
+[source,java]
+----
+@Configuration
+public class Config extends AbstractElasticsearchConfiguration {
+
+  @Override
+  public RestHighLevelClient elasticsearchClient() {
+    return RestClients.create(ClientConfiguration.create("localhost:9200")).rest()
+  }
+
+  @Bean
+  @Override
+  public EntityMapper entityMapper() { <1>
+
+    ElasticsearchEntityMapper entityMapper = new ElasticsearchEntityMapper(elasticsearchMappingContext(),
+      new DefaultConversionService()); <2>
+  	entityMapper.setConversions(elasticsearchCustomConversions()); <3>
+
+  	return entityMapper;
+  }
+}
+----
+<1> Overwrite the default `EntityMapper` from `ElasticsearchConfigurationSupport` and expose it as bean.
+<2> Use the provided `SimpleElasticsearchMappingContext` to avoid inconsistencies and provide a `GenericConversionService`
+for `Converter` registration.
+<3> Optionally set `CustomConversions` if applicable.
+====
+
+[[elasticsearch.mapping.meta-model.rules]]
+=== Mapping Rules
+
+==== Type Hints
+
+Mapping uses _type hints_ embedded in the document sent to the server to allow generic type mapping. Those type hints are
+represented as `_class` attributes within the document on the server and will be written for each aggregate root.
+
+.Type Hints
+====
+[source,java]
+----
+public class Person { <1>
+
+  @Id String id;
+  String firstname;
+  String lastname;
+}
+----
+[source,json]
+----
+{
+  "_class" : "com.example.Person", <1>
+  "id" : "cb7bef",
+  "firstname" : "Sarah",
+  "lastname" : "Connor"
+}
+----
+<1> By default the domain types class name is used for the type hint.
+====
+
+Type hints can be configured to hold custom information. Use the `@TypeAlias` annotation to do so.
+
+NOTE: Make sure to add types with `@TypeAlias` to the initial entity set (`AbstractElasticsearchConfiguration#getInitialEntitySet`)
+to already have entity information available when first reading data from the store.
+
+.Type Hints with Alias
+====
+[source,java]
+----
+@TypeAlias("human") <1>
+public class Person {
+
+  @Id String id;
+  // ...
+}
+----
+[source,json]
+----
+{
+  "_class" : "human", <1>
+  "id" : ...
+}
+----
+<1> The configured alias is used when writing the entity.
+====
+
+NOTE: Type hints will not be written for nested Objects unless the properties type is `Object`, an interface or the
+actual value type does not match the properties declaration.
+
+==== Geospatial Types
+
+Geospatial types like `Point` & `GeoPoint` are converted into _lat/lon_ pairs.
+
+.Geospatial types
+====
+[source,java]
+----
+public class Address {
+
+  String city, street;
+  Point location;
+}
+----
+[source,json]
+----
+{
+  "city" : "Los Angeles",
+  "street" : "2800 East Observatory Road",
+  "location" : { "lat" : 34.118347, "lon" : -118.3026284 }
+}
+----
+====
+
+==== Collections
+
+For values inside Collections apply the same mapping rules as for aggregate roots when it comes to _type hints_
+and <<elasticsearch.mapping.meta-model.conversions>>.
+
+.Collections
+====
+[source,java]
+----
+public class Person {
+
+  // ...
+
+  List<Person> friends;
+
+}
+----
+[source,json]
+----
+{
+  // ...
+
+  "friends" : [ { "firstname" : "Kyle", "lastname" : "Reese" } ]
+}
+----
+====
+
+==== Maps
+
+For values inside Maps apply the same mapping rules as for aggregate roots when it comes to _type hints_
+and <<elasticsearch.mapping.meta-model.conversions>>.
+However the Map key needs to a String to be processed by Elasticsearch.
+
+.Collections
+====
+[source,java]
+----
+public class Person {
+
+  // ...
+
+  Map<String, Address> knownLocations;
+
+}
+----
+[source,json]
+----
+{
+  // ...
+
+  "knownLocations" : {
+    "arrivedAt" : {
+       "city" : "Los Angeles",
+       "street" : "2800 East Observatory Road",
+       "location" : { "lat" : 34.118347, "lon" : -118.3026284 }
+     }
+  }
+}
+----
+====
+
+[[elasticsearch.mapping.meta-model.conversions]]
+=== Custom Conversions
+
+Looking at the `Configuration` from the <<elasticsearch.mapping.meta-model, previous section>> `ElasticsearchCustomConversions`
+allows to register specific rules for mapping domain and simple types.
+
+.Meta Model Object Mapping Configuration
+====
+[source,java]
+----
+@Configuration
+public class Config extends AbstractElasticsearchConfiguration {
+
+  @Override
+  public RestHighLevelClient elasticsearchClient() {
+    return RestClients.create(ClientConfiguration.create("localhost:9200")).rest()
+  }
+
+  @Bean
+  @Override
+  public EntityMapper entityMapper() {
+
+    ElasticsearchEntityMapper entityMapper = new ElasticsearchEntityMapper(elasticsearchMappingContext(),
+      new DefaultConversionService());
+  	entityMapper.setConversions(elasticsearchCustomConversions()); <1>
+
+  	return entityMapper;
+  }
+
+  @Bean
+  @Override
+  public ElasticsearchCustomConversions elasticsearchCustomConversions() {
+    return new ElasticsearchCustomConversions(Arrays.asList(new AddressToMap(), new MapToAddress())); <2>
+  }
+
+  @WritingConverter <3>
+  static class AddressToMap implements Converter<Address, Map<String, Object>> {
+
+    @Override
+    public Map<String, Object> convert(Address source) {
+
+      LinkedHashMap<String, Object> target = new LinkedHashMap<>();
+      target.put("ciudad", source.getCity());
+      // ...
+
+      return target;
+    }
+  }
+
+  @ReadingConverter <4>
+  static class MapToAddress implements Converter<Map<String, Object>, Address> {
+
+    @Override
+    public Address convert(Map<String, Object> source) {
+
+      // ...
+      return address;
+    }
+  }
+}
+----
+[source,json]
+----
+{
+  "ciudad" : "Los Angeles",
+  "calle" : "2800 East Observatory Road",
+  "localidad" : { "lat" : 34.118347, "lon" : -118.3026284 }
+}
+----
+<1> Register `ElasticsearchCustomConversions` with the `EntityMapper`.
+<2> Add `Converter` implementations.
+<3> Set up the `Converter` used for writing `DomainType` to Elasticsearch.
+<4> Set up the `Converter` used for reading `DomainType` from search result.
+====

src/main/java/org/springframework/data/elasticsearch/config/AbstractElasticsearchConfiguration.java

@@ -43,6 +43,6 @@ public abstract class AbstractElasticsearchConfiguration extends ElasticsearchCo
 	 */
 	@Bean
 	public ElasticsearchOperations elasticsearchOperations() {
-		return new ElasticsearchRestTemplate(elasticsearchClient(), elasticsearchConverter());
+		return new ElasticsearchRestTemplate(elasticsearchClient(), elasticsearchConverter(), resultsMapper());
 	}
 }

src/main/java/org/springframework/data/elasticsearch/config/AbstractReactiveElasticsearchConfiguration.java

@@ -50,7 +50,7 @@ public abstract class AbstractReactiveElasticsearchConfiguration extends Elastic
 	public ReactiveElasticsearchOperations reactiveElasticsearchTemplate() {
 
 		ReactiveElasticsearchTemplate template = new ReactiveElasticsearchTemplate(reactiveElasticsearchClient(),
-				elasticsearchConverter());
+				elasticsearchConverter(), resultsMapper());
 		template.setIndicesOptions(indicesOptions());
 		template.setRefreshPolicy(refreshPolicy());
 

src/main/java/org/springframework/data/elasticsearch/config/ElasticsearchConfigurationSupport.java

@@ -30,6 +30,10 @@ import org.springframework.core.convert.converter.Converter;
 import org.springframework.core.type.filter.AnnotationTypeFilter;
 import org.springframework.data.annotation.Persistent;
 import org.springframework.data.elasticsearch.annotations.Document;
+import org.springframework.data.elasticsearch.core.DefaultEntityMapper;
+import org.springframework.data.elasticsearch.core.DefaultResultMapper;
+import org.springframework.data.elasticsearch.core.EntityMapper;
+import org.springframework.data.elasticsearch.core.ResultsMapper;
 import org.springframework.data.elasticsearch.core.convert.ElasticsearchConverter;
 import org.springframework.data.elasticsearch.core.convert.ElasticsearchCustomConversions;
 import org.springframework.data.elasticsearch.core.convert.MappingElasticsearchConverter;
@@ -62,18 +66,48 @@ public class ElasticsearchConfigurationSupport {
 
 		SimpleElasticsearchMappingContext mappingContext = new SimpleElasticsearchMappingContext();
 		mappingContext.setInitialEntitySet(getInitialEntitySet());
-		mappingContext.setSimpleTypeHolder(customConversions().getSimpleTypeHolder());
+		mappingContext.setSimpleTypeHolder(elasticsearchCustomConversions().getSimpleTypeHolder());
 
 		return mappingContext;
 	}
 
+	/**
+	 * Returns the {@link EntityMapper} used for mapping source &lt;&gt; DomainType. <br />
+	 * <strong>Hint</strong>: you can use {@link org.springframework.data.elasticsearch.core.ElasticsearchEntityMapper} as
+	 * an alternative to the {@link DefaultEntityMapper}.
+	 *
+	 * <pre>{@code
+	 * ElasticsearchEntityMapper entityMapper = new ElasticsearchEntityMapper(elasticsearchMappingContext(),
+	 * 			new DefaultConversionService());
+	 * entityMapper.setConversions(elasticsearchCustomConversions());
+	 * }
+	 * </pre>
+	 *
+	 * @return never {@literal null}.
+	 */
+	@Bean
+	public EntityMapper entityMapper() {
+		return new DefaultEntityMapper(elasticsearchMappingContext());
+	}
+
+	/**
+	 * Returns the {@link ResultsMapper} to be used for search responses.
+	 *
+	 * @see #entityMapper()
+	 * @return never {@literal null}.
+	 */
+	@Bean
+	public ResultsMapper resultsMapper() {
+		return new DefaultResultMapper(elasticsearchMappingContext(), entityMapper());
+	}
+
 	/**
 	 * Register custom {@link Converter}s in a {@link ElasticsearchCustomConversions} object if required.
 	 *
 	 * @return never {@literal null}.
 	 */
 	@Bean
-	public ElasticsearchCustomConversions customConversions() {
+	public ElasticsearchCustomConversions elasticsearchCustomConversions() {
 		return new ElasticsearchCustomConversions(Collections.emptyList());
 	}
 

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

@@ -15,24 +15,49 @@
  */
 package org.springframework.data.elasticsearch.core;
 
+import org.springframework.data.projection.ProjectionFactory;
+import org.springframework.data.projection.SpelAwareProxyProjectionFactory;
 import org.springframework.util.Assert;
 
 /**
  * @author Artur Konczak
+ * @author Christoph Strobl
  */
 public abstract class AbstractResultMapper implements ResultsMapper {
 
-	private EntityMapper entityMapper;
+	private final EntityMapper entityMapper;
+	private final ProjectionFactory projectionFactory;
 
 	public AbstractResultMapper(EntityMapper entityMapper) {
+		this(entityMapper, new SpelAwareProxyProjectionFactory());
+	}
+
+	/**
+	 *
+	 * @param entityMapper
+	 * @param projectionFactory
+	 * @since 3.2
+	 */
+	public AbstractResultMapper(EntityMapper entityMapper, ProjectionFactory projectionFactory) {
 
 		Assert.notNull(entityMapper, "EntityMapper must not be null!");
+		Assert.notNull(projectionFactory, "ProjectionFactory must not be null!");
 
 		this.entityMapper = entityMapper;
+		this.projectionFactory = projectionFactory;
 	}
 
 	@Override
 	public EntityMapper getEntityMapper() {
 		return this.entityMapper;
 	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.elasticsearch.core.ResultsMapper#getProjectionFactory()
+	 */
+	@Override
+	public ProjectionFactory getProjectionFactory() {
+		return projectionFactory;
+	}
 }

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

@@ -17,12 +17,15 @@ package org.springframework.data.elasticsearch.core;
 
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 
 import org.springframework.data.annotation.ReadOnlyProperty;
 import org.springframework.data.elasticsearch.core.geo.CustomGeoModule;
 import org.springframework.data.elasticsearch.core.mapping.ElasticsearchPersistentEntity;
 import org.springframework.data.elasticsearch.core.mapping.ElasticsearchPersistentProperty;
+import org.springframework.data.mapping.MappingException;
 import org.springframework.data.mapping.context.MappingContext;
 import org.springframework.util.Assert;
 
@@ -40,6 +43,7 @@ import com.fasterxml.jackson.databind.ser.BeanSerializerModifier;
  * @author Artur Konczak
  * @author Petar Tahchiev
  * @author Oliver Gierke
+ * @author Christoph Strobl
  */
 public class DefaultEntityMapper implements EntityMapper {
 
@@ -52,14 +56,14 @@ public class DefaultEntityMapper implements EntityMapper {
 	 */
 	public DefaultEntityMapper(
 			MappingContext<? extends ElasticsearchPersistentEntity<?>, ElasticsearchPersistentProperty> context) {
-		
+
 		Assert.notNull(context, "MappingContext must not be null!");
 
 		objectMapper = new ObjectMapper();
-		
+
 		objectMapper.registerModule(new SpringDataElasticsearchModule(context));
 		objectMapper.registerModule(new CustomGeoModule());
-		
+
 		objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
 		objectMapper.configure(DeserializationFeature.ACCEPT_SINGLE_VALUE_AS_ARRAY, true);
 	}
@@ -73,6 +77,20 @@ public class DefaultEntityMapper implements EntityMapper {
 		return objectMapper.writeValueAsString(object);
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.elasticsearch.core.EntityMapper#mapObject(java.lang.Object)
+	 */
+	@Override
+	public Map<String, Object> mapObject(Object source) {
+
+		try {
+			return objectMapper.readValue(mapToString(source), HashMap.class);
+		} catch (IOException e) {
+			throw new MappingException(e.getMessage(), e);
+		}
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.elasticsearch.core.EntityMapper#mapToObject(java.lang.String, java.lang.Class)
@@ -82,6 +100,20 @@ public class DefaultEntityMapper implements EntityMapper {
 		return objectMapper.readValue(source, clazz);
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.elasticsearch.core.EntityMapper#readObject(java.util.Map, java.lang.Class)
+	 */
+	@Override
+	public <T> T readObject (Map<String, Object> source, Class<T> targetType) {
+
+		try {
+			return mapToObject(mapToString(source), targetType);
+		} catch (IOException e) {
+			throw new MappingException(e.getMessage(), e);
+		}
+	}
+
 	/**
 	 * A simple Jackson module to register the {@link SpringDataSerializerModifier}.
 	 *
@@ -101,7 +133,7 @@ public class DefaultEntityMapper implements EntityMapper {
 				MappingContext<? extends ElasticsearchPersistentEntity<?>, ElasticsearchPersistentProperty> context) {
 
 			Assert.notNull(context, "MappingContext must not be null!");
-			
+
 			setSerializerModifier(new SpringDataSerializerModifier(context));
 		}
 

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

@@ -43,12 +43,13 @@ import org.springframework.data.elasticsearch.core.mapping.SimpleElasticsearchMa
 import org.springframework.data.mapping.PersistentPropertyAccessor;
 import org.springframework.data.mapping.context.MappingContext;
 import org.springframework.data.mapping.model.ConvertingPropertyAccessor;
+import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
+import org.springframework.util.StringUtils;
 
 import com.fasterxml.jackson.core.JsonEncoding;
 import com.fasterxml.jackson.core.JsonFactory;
 import com.fasterxml.jackson.core.JsonGenerator;
-import org.springframework.util.StringUtils;
 
 /**
  * @author Artur Konczak
@@ -59,24 +60,20 @@ import org.springframework.util.StringUtils;
  * @author Mark Paluch
  * @author Ilkang Na
  * @author Sascha Woo
+ * @author Christoph Strobl
  */
 public class DefaultResultMapper extends AbstractResultMapper {
 
 	private final MappingContext<? extends ElasticsearchPersistentEntity<?>, ElasticsearchPersistentProperty> mappingContext;
-
 	private final ConversionService conversionService = new DefaultConversionService();
 
 	public DefaultResultMapper() {
 		this(new SimpleElasticsearchMappingContext());
 	}
 
-	public DefaultResultMapper(MappingContext<? extends ElasticsearchPersistentEntity<?>, ElasticsearchPersistentProperty> mappingContext) {
-		
-		super(new DefaultEntityMapper(mappingContext));
-		
-		Assert.notNull(mappingContext, "MappingContext must not be null!");
-		
-		this.mappingContext = mappingContext;
+	public DefaultResultMapper(
+			MappingContext<? extends ElasticsearchPersistentEntity<?>, ElasticsearchPersistentProperty> mappingContext) {
+		this(mappingContext, initEntityMapper(mappingContext));
 	}
 
 	public DefaultResultMapper(EntityMapper entityMapper) {
@@ -85,18 +82,23 @@ public class DefaultResultMapper extends AbstractResultMapper {
 
 	public DefaultResultMapper(
 			MappingContext<? extends ElasticsearchPersistentEntity<?>, ElasticsearchPersistentProperty> mappingContext,
-			EntityMapper entityMapper) {
-		
-		super(entityMapper);
-		
-		Assert.notNull(mappingContext, "MappingContext must not be null!");
-		
+			@Nullable EntityMapper entityMapper) {
+
+		super(entityMapper != null ? entityMapper : initEntityMapper(mappingContext));
 		this.mappingContext = mappingContext;
 	}
 
+	static EntityMapper initEntityMapper(
+			MappingContext<? extends ElasticsearchPersistentEntity<?>, ElasticsearchPersistentProperty> mappingContext) {
+
+		Assert.notNull(mappingContext, "MappingContext must not be null!");
+		return new DefaultEntityMapper(mappingContext);
+
+	}
+
 	@Override
 	public <T> AggregatedPage<T> mapResults(SearchResponse response, Class<T> clazz, Pageable pageable) {
-		
+
 		long totalHits = response.getHits().getTotalHits();
 		float maxScore = response.getHits().getMaxScore();
 
@@ -113,7 +115,7 @@ public class DefaultResultMapper extends AbstractResultMapper {
 				setPersistentEntityId(result, hit.getId(), clazz);
 				setPersistentEntityVersion(result, hit.getVersion(), clazz);
 				setPersistentEntityScore(result, hit.getScore(), clazz);
-				
+
 				populateScriptFields(result, hit);
 				results.add(result);
 			}
@@ -200,14 +202,14 @@ public class DefaultResultMapper extends AbstractResultMapper {
 	}
 
 	private <T> void setPersistentEntityId(T result, String id, Class<T> clazz) {
-		
+
 		if (clazz.isAnnotationPresent(Document.class)) {
-			
+
 			ElasticsearchPersistentEntity<?> persistentEntity = mappingContext.getRequiredPersistentEntity(clazz);
 			ElasticsearchPersistentProperty idProperty = persistentEntity.getIdProperty();
 
-			PersistentPropertyAccessor<T> accessor = new ConvertingPropertyAccessor<>(persistentEntity.getPropertyAccessor(result),
-					conversionService);
+			PersistentPropertyAccessor<T> accessor = new ConvertingPropertyAccessor<>(
+					persistentEntity.getPropertyAccessor(result), conversionService);
 
 			// Only deal with String because ES generated Ids are strings !
 			if (idProperty != null && idProperty.getType().isAssignableFrom(String.class)) {
@@ -217,9 +219,9 @@ public class DefaultResultMapper extends AbstractResultMapper {
 	}
 
 	private <T> void setPersistentEntityVersion(T result, long version, Class<T> clazz) {
-		
+
 		if (clazz.isAnnotationPresent(Document.class)) {
-			
+
 			ElasticsearchPersistentEntity<?> persistentEntity = mappingContext.getPersistentEntity(clazz);
 			ElasticsearchPersistentProperty versionProperty = persistentEntity.getVersionProperty();
 

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

@@ -0,0 +1,683 @@
+/*
+ * Copyright 2019 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
+ *
+ *      http://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 java.io.IOException;
+import java.util.*;
+import java.util.Map.Entry;
+
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.core.CollectionFactory;
+import org.springframework.core.convert.ConversionService;
+import org.springframework.core.convert.support.DefaultConversionService;
+import org.springframework.core.convert.support.GenericConversionService;
+import org.springframework.data.convert.CustomConversions;
+import org.springframework.data.convert.EntityConverter;
+import org.springframework.data.convert.EntityInstantiator;
+import org.springframework.data.convert.EntityInstantiators;
+import org.springframework.data.convert.EntityReader;
+import org.springframework.data.convert.EntityWriter;
+import org.springframework.data.elasticsearch.core.convert.ElasticsearchCustomConversions;
+import org.springframework.data.elasticsearch.core.convert.ElasticsearchTypeMapper;
+import org.springframework.data.elasticsearch.core.mapping.ElasticsearchPersistentEntity;
+import org.springframework.data.elasticsearch.core.mapping.ElasticsearchPersistentProperty;
+import org.springframework.data.mapping.MappingException;
+import org.springframework.data.mapping.PersistentPropertyAccessor;
+import org.springframework.data.mapping.context.MappingContext;
+import org.springframework.data.mapping.model.ConvertingPropertyAccessor;
+import org.springframework.data.mapping.model.PersistentEntityParameterValueProvider;
+import org.springframework.data.mapping.model.PropertyValueProvider;
+import org.springframework.data.util.ClassTypeInformation;
+import org.springframework.data.util.Streamable;
+import org.springframework.data.util.TypeInformation;
+import org.springframework.format.datetime.DateFormatterRegistrar;
+import org.springframework.lang.Nullable;
+import org.springframework.util.ClassUtils;
+import org.springframework.util.ObjectUtils;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+/**
+ * Elasticsearch specific {@link EntityReader} & {@link EntityWriter} implementation based on domain type
+ * {@link ElasticsearchPersistentEntity metadata}.
+ *
+ * @author Christoph Strobl
+ * @since 3.2
+ */
+public class ElasticsearchEntityMapper implements
+		EntityConverter<ElasticsearchPersistentEntity<?>, ElasticsearchPersistentProperty, Object, Map<String, Object>>,
+		EntityWriter<Object, Map<String, Object>>, EntityReader<Object, Map<String, Object>>, InitializingBean,
+		EntityMapper {
+
+	private final MappingContext<? extends ElasticsearchPersistentEntity<?>, ElasticsearchPersistentProperty> mappingContext;
+	private final GenericConversionService conversionService;
+
+	private CustomConversions conversions = new ElasticsearchCustomConversions(Collections.emptyList());
+	private EntityInstantiators instantiators = new EntityInstantiators();
+
+	private ElasticsearchTypeMapper typeMapper;
+
+	public ElasticsearchEntityMapper(
+			MappingContext<? extends ElasticsearchPersistentEntity<?>, ElasticsearchPersistentProperty> mappingContext,
+			@Nullable GenericConversionService conversionService) {
+
+		this.mappingContext = mappingContext;
+		this.conversionService = conversionService != null ? conversionService : new DefaultConversionService();
+		this.typeMapper = ElasticsearchTypeMapper.defaultTypeMapper(mappingContext);
+	}
+
+	// --> READ
+
+	@Override
+	public <T> T readObject(Map<String, Object> source, Class<T> targetType) {
+		return read(targetType, source);
+	}
+
+	@Override
+	@Nullable
+	public <R> R read(Class<R> type, Map<String, Object> source) {
+		return doRead(source, ClassTypeInformation.from((Class<R>) ClassUtils.getUserClass(type)));
+	}
+
+	@Nullable
+	protected <R> R doRead(Map<String, Object> source, TypeInformation<R> typeHint) {
+
+		if (source == null) {
+			return null;
+		}
+
+		typeHint = (TypeInformation<R>) typeMapper.readType(source, typeHint);
+
+		if (conversions.hasCustomReadTarget(Map.class, typeHint.getType())) {
+			return conversionService.convert(source, typeHint.getType());
+		}
+
+		if (typeHint.isMap() || ClassTypeInformation.OBJECT.equals(typeHint)) {
+			return (R) source;
+		}
+
+		ElasticsearchPersistentEntity<?> entity = mappingContext.getRequiredPersistentEntity(typeHint);
+		return readEntity(entity, source);
+	}
+
+	protected <R> R readEntity(ElasticsearchPersistentEntity<?> entity, Map<String, Object> source) {
+
+		ElasticsearchPersistentEntity<?> targetEntity = computeClosestEntity(entity, source);
+
+		ElasticsearchPropertyValueProvider propertyValueProvider = new ElasticsearchPropertyValueProvider(
+				new MapValueAccessor(source));
+
+		EntityInstantiator instantiator = instantiators.getInstantiatorFor(targetEntity);
+
+		R instance = (R) instantiator.createInstance(targetEntity,
+				new PersistentEntityParameterValueProvider<>(targetEntity, propertyValueProvider, null));
+
+		if (targetEntity.requiresPropertyPopulation()) {
+			return readProperties(targetEntity, instance, propertyValueProvider);
+		}
+
+		return instance;
+	}
+
+	protected <R> R readProperties(ElasticsearchPersistentEntity<?> entity, R instance,
+			ElasticsearchPropertyValueProvider valueProvider) {
+
+		PersistentPropertyAccessor<R> accessor = new ConvertingPropertyAccessor<>(entity.getPropertyAccessor(instance),
+				conversionService);
+
+		for (ElasticsearchPersistentProperty prop : entity) {
+
+			if (entity.isConstructorArgument(prop) || prop.isScoreProperty()) {
+				continue;
+			}
+
+			Object value = valueProvider.getPropertyValue(prop);
+			if (value != null) {
+				accessor.setProperty(prop, valueProvider.getPropertyValue(prop));
+			}
+		}
+
+		return accessor.getBean();
+	}
+
+	protected <R> R readValue(@Nullable Object source, ElasticsearchPersistentProperty property,
+			TypeInformation<R> targetType) {
+
+		if (source == null) {
+			return null;
+		}
+
+		Class<?> rawType = targetType.getType();
+		if (conversions.hasCustomReadTarget(source.getClass(), rawType)) {
+			return (R) conversionService.convert(source, rawType);
+		} else if (source instanceof List) {
+			return readCollectionValue((List) source, property, targetType);
+		} else if (source instanceof Map) {
+			return readMapValue((Map<String, Object>) source, property, targetType);
+		} else if (Enum.class.isAssignableFrom(rawType)) {
+			return (R) Enum.valueOf((Class<Enum>) rawType, source.toString());
+		}
+
+		return (R) readSimpleValue(source, targetType);
+	}
+
+	private <R> R readMapValue(@Nullable Map<String, Object> source, ElasticsearchPersistentProperty property,
+			TypeInformation<R> targetType) {
+
+		TypeInformation information = typeMapper.readType(source);
+		if (property.isEntity() && !property.isMap() || information != null) {
+
+			ElasticsearchPersistentEntity<?> targetEntity = information != null
+					? mappingContext.getRequiredPersistentEntity(information)
+					: mappingContext.getRequiredPersistentEntity(property);
+			return readEntity(targetEntity, source);
+		}
+
+		Map<String, Object> target = new LinkedHashMap();
+		for (Entry<String, Object> entry : source.entrySet()) {
+
+			if (conversions.isSimpleType(entry.getValue().getClass())) {
+				target.put(entry.getKey(),
+						readSimpleValue(entry.getValue(), targetType.isMap() ? targetType.getComponentType() : targetType));
+			} else {
+
+				ElasticsearchPersistentEntity<?> targetEntity = computeGenericValueTypeForRead(property, entry.getValue());
+
+				if (targetEntity.getTypeInformation().isMap()) {
+
+					Map<String, Object> valueMap = (Map) entry.getValue();
+					if (typeMapper.containsTypeInformation(valueMap)) {
+						target.put(entry.getKey(), readEntity(targetEntity, (Map) entry.getValue()));
+					} else {
+						target.put(entry.getKey(), readValue(valueMap, property, targetEntity.getTypeInformation()));
+					}
+
+				} else if (targetEntity.getTypeInformation().isCollectionLike()) {
+					target.put(entry.getKey(),
+							readValue(entry.getValue(), property, targetEntity.getTypeInformation().getActualType()));
+				} else {
+					target.put(entry.getKey(), readEntity(targetEntity, (Map) entry.getValue()));
+				}
+
+			}
+		}
+
+		return (R) target;
+	}
+
+	private <R> R readCollectionValue(@Nullable List<?> source, ElasticsearchPersistentProperty property,
+			TypeInformation<R> targetType) {
+
+		if (source == null) {
+			return null;
+		}
+
+		List<?> sourceList = source;
+		Collection target = createCollectionForValue(targetType, sourceList.size());
+
+		for (Object value : sourceList) {
+
+			if (conversions.isSimpleType(value.getClass())) {
+				target.add(
+						readSimpleValue(value, targetType.getComponentType() != null ? targetType.getComponentType() : targetType));
+			} else {
+
+				if (value instanceof List) {
+					target.add(readValue(value, property, property.getTypeInformation().getActualType()));
+				} else {
+					target.add(readEntity(computeGenericValueTypeForRead(property, value), (Map) value));
+				}
+			}
+		}
+
+		return (R) target;
+	}
+
+	protected Object readSimpleValue(@Nullable Object value, TypeInformation<?> targetType) {
+
+		if (value == null) {
+			return null;
+		}
+
+		if (ClassTypeInformation.OBJECT.equals(targetType)
+				|| (targetType != null && value.getClass().equals(targetType.getActualType()))) {
+			return value;
+		}
+
+		if (conversionService.canConvert(value.getClass(), targetType.getType())) {
+			return conversionService.convert(value, targetType.getType());
+		}
+
+		throw new MappingException(
+				String.format("Unable to map %s of type %s to %s", value, value.getClass(), targetType.getType()));
+	}
+
+	// --> WRITE
+
+	@Override
+	public Map<String, Object> mapObject(Object source) {
+
+		LinkedHashMap<String, Object> target = new LinkedHashMap<>();
+		write(source, target);
+		return target;
+	}
+
+	@Override
+	public void write(@Nullable Object source, Map<String, Object> sink) {
+
+		if (source == null) {
+			return;
+		}
+
+		if (source instanceof Map) {
+
+			sink.putAll((Map) source);
+			return;
+		}
+
+		Class<?> entityType = ClassUtils.getUserClass(source.getClass());
+		TypeInformation<? extends Object> type = ClassTypeInformation.from(entityType);
+
+		if (requiresTypeHint(type, source.getClass(), null)) {
+			typeMapper.writeType(source.getClass(), sink);
+		}
+
+		doWrite(source, sink, type);
+	}
+
+	protected void doWrite(@Nullable Object source, Map<String, Object> sink,
+			@Nullable TypeInformation<? extends Object> typeHint) {
+
+		if (source == null) {
+			return;
+		}
+
+		Class<?> entityType = source.getClass();
+		Optional<Class<?>> customTarget = conversions.getCustomWriteTarget(entityType, Map.class);
+
+		if (customTarget.isPresent()) {
+
+			sink.putAll(conversionService.convert(source, Map.class));
+			return;
+		}
+
+		if (typeHint != null) {
+
+			ElasticsearchPersistentEntity<?> entity = typeHint.getType().equals(entityType)
+					? mappingContext.getRequiredPersistentEntity(typeHint)
+					: mappingContext.getRequiredPersistentEntity(entityType);
+
+			writeEntity(entity, source, sink, null);
+			return;
+		}
+
+		// write Entity
+		ElasticsearchPersistentEntity<?> entity = mappingContext.getRequiredPersistentEntity(entityType);
+		writeEntity(entity, source, sink, null);
+	}
+
+	protected void writeEntity(ElasticsearchPersistentEntity<?> entity, Object source, Map<String, Object> sink,
+			@Nullable TypeInformation containingStructure) {
+
+		PersistentPropertyAccessor<?> accessor = entity.getPropertyAccessor(source);
+
+		if (requiresTypeHint(entity.getTypeInformation(), source.getClass(), containingStructure)) {
+			typeMapper.writeType(source.getClass(), sink);
+		}
+		writeProperties(entity, accessor, sink);
+	}
+
+	protected void writeProperties(ElasticsearchPersistentEntity<?> entity, PersistentPropertyAccessor<?> accessor,
+			Map<String, Object> sink) {
+
+		for (ElasticsearchPersistentProperty property : entity) {
+
+			if (!property.isWritable()) {
+				continue;
+			}
+
+			Object value = accessor.getProperty(property);
+
+			if (value == null) {
+				continue;
+			}
+
+			if (!conversions.isSimpleType(value.getClass())) {
+				writeProperty(property, value, sink);
+			} else {
+				sink.put(property.getFieldName(), getWriteSimpleValue(value));
+			}
+		}
+	}
+
+	protected void writeProperty(ElasticsearchPersistentProperty property, Object value, Map<String, Object> sink) {
+
+		Optional<Class<?>> customWriteTarget = conversions.getCustomWriteTarget(value.getClass());
+
+		if (customWriteTarget.isPresent()) {
+
+			Class<?> writeTarget = customWriteTarget.get();
+			sink.put(property.getFieldName(), conversionService.convert(value, writeTarget));
+			return;
+		}
+
+		TypeInformation<?> typeHint = property.getTypeInformation();
+		if (typeHint.equals(ClassTypeInformation.OBJECT)) {
+
+			if (value instanceof List) {
+				typeHint = ClassTypeInformation.LIST;
+			} else if (value instanceof Map) {
+				typeHint = ClassTypeInformation.MAP;
+			} else if (value instanceof Set) {
+				typeHint = ClassTypeInformation.SET;
+			} else if (value instanceof Collection) {
+				typeHint = ClassTypeInformation.COLLECTION;
+			}
+		}
+
+		sink.put(property.getFieldName(), getWriteComplexValue(property, typeHint, value));
+	}
+
+	protected Object getWriteSimpleValue(Object value) {
+
+		Optional<Class<?>> customWriteTarget = conversions.getCustomWriteTarget(value.getClass());
+
+		if (!customWriteTarget.isPresent()) {
+			return Enum.class.isAssignableFrom(value.getClass()) ? ((Enum<?>) value).name() : value;
+		}
+
+		return conversionService.convert(value, customWriteTarget.get());
+	}
+
+	protected Object getWriteComplexValue(ElasticsearchPersistentProperty property, TypeInformation<?> typeHint,
+			Object value) {
+
+		if (typeHint.isCollectionLike() || value instanceof Iterable) {
+			return writeCollectionValue(value, property, typeHint);
+		}
+		if (typeHint.isMap()) {
+			return writeMapValue((Map<String, Object>) value, property, typeHint);
+		}
+
+		if (property.isEntity() || !conversions.isSimpleType(value.getClass())) {
+			return writeEntity(value, property, typeHint);
+		}
+
+		return value;
+	}
+
+	private Object writeEntity(Object value, ElasticsearchPersistentProperty property, TypeInformation<?> typeHint) {
+		Map<String, Object> target = new LinkedHashMap<>();
+		writeEntity(mappingContext.getRequiredPersistentEntity(value.getClass()), value, target,
+				property.getTypeInformation());
+		return target;
+	}
+
+	private Object writeMapValue(Map<String, Object> value, ElasticsearchPersistentProperty property,
+			TypeInformation<?> typeHint) {
+
+		Map<Object, Object> target = new LinkedHashMap<>();
+		Streamable<Entry<String, Object>> mapSource = Streamable.of(value.entrySet());
+
+		if (!typeHint.getActualType().getType().equals(Object.class)
+				&& conversions.isSimpleType(typeHint.getMapValueType().getType())) {
+			mapSource.forEach(it -> target.put(it.getKey(), getWriteSimpleValue(it.getValue())));
+		} else {
+
+			mapSource.forEach(it -> {
+
+				Object converted = null;
+				if (it.getValue() != null) {
+
+					if (conversions.isSimpleType(it.getValue().getClass())) {
+						converted = getWriteSimpleValue(it.getValue());
+					} else {
+						converted = getWriteComplexValue(property, ClassTypeInformation.from(it.getValue().getClass()),
+								it.getValue());
+					}
+				}
+
+				target.put(it.getKey(), converted);
+
+			});
+		}
+
+		return target;
+	}
+
+	private Object writeCollectionValue(Object value, ElasticsearchPersistentProperty property,
+			TypeInformation<?> typeHint) {
+
+		Streamable collectionSource = value instanceof Iterable ? Streamable.of((Iterable) value)
+				: Streamable.of(ObjectUtils.toObjectArray(value));
+
+		List<Object> target = new ArrayList<>();
+		if (!typeHint.getActualType().getType().equals(Object.class)
+				&& conversions.isSimpleType(typeHint.getActualType().getType())) {
+
+			collectionSource.map(this::getWriteSimpleValue).forEach(target::add);
+		} else {
+
+			Streamable.of((Iterable) value).map(it -> {
+
+				if (it == null) {
+					return null;
+				}
+
+				if (conversions.isSimpleType(it.getClass())) {
+					return getWriteSimpleValue(it);
+				}
+
+				return getWriteComplexValue(property, ClassTypeInformation.from(it.getClass()), it);
+			}).forEach(target::add);
+
+		}
+		return target;
+	}
+
+	// --> GETTERS / SETTERS
+
+	@Override
+	public MappingContext<? extends ElasticsearchPersistentEntity<?>, ElasticsearchPersistentProperty> getMappingContext() {
+		return mappingContext;
+	}
+
+	@Override
+	public ConversionService getConversionService() {
+		return conversionService;
+	}
+
+	/**
+	 * Set the {@link CustomConversions} to be applied during the mapping process. <br />
+	 * Conversions are registered after {@link #afterPropertiesSet() bean initialization}.
+	 *
+	 * @param conversions must not be {@literal null}.
+	 */
+	public void setConversions(CustomConversions conversions) {
+		this.conversions = conversions;
+	}
+
+	/**
+	 * Set the {@link ElasticsearchTypeMapper} to use for reading / writing type hints.
+	 *
+	 * @param typeMapper must not be {@literal null}.
+	 */
+	public void setTypeMapper(ElasticsearchTypeMapper typeMapper) {
+		this.typeMapper = typeMapper;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet()
+	 */
+	@Override
+	public void afterPropertiesSet() {
+
+		DateFormatterRegistrar.addDateConverters(conversionService);
+		conversions.registerConvertersIn(conversionService);
+	}
+
+	// --> LEGACY
+
+	@Override
+	public String mapToString(Object source) throws IOException {
+
+		Map<String, Object> sink = new LinkedHashMap<>();
+		write(source, sink);
+
+		return new ObjectMapper().writeValueAsString(sink);
+	}
+
+	@Override
+	public <T> T mapToObject(String source, Class<T> clazz) throws IOException {
+		return read(clazz, new ObjectMapper().readerFor(HashMap.class).readValue(source));
+	}
+
+	// --> PRIVATE HELPERS
+
+	private boolean requiresTypeHint(TypeInformation<?> type, Class<?> actualType,
+			@Nullable TypeInformation<?> containingStructure) {
+
+		if (containingStructure != null) {
+
+			if (containingStructure.isCollectionLike()) {
+				if (containingStructure.getActualType().equals(type) && type.getType().equals(actualType)) {
+					return false;
+				}
+			}
+			if (containingStructure.isMap()) {
+				if (containingStructure.getMapValueType().equals(type) && type.getType().equals(actualType)) {
+					return false;
+				}
+			}
+			if (containingStructure.equals(type) && type.getType().equals(actualType)) {
+				return false;
+			}
+		}
+
+		return !conversions.isSimpleType(type.getType()) && !type.isCollectionLike()
+				&& !conversions.hasCustomWriteTarget(type.getType());
+	}
+
+	/**
+	 * Compute the type to use by checking the given entity against the store type;
+	 * 
+	 * @param entity
+	 * @param source
+	 * @return
+	 */
+	private ElasticsearchPersistentEntity<?> computeClosestEntity(ElasticsearchPersistentEntity<?> entity,
+			Map<String, Object> source) {
+
+		TypeInformation<?> typeToUse = typeMapper.readType(source);
+
+		if (typeToUse == null) {
+			return entity;
+		}
+
+		if (!entity.getTypeInformation().getType().isInterface() && !entity.getTypeInformation().isCollectionLike()
+				&& !entity.getTypeInformation().isMap()
+				&& !ClassUtils.isAssignableValue(entity.getType(), typeToUse.getType())) {
+			return entity;
+		}
+
+		return mappingContext.getRequiredPersistentEntity(typeToUse);
+	}
+
+	private ElasticsearchPersistentEntity<?> computeGenericValueTypeForRead(ElasticsearchPersistentProperty property,
+			Object value) {
+
+		return property.getTypeInformation().getActualType().equals(ClassTypeInformation.OBJECT)
+				? mappingContext.getRequiredPersistentEntity(value.getClass())
+				: mappingContext.getRequiredPersistentEntity(property.getTypeInformation().getActualType());
+	}
+
+	private Collection<?> createCollectionForValue(TypeInformation<?> collectionTypeInformation, int size) {
+
+		Class<?> collectionType = collectionTypeInformation.isSubTypeOf(Collection.class) //
+				? collectionTypeInformation.getType() //
+				: List.class;
+
+		TypeInformation<?> componentType = collectionTypeInformation.getComponentType() != null //
+				? collectionTypeInformation.getComponentType() //
+				: ClassTypeInformation.OBJECT;
+
+		return collectionTypeInformation.getType().isArray() //
+				? new ArrayList<>(size) //
+				: CollectionFactory.createCollection(collectionType, componentType.getType(), size);
+
+	}
+
+	// --> OHTER STUFF
+
+	class ElasticsearchPropertyValueProvider implements PropertyValueProvider<ElasticsearchPersistentProperty> {
+
+		final MapValueAccessor mapValueAccessor;
+
+		public ElasticsearchPropertyValueProvider(MapValueAccessor mapValueAccessor) {
+			this.mapValueAccessor = mapValueAccessor;
+		}
+
+		@Override
+		public <T> T getPropertyValue(ElasticsearchPersistentProperty property) {
+			return (T) readValue(mapValueAccessor.get(property), property, property.getTypeInformation());
+		}
+
+	}
+
+	static class MapValueAccessor {
+
+		final Map<String, Object> target;
+
+		MapValueAccessor(Map<String, Object> target) {
+			this.target = target;
+		}
+
+		public Object get(ElasticsearchPersistentProperty property) {
+
+			String fieldName = property.getFieldName();
+
+			if (!fieldName.contains(".")) {
+				return target.get(fieldName);
+			}
+
+			Iterator<String> parts = Arrays.asList(fieldName.split("\\.")).iterator();
+			Map<String, Object> source = target;
+			Object result = null;
+
+			while (source != null && parts.hasNext()) {
+
+				result = source.get(parts.next());
+
+				if (parts.hasNext()) {
+					source = getAsMap(result);
+				}
+			}
+
+			return result;
+		}
+
+		private Map<String, Object> getAsMap(Object result) {
+
+			if (result instanceof Map) {
+				return (Map) result;
+			}
+
+			throw new IllegalArgumentException(String.format("%s is no Map.", result));
+		}
+	}
+
+}

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

@@ -15,17 +15,25 @@
  */
 package org.springframework.data.elasticsearch.core;
 
-import static org.elasticsearch.client.Requests.refreshRequest;
-import static org.elasticsearch.index.query.QueryBuilders.moreLikeThisQuery;
-import static org.elasticsearch.index.query.QueryBuilders.wrapperQuery;
-import static org.springframework.data.elasticsearch.core.MappingBuilder.buildMapping;
+import static org.elasticsearch.client.Requests.*;
+import static org.elasticsearch.index.query.QueryBuilders.*;
+import static org.springframework.data.elasticsearch.core.MappingBuilder.*;
 import static org.springframework.util.CollectionUtils.isEmpty;
-import static org.springframework.util.StringUtils.hasText;
+import static org.springframework.util.StringUtils.*;
 
 import java.io.BufferedReader;
 import java.io.IOException;
 import java.io.InputStreamReader;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.NoSuchElementException;
+import java.util.Optional;
+import java.util.Set;
 
 import org.apache.http.util.EntityUtils;
 import org.elasticsearch.action.ActionFuture;
@@ -105,11 +113,11 @@ import org.springframework.data.elasticsearch.core.mapping.SimpleElasticsearchMa
 import org.springframework.data.elasticsearch.core.query.*;
 import org.springframework.data.util.CloseableIterator;
 import org.springframework.util.Assert;
+import org.springframework.util.StringUtils;
 
 import com.fasterxml.jackson.core.type.TypeReference;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
-import org.springframework.util.StringUtils;
 
 /**
  * ElasticsearchRestTemplate
@@ -132,6 +140,7 @@ import org.springframework.util.StringUtils;
  * @author Zetang Zeng
  * @author Peter Nowak
  * @author Ivan Greene
+ * @author Christoph Strobl
  */
 public class ElasticsearchRestTemplate
 		implements ElasticsearchOperations, EsClient<RestHighLevelClient>, ApplicationContextAware {
@@ -345,7 +354,8 @@ public class ElasticsearchRestTemplate
 		return queryForPage(queries, clazz, resultsMapper);
 	}
 
-	private <T> List<Page<T>> doMultiSearch(List<SearchQuery> queries, Class<T> clazz, MultiSearchRequest request, SearchResultMapper resultsMapper) {
+	private <T> List<Page<T>> doMultiSearch(List<SearchQuery> queries, Class<T> clazz, MultiSearchRequest request,
+			SearchResultMapper resultsMapper) {
 		MultiSearchResponse.Item[] items = getMultiSearchResult(request);
 		List<Page<T>> res = new ArrayList<>(queries.size());
 		int c = 0;
@@ -355,7 +365,8 @@ public class ElasticsearchRestTemplate
 		return res;
 	}
 
-	private List<Page<?>> doMultiSearch(List<SearchQuery> queries, List<Class<?>> classes, MultiSearchRequest request, SearchResultMapper resultsMapper) {
+	private List<Page<?>> doMultiSearch(List<SearchQuery> queries, List<Class<?>> classes, MultiSearchRequest request,
+			SearchResultMapper resultsMapper) {
 		MultiSearchResponse.Item[] items = getMultiSearchResult(request);
 		List<Page<?>> res = new ArrayList<>(queries.size());
 		int c = 0;
@@ -726,8 +737,7 @@ public class ElasticsearchRestTemplate
 	private UpdateRequest prepareUpdate(UpdateQuery query) {
 		String indexName = hasText(query.getIndexName()) ? query.getIndexName()
 				: getPersistentEntityFor(query.getClazz()).getIndexName();
-		String type = hasText(query.getType()) ? query.getType()
-				: getPersistentEntityFor(query.getClazz()).getIndexType();
+		String type = hasText(query.getType()) ? query.getType() : getPersistentEntityFor(query.getClazz()).getIndexType();
 		Assert.notNull(indexName, "No index defined for Query");
 		Assert.notNull(type, "No type define for Query");
 		Assert.notNull(query.getId(), "No Id define for Query");
@@ -880,6 +890,11 @@ public class ElasticsearchRestTemplate
 				}
 				return new AggregatedPageImpl<T>(Collections.EMPTY_LIST, response.getScrollId());
 			}
+
+			@Override
+			public <T> T mapSearchHit(SearchHit searchHit, Class<T> type) {
+				return null;
+			}
 		};
 
 		Page<String> scrolledResult = startScroll(scrollTimeInMillis, searchQuery, String.class, onlyIdResultMapper);
@@ -1132,7 +1147,7 @@ public class ElasticsearchRestTemplate
 			if (highlightBuilder == null) {
 				highlightBuilder = new HighlightBuilder();
 			}
-			if(searchQuery.getHighlightFields() != null) {
+			if (searchQuery.getHighlightFields() != null) {
 				for (HighlightBuilder.Field highlightField : searchQuery.getHighlightFields()) {
 					highlightBuilder.field(highlightField);
 				}
@@ -1326,7 +1341,8 @@ public class ElasticsearchRestTemplate
 			String indexName = StringUtils.isEmpty(query.getIndexName())
 					? retrieveIndexNameFromPersistentEntity(query.getObject().getClass())[0]
 					: query.getIndexName();
-			String type = StringUtils.isEmpty(query.getType()) ? retrieveTypeFromPersistentEntity(query.getObject().getClass())[0]
+			String type = StringUtils.isEmpty(query.getType())
+					? retrieveTypeFromPersistentEntity(query.getObject().getClass())[0]
 					: query.getType();
 
 			IndexRequest indexRequest = null;

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

@@ -99,21 +99,7 @@ import org.springframework.data.elasticsearch.core.facet.FacetRequest;
 import org.springframework.data.elasticsearch.core.mapping.ElasticsearchPersistentEntity;
 import org.springframework.data.elasticsearch.core.mapping.ElasticsearchPersistentProperty;
 import org.springframework.data.elasticsearch.core.mapping.SimpleElasticsearchMappingContext;
-import org.springframework.data.elasticsearch.core.query.AliasQuery;
-import org.springframework.data.elasticsearch.core.query.CriteriaQuery;
-import org.springframework.data.elasticsearch.core.query.DeleteQuery;
-import org.springframework.data.elasticsearch.core.query.FetchSourceFilter;
-import org.springframework.data.elasticsearch.core.query.GetQuery;
-import org.springframework.data.elasticsearch.core.query.IndexBoost;
-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.elasticsearch.core.query.ScriptField;
-import org.springframework.data.elasticsearch.core.query.SearchQuery;
-import org.springframework.data.elasticsearch.core.query.SourceFilter;
-import org.springframework.data.elasticsearch.core.query.StringQuery;
-import org.springframework.data.elasticsearch.core.query.UpdateQuery;
+import org.springframework.data.elasticsearch.core.query.*;
 import org.springframework.data.util.CloseableIterator;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
@@ -138,6 +124,7 @@ import org.springframework.util.StringUtils;
  * @author Jean-Baptiste Nizet
  * @author Zetang Zeng
  * @author Ivan Greene
+ * @author Christoph Strobl
  */
 public class ElasticsearchTemplate implements ElasticsearchOperations, EsClient<Client>, ApplicationContextAware {
 
@@ -223,8 +210,8 @@ public class ElasticsearchTemplate implements ElasticsearchOperations, EsClient<
 
 			ElasticsearchPersistentProperty property = persistentEntity.getRequiredIdProperty();
 
-			xContentBuilder = buildMapping(clazz, persistentEntity.getIndexType(),
-					property.getFieldName(), persistentEntity.getParentType());
+			xContentBuilder = buildMapping(clazz, persistentEntity.getIndexType(), property.getFieldName(),
+					persistentEntity.getParentType());
 		} catch (Exception e) {
 			throw new ElasticsearchException("Failed to build mapping for " + clazz.getSimpleName(), e);
 		}
@@ -332,7 +319,8 @@ public class ElasticsearchTemplate implements ElasticsearchOperations, EsClient<
 		return doMultiSearch(queries, clazz, request, mapper);
 	}
 
-	private <T> List<Page<T>> doMultiSearch(List<SearchQuery> queries, Class<T> clazz, MultiSearchRequest request, SearchResultMapper resultsMapper) {
+	private <T> List<Page<T>> doMultiSearch(List<SearchQuery> queries, Class<T> clazz, MultiSearchRequest request,
+			SearchResultMapper resultsMapper) {
 		MultiSearchResponse.Item[] items = getMultiSearchResult(request);
 		List<Page<T>> res = new ArrayList<>(queries.size());
 		int c = 0;
@@ -342,7 +330,8 @@ public class ElasticsearchTemplate implements ElasticsearchOperations, EsClient<
 		return res;
 	}
 
-	private List<Page<?>> doMultiSearch(List<SearchQuery> queries, List<Class<?>> classes, MultiSearchRequest request, SearchResultMapper resultsMapper) {
+	private List<Page<?>> doMultiSearch(List<SearchQuery> queries, List<Class<?>> classes, MultiSearchRequest request,
+			SearchResultMapper resultsMapper) {
 		MultiSearchResponse.Item[] items = getMultiSearchResult(request);
 		List<Page<?>> res = new ArrayList<>(queries.size());
 		int c = 0;
@@ -446,7 +435,8 @@ public class ElasticsearchTemplate implements ElasticsearchOperations, EsClient<
 	@Override
 	public <T> CloseableIterator<T> stream(CriteriaQuery query, Class<T> clazz) {
 		final long scrollTimeInMillis = TimeValue.timeValueMinutes(1).millis();
-		return doStream(scrollTimeInMillis, (ScrolledPage<T>) startScroll(scrollTimeInMillis, query, clazz), clazz, resultsMapper);
+		return doStream(scrollTimeInMillis, (ScrolledPage<T>) startScroll(scrollTimeInMillis, query, clazz), clazz,
+				resultsMapper);
 	}
 
 	@Override
@@ -457,10 +447,12 @@ public class ElasticsearchTemplate implements ElasticsearchOperations, EsClient<
 	@Override
 	public <T> CloseableIterator<T> stream(SearchQuery query, final Class<T> clazz, final SearchResultMapper mapper) {
 		final long scrollTimeInMillis = TimeValue.timeValueMinutes(1).millis();
-		return doStream(scrollTimeInMillis, (ScrolledPage<T>) startScroll(scrollTimeInMillis, query, clazz, mapper), clazz, mapper);
+		return doStream(scrollTimeInMillis, (ScrolledPage<T>) startScroll(scrollTimeInMillis, query, clazz, mapper), clazz,
+				mapper);
 	}
 
-	private <T> CloseableIterator<T> doStream(final long scrollTimeInMillis, final ScrolledPage<T> page, final Class<T> clazz, final SearchResultMapper mapper) {
+	private <T> CloseableIterator<T> doStream(final long scrollTimeInMillis, final ScrolledPage<T> page,
+			final Class<T> clazz, final SearchResultMapper mapper) {
 		return new CloseableIterator<T>() {
 
 			/** As we couldn't retrieve single result with scroll, store current hits. */
@@ -776,6 +768,11 @@ public class ElasticsearchTemplate implements ElasticsearchOperations, EsClient<
 				}
 				return new AggregatedPageImpl<T>(Collections.EMPTY_LIST, response.getScrollId());
 			}
+
+			@Override
+			public <T> T mapSearchHit(SearchHit searchHit, Class<T> type) {
+				return null;
+			}
 		};
 
 		Page<String> scrolledResult = startScroll(scrollTimeInMillis, searchQuery, String.class, onlyIdResultMapper);
@@ -784,8 +781,9 @@ public class ElasticsearchTemplate implements ElasticsearchOperations, EsClient<
 
 		do {
 			ids.addAll(scrolledResult.getContent());
-			scrolledResult = continueScroll(((ScrolledPage<T>)scrolledResult).getScrollId(), scrollTimeInMillis, String.class, onlyIdResultMapper);
-		} while(scrolledResult.getContent().size() != 0);
+			scrolledResult = continueScroll(((ScrolledPage<T>) scrolledResult).getScrollId(), scrollTimeInMillis,
+					String.class, onlyIdResultMapper);
+		} while (scrolledResult.getContent().size() != 0);
 
 		for (String id : ids) {
 			bulkRequestBuilder.add(client.prepareDelete(indexName, typeName, id));
@@ -821,12 +819,10 @@ public class ElasticsearchTemplate implements ElasticsearchOperations, EsClient<
 
 	private SearchRequestBuilder prepareScroll(Query query, long scrollTimeInMillis) {
 		SearchRequestBuilder requestBuilder = client.prepareSearch(toArray(query.getIndices()))
-				.setTypes(toArray(query.getTypes()))
-				.setScroll(TimeValue.timeValueMillis(scrollTimeInMillis))
-				.setFrom(0)
+				.setTypes(toArray(query.getTypes())).setScroll(TimeValue.timeValueMillis(scrollTimeInMillis)).setFrom(0)
 				.setVersion(true);
 
-		if(query.getPageable().isPaged()){
+		if (query.getPageable().isPaged()) {
 			requestBuilder.setSize(query.getPageable().getPageSize());
 		}
 
@@ -880,25 +876,28 @@ public class ElasticsearchTemplate implements ElasticsearchOperations, EsClient<
 		return resultsMapper.mapResults(response, clazz, null);
 	}
 
-	public <T> Page<T> startScroll(long scrollTimeInMillis, SearchQuery searchQuery, Class<T> clazz, SearchResultMapper mapper) {
+	public <T> Page<T> startScroll(long scrollTimeInMillis, SearchQuery searchQuery, Class<T> clazz,
+			SearchResultMapper mapper) {
 		SearchResponse response = doScroll(prepareScroll(searchQuery, scrollTimeInMillis, clazz), searchQuery);
 		return mapper.mapResults(response, clazz, null);
 	}
 
-	public <T> Page<T> startScroll(long scrollTimeInMillis, CriteriaQuery criteriaQuery, Class<T> clazz, SearchResultMapper mapper) {
+	public <T> Page<T> startScroll(long scrollTimeInMillis, CriteriaQuery criteriaQuery, Class<T> clazz,
+			SearchResultMapper mapper) {
 		SearchResponse response = doScroll(prepareScroll(criteriaQuery, scrollTimeInMillis, clazz), criteriaQuery);
 		return mapper.mapResults(response, clazz, null);
 	}
 
 	public <T> Page<T> continueScroll(@Nullable String scrollId, long scrollTimeInMillis, Class<T> clazz) {
-		SearchResponse response = getSearchResponse(client.prepareSearchScroll(scrollId)
-				.setScroll(TimeValue.timeValueMillis(scrollTimeInMillis)).execute());
+		SearchResponse response = getSearchResponse(
+				client.prepareSearchScroll(scrollId).setScroll(TimeValue.timeValueMillis(scrollTimeInMillis)).execute());
 		return resultsMapper.mapResults(response, clazz, Pageable.unpaged());
 	}
 
-	public <T> Page<T> continueScroll(@Nullable String scrollId, long scrollTimeInMillis, Class<T> clazz, SearchResultMapper mapper) {
-		SearchResponse response = getSearchResponse(client.prepareSearchScroll(scrollId)
-				.setScroll(TimeValue.timeValueMillis(scrollTimeInMillis)).execute());
+	public <T> Page<T> continueScroll(@Nullable String scrollId, long scrollTimeInMillis, Class<T> clazz,
+			SearchResultMapper mapper) {
+		SearchResponse response = getSearchResponse(
+				client.prepareSearchScroll(scrollId).setScroll(TimeValue.timeValueMillis(scrollTimeInMillis)).execute());
 		return mapper.mapResults(response, clazz, Pageable.unpaged());
 	}
 
@@ -911,15 +910,16 @@ public class ElasticsearchTemplate implements ElasticsearchOperations, EsClient<
 	public <T> Page<T> moreLikeThis(MoreLikeThisQuery query, Class<T> clazz) {
 
 		ElasticsearchPersistentEntity persistentEntity = getPersistentEntityFor(clazz);
-		String indexName = !StringUtils.isEmpty(query.getIndexName()) ? query.getIndexName() : persistentEntity.getIndexName();
+		String indexName = !StringUtils.isEmpty(query.getIndexName()) ? query.getIndexName()
+				: persistentEntity.getIndexName();
 		String type = !StringUtils.isEmpty(query.getType()) ? query.getType() : persistentEntity.getIndexType();
 
 		Assert.notNull(indexName, "No 'indexName' defined for MoreLikeThisQuery");
 		Assert.notNull(type, "No 'type' defined for MoreLikeThisQuery");
 		Assert.notNull(query.getId(), "No document id defined for MoreLikeThisQuery");
 
-
-		MoreLikeThisQueryBuilder moreLikeThisQueryBuilder = moreLikeThisQuery(toArray(new MoreLikeThisQueryBuilder.Item(indexName, type, query.getId())));
+		MoreLikeThisQueryBuilder moreLikeThisQueryBuilder = moreLikeThisQuery(
+				toArray(new MoreLikeThisQueryBuilder.Item(indexName, type, query.getId())));
 
 		if (query.getMinTermFreq() != null) {
 			moreLikeThisQueryBuilder.minTermFreq(query.getMinTermFreq());
@@ -978,7 +978,7 @@ public class ElasticsearchTemplate implements ElasticsearchOperations, EsClient<
 			if (highlightBuilder == null) {
 				highlightBuilder = new HighlightBuilder();
 			}
-			if(searchQuery.getHighlightFields() != null) {
+			if (searchQuery.getHighlightFields() != null) {
 				for (HighlightBuilder.Field highlightField : searchQuery.getHighlightFields()) {
 					highlightBuilder.field(highlightField);
 				}
@@ -1075,9 +1075,9 @@ public class ElasticsearchTemplate implements ElasticsearchOperations, EsClient<
 	@Override
 	public Map getSetting(String indexName) {
 		Assert.notNull(indexName, "No index defined for getSettings");
-		Settings settings = client.admin().indices().getSettings(new GetSettingsRequest()).actionGet()
-				.getIndexToSettings().get(indexName);
-		return settings.keySet().stream().collect(Collectors.toMap((key)->key, (key)->settings.get(key)));
+		Settings settings = client.admin().indices().getSettings(new GetSettingsRequest()).actionGet().getIndexToSettings()
+				.get(indexName);
+		return settings.keySet().stream().collect(Collectors.toMap((key) -> key, (key) -> settings.get(key)));
 	}
 
 	private <T> SearchRequestBuilder prepareSearch(Query query, Class<T> clazz) {
@@ -1091,9 +1091,7 @@ public class ElasticsearchTemplate implements ElasticsearchOperations, EsClient<
 
 		int startRecord = 0;
 		SearchRequestBuilder searchRequestBuilder = client.prepareSearch(toArray(query.getIndices()))
-				.setSearchType(query.getSearchType())
-				.setTypes(toArray(query.getTypes()))
-				.setVersion(true)
+				.setSearchType(query.getSearchType()).setTypes(toArray(query.getTypes())).setVersion(true)
 				.setTrackScores(query.getTrackScores());
 
 		if (query.getSourceFilter() != null) {
@@ -1152,7 +1150,8 @@ public class ElasticsearchTemplate implements ElasticsearchOperations, EsClient<
 			String indexName = StringUtils.isEmpty(query.getIndexName())
 					? retrieveIndexNameFromPersistentEntity(query.getObject().getClass())[0]
 					: query.getIndexName();
-			String type = StringUtils.isEmpty(query.getType()) ? retrieveTypeFromPersistentEntity(query.getObject().getClass())[0]
+			String type = StringUtils.isEmpty(query.getType())
+					? retrieveTypeFromPersistentEntity(query.getObject().getClass())[0]
 					: query.getType();
 
 			IndexRequestBuilder indexRequestBuilder = null;

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

@@ -16,6 +16,9 @@
 package org.springframework.data.elasticsearch.core;
 
 import java.io.IOException;
+import java.util.Map;
+
+import org.springframework.lang.Nullable;
 
 /**
  * DocumentMapper interface, it will allow to customize how we mapping object to json
@@ -23,10 +26,32 @@ import java.io.IOException;
  * @author Artur Konczak
  * @author Rizwan Idrees
  * @author Mohsin Husen
+ * @author Christoph Strobl
  */
 public interface EntityMapper {
 
-	public String mapToString(Object object) throws IOException;
+	String mapToString(Object object) throws IOException;
 
-	public <T> T mapToObject(String source, Class<T> clazz) throws IOException;
+	<T> T mapToObject(String source, Class<T> clazz) throws IOException;
+
+	/**
+	 * Map the given {@literal source} to {@link Map}.
+	 *
+	 * @param source must not be {@literal null}.
+	 * @return never {@literal null}
+	 * @since 3.2
+	 */
+	Map<String, Object> mapObject(Object source);
+
+	/**
+	 * Map the given {@link Map} into an instance of the {@literal targetType}.
+	 *
+	 * @param source must not be {@literal null}.
+	 * @param targetType must not be {@literal null}.
+	 * @param <T>
+	 * @return can be {@literal null}.
+	 * @since 3.2
+	 */
+	@Nullable
+	<T> T readObject(Map<String, Object> source, Class<T> targetType);
 }

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

@@ -16,12 +16,27 @@
 package org.springframework.data.elasticsearch.core;
 
 import org.elasticsearch.action.get.GetResponse;
+import org.elasticsearch.index.get.GetResult;
+import org.springframework.lang.Nullable;
 
 /**
  * @author Artur Konczak
  * @author Mohsin Husen
+ * @author Christoph Strobl
  */
 public interface GetResultMapper {
 
 	<T> T mapResult(GetResponse response, Class<T> clazz);
+
+	/**
+	 * Map a single {@link GetResult} to the given {@link Class type}.
+	 *
+	 * @param getResult must not be {@literal null}.
+	 * @param type must not be {@literal null}.
+	 * @param <T>
+	 * @return can be {@literal null}.
+	 * @since 3.2
+	 */
+	@Nullable
+	<T> T mapGetResult(GetResult getResult, Class<T> type);
 }

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

@@ -98,11 +98,17 @@ public class ReactiveElasticsearchTemplate implements ReactiveElasticsearchOpera
 	}
 
 	public ReactiveElasticsearchTemplate(ReactiveElasticsearchClient client, ElasticsearchConverter converter) {
+		this(client, converter, new DefaultResultMapper(converter.getMappingContext()));
+	}
+
+	public ReactiveElasticsearchTemplate(ReactiveElasticsearchClient client, ElasticsearchConverter converter,
+			ResultsMapper resultsMapper) {
 
 		this.client = client;
 		this.converter = converter;
 		this.mappingContext = converter.getMappingContext();
-		this.resultMapper = new DefaultResultMapper(converter.getMappingContext());
+
+		this.resultMapper = resultsMapper;
 		this.exceptionTranslator = new ElasticsearchExceptionTranslator();
 		this.operations = new EntityOperations(this.mappingContext);
 	}
@@ -184,7 +190,7 @@ public class ReactiveElasticsearchTemplate implements ReactiveElasticsearchOpera
 		Assert.notNull(id, "Id must not be null!");
 
 		return doFindById(id, getPersistentEntity(entityType), index, type)
-				.map(it -> resultMapper.mapEntity(it, entityType));
+				.map(it -> resultMapper.mapGetResult(it, entityType));
 	}
 
 	private Mono<GetResult> doFindById(String id, ElasticsearchPersistentEntity<?> entity, @Nullable String index,
@@ -201,6 +207,7 @@ public class ReactiveElasticsearchTemplate implements ReactiveElasticsearchOpera
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.elasticsearch.core.ReactiveElasticsearchOperations#exists(String, Class, String, String)
+	 * @see org.springframework.data.elasticsearch.core.ReactiveElasticsearchOperations#exists(String, Class, String, String)
 	 */
 	@Override
 	public Mono<Boolean> exists(String id, Class<?> entityType, String index, String type) {
@@ -230,7 +237,7 @@ public class ReactiveElasticsearchTemplate implements ReactiveElasticsearchOpera
 			Class<T> resultType) {
 
 		return doFind(query, getPersistentEntity(entityType), index, type)
-				.map(it -> resultMapper.mapEntity(it, resultType));
+				.map(it -> resultMapper.mapSearchHit(it, resultType));
 	}
 
 	private Flux<SearchHit> doFind(Query query, ElasticsearchPersistentEntity<?> entity, @Nullable String index,

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

@@ -16,11 +16,15 @@
 package org.springframework.data.elasticsearch.core;
 
 import java.io.IOException;
+import java.util.Map;
 
 import org.elasticsearch.index.get.GetResult;
 import org.elasticsearch.search.SearchHit;
 import org.springframework.data.elasticsearch.ElasticsearchException;
+import org.springframework.data.projection.ProjectionFactory;
+import org.springframework.data.projection.SpelAwareProxyProjectionFactory;
 import org.springframework.lang.Nullable;
+import org.springframework.util.ClassUtils;
 import org.springframework.util.StringUtils;
 
 /**
@@ -35,6 +39,16 @@ public interface ResultsMapper extends SearchResultMapper, GetResultMapper, Mult
 
 	EntityMapper getEntityMapper();
 
+	/**
+	 * Get the configured {@link ProjectionFactory}. <br />
+	 * <strong>NOTE</strong> Should be overwritten in implementation to make use of the type cache.
+	 *
+	 * @since 3.2
+	 */
+	default ProjectionFactory getProjectionFactory() {
+		return new SpelAwareProxyProjectionFactory();
+	}
+
 	@Nullable
 	default <T> T mapEntity(String source, Class<T> clazz) {
 
@@ -58,19 +72,28 @@ public interface ResultsMapper extends SearchResultMapper, GetResultMapper, Mult
 	 * @since 3.2
 	 */
 	@Nullable
-	default <T> T mapEntity(GetResult getResult, Class<T> type) {
+	default <T> T mapGetResult(GetResult getResult, Class<T> type) {
 
 		if (getResult.isSourceEmpty()) {
 			return null;
 		}
 
-		String sourceString = getResult.sourceAsString();
-
-		if (sourceString.startsWith("{\"id\":null,")) {
-			sourceString = sourceString.replaceFirst("\"id\":null", "\"id\":\"" + getResult.getId() + "\"");
+		Map<String, Object> source = getResult.getSource();
+		if (!source.containsKey("id") || source.get("id") == null) {
+			source.put("id", getResult.getId());
 		}
 
-		return mapEntity(sourceString, type);
+		Object mappedResult = getEntityMapper().readObject(source, type);
+
+		if (mappedResult == null) {
+			return (T) mappedResult;
+		}
+
+		if (type.isInterface() || !ClassUtils.isAssignableValue(type, mappedResult)) {
+			return getProjectionFactory().createProjection(type, mappedResult);
+		}
+
+		return (T) mappedResult;
 	}
 
 	/**
@@ -83,18 +106,27 @@ public interface ResultsMapper extends SearchResultMapper, GetResultMapper, Mult
 	 * @since 3.2
 	 */
 	@Nullable
-	default <T> T mapEntity(SearchHit searchHit, Class<T> type) {
+	default <T> T mapSearchHit(SearchHit searchHit, Class<T> type) {
 
 		if (!searchHit.hasSource()) {
 			return null;
 		}
 
-		String sourceString = searchHit.getSourceAsString();
-
-		if (sourceString.startsWith("{\"id\":null,")) {
-			sourceString = sourceString.replaceFirst("\"id\":null", "\"id\":\"" + searchHit.getId() + "\"");
+		Map<String, Object> source = searchHit.getSourceAsMap();
+		if (!source.containsKey("id") || source.get("id") == null) {
+			source.put("id", searchHit.getId());
 		}
 
-		return mapEntity(sourceString, type);
+		Object mappedResult = getEntityMapper().readObject(source, type);
+
+		if (mappedResult == null) {
+			return (T) mappedResult;
+		}
+
+		if (type.isInterface()) {
+			return getProjectionFactory().createProjection(type, mappedResult);
+		}
+
+		return (T) mappedResult;
 	}
 }

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

@@ -16,14 +16,29 @@
 package org.springframework.data.elasticsearch.core;
 
 import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.search.SearchHit;
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.elasticsearch.core.aggregation.AggregatedPage;
+import org.springframework.lang.Nullable;
 
 /**
  * @author Artur Konczak
  * @author Petar Tahchiev
+ * @author Christoph Strobl
  */
 public interface SearchResultMapper {
 
 	<T> AggregatedPage<T> mapResults(SearchResponse response, Class<T> clazz, Pageable pageable);
+
+	/**
+	 * Map a single {@link SearchHit} to the given {@link Class type}.
+	 *
+	 * @param searchHit must not be {@literal null}.
+	 * @param type must not be {@literal null}.
+	 * @param <T>
+	 * @return can be {@literal null}.
+	 * @since 3.2
+	 */
+	@Nullable
+	<T> T mapSearchHit(SearchHit searchHit, Class<T> type);
 }

src/main/java/org/springframework/data/elasticsearch/core/convert/ElasticsearchCustomConversions.java

@@ -15,22 +15,94 @@
  */
 package org.springframework.data.elasticsearch.core.convert;
 
+import java.math.BigDecimal;
+import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.UUID;
 
+import org.springframework.core.convert.converter.Converter;
 import org.springframework.data.convert.CustomConversions;
+import org.springframework.data.convert.ReadingConverter;
+import org.springframework.data.convert.WritingConverter;
+import org.springframework.data.elasticsearch.core.mapping.ElasticsearchSimpleTypes;
+import org.springframework.util.NumberUtils;
 
 /**
+ * Elasticsearch specific {@link CustomConversions}.
+ *
  * @author Christoph Strobl
  * @since 3.2
  */
 public class ElasticsearchCustomConversions extends CustomConversions {
 
+	private static final StoreConversions STORE_CONVERSIONS;
+	private static final List<Object> STORE_CONVERTERS;
+
+	static {
+
+		List<Object> converters = new ArrayList<>();
+		converters.addAll(GeoConverters.getConvertersToRegister());
+		converters.add(StringToUUIDConverter.INSTANCE);
+		converters.add(UUIDToStringConverter.INSTANCE);
+		converters.add(BigDecimalToDoubleConverter.INSTANCE);
+		converters.add(DoubleToBigDecimalConverter.INSTANCE);
+
+		STORE_CONVERTERS = Collections.unmodifiableList(converters);
+		STORE_CONVERSIONS = StoreConversions.of(ElasticsearchSimpleTypes.HOLDER, STORE_CONVERTERS);
+	}
+
 	/**
 	 * Creates a new {@link CustomConversions} instance registering the given converters.
 	 *
 	 * @param converters must not be {@literal null}.
 	 */
 	public ElasticsearchCustomConversions(Collection<?> converters) {
-		super(StoreConversions.NONE, converters);
+		super(STORE_CONVERSIONS, converters);
+	}
+
+	@ReadingConverter
+	enum StringToUUIDConverter implements Converter<String, UUID> {
+
+		INSTANCE;
+
+		@Override
+		public UUID convert(String source) {
+			return UUID.fromString(source);
+		}
+	}
+
+	@WritingConverter
+	enum UUIDToStringConverter implements Converter<UUID, String> {
+
+		INSTANCE;
+
+		@Override
+		public String convert(UUID source) {
+			return source.toString();
+		}
+	}
+
+	@ReadingConverter
+	enum DoubleToBigDecimalConverter implements Converter<Double, BigDecimal> {
+
+		INSTANCE;
+
+		@Override
+		public BigDecimal convert(Double source) {
+			return NumberUtils.convertNumberToTargetClass(source, BigDecimal.class);
+		}
+	}
+
+	@WritingConverter
+	enum BigDecimalToDoubleConverter implements Converter<BigDecimal, Double> {
+
+		INSTANCE;
+
+		@Override
+		public Double convert(BigDecimal source) {
+			return NumberUtils.convertNumberToTargetClass(source, Double.class);
+		}
 	}
 }

src/main/java/org/springframework/data/elasticsearch/core/convert/ElasticsearchDefaultTypeMapper.java

@@ -0,0 +1,47 @@
+/*
+ * Copyright 2019 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
+ *
+ *      http://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.convert;
+
+import java.util.List;
+import java.util.Map;
+
+import org.springframework.data.convert.DefaultTypeMapper;
+import org.springframework.data.convert.TypeAliasAccessor;
+import org.springframework.data.mapping.context.MappingContext;
+import org.springframework.lang.Nullable;
+
+/**
+ * Elasticsearch specific {@link org.springframework.data.convert.TypeMapper} implementation.
+ *
+ * @author Christoph Strobl
+ * @since 3.2
+ */
+class ElasticsearchDefaultTypeMapper extends DefaultTypeMapper<Map<String, Object>> implements ElasticsearchTypeMapper {
+
+	private final @Nullable String typeKey;
+
+	ElasticsearchDefaultTypeMapper(@Nullable String typeKey, TypeAliasAccessor accessor,
+			@Nullable MappingContext mappingContext, List additionalMappers) {
+
+		super(accessor, mappingContext, additionalMappers);
+		this.typeKey = typeKey;
+	}
+
+	@Override
+	public boolean isTypeKey(String key) {
+		return typeKey != null && typeKey.equals(key);
+	}
+}

src/main/java/org/springframework/data/elasticsearch/core/convert/ElasticsearchTypeMapper.java

@@ -0,0 +1,92 @@
+/*
+ * Copyright 2019 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
+ *
+ *      http://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.convert;
+
+import java.util.Collections;
+import java.util.Map;
+
+import org.springframework.data.convert.SimpleTypeInformationMapper;
+import org.springframework.data.convert.TypeAliasAccessor;
+import org.springframework.data.convert.TypeMapper;
+import org.springframework.data.elasticsearch.core.mapping.ElasticsearchPersistentEntity;
+import org.springframework.data.elasticsearch.core.mapping.ElasticsearchPersistentProperty;
+import org.springframework.data.mapping.Alias;
+import org.springframework.data.mapping.context.MappingContext;
+import org.springframework.lang.Nullable;
+
+/**
+ * Elasticsearch specific {@link TypeMapper} definition.
+ *
+ * @author Christoph Strobl
+ * @since 3.2
+ */
+public interface ElasticsearchTypeMapper extends TypeMapper<Map<String, Object>> {
+
+	String DEFAULT_TYPE_KEY = "_class";
+
+	/**
+	 * Returns whether the given key is the type key.
+	 *
+	 * @return {@literal true} if given {@literal key} is used as type hint key.
+	 */
+	boolean isTypeKey(String key);
+
+	default boolean containsTypeInformation(Map<String, Object> source) {
+		return readType(source) != null;
+	}
+
+	static ElasticsearchTypeMapper defaultTypeMapper(
+			MappingContext<? extends ElasticsearchPersistentEntity<?>, ElasticsearchPersistentProperty> mappingContext) {
+		return new ElasticsearchDefaultTypeMapper(DEFAULT_TYPE_KEY, new MapTypeAliasAccessor(DEFAULT_TYPE_KEY),
+				mappingContext, Collections.singletonList(new SimpleTypeInformationMapper()));
+	}
+
+	/**
+	 * {@link TypeAliasAccessor} to store aliases in a {@link Map}.
+	 *
+	 * @author Christoph Strobl
+	 */
+	final class MapTypeAliasAccessor implements TypeAliasAccessor<Map<String, Object>> {
+
+		private final @Nullable String typeKey;
+
+		public MapTypeAliasAccessor(@Nullable String typeKey) {
+			this.typeKey = typeKey;
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.convert.TypeAliasAccessor#readAliasFrom(java.lang.Object)
+		 */
+		public Alias readAliasFrom(Map<String, Object> source) {
+			return Alias.ofNullable(source.get(typeKey));
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.convert.TypeAliasAccessor#writeTypeTo(java.lang.Object, java.lang.Object)
+		 */
+		public void writeTypeTo(Map<String, Object> sink, Object alias) {
+
+			if (typeKey == null) {
+				return;
+			}
+
+			sink.put(typeKey, alias);
+		}
+	}
+
+}

src/main/java/org/springframework/data/elasticsearch/core/convert/GeoConverters.java

@@ -0,0 +1,100 @@
+/*
+ * Copyright 2019 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
+ *
+ *      http://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.convert;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import org.springframework.core.convert.converter.Converter;
+import org.springframework.data.convert.ReadingConverter;
+import org.springframework.data.convert.WritingConverter;
+import org.springframework.data.elasticsearch.core.geo.GeoPoint;
+import org.springframework.data.geo.Point;
+import org.springframework.util.NumberUtils;
+
+/**
+ * Set of {@link Converter converters} specific to Elasticsearch Geo types.
+ *
+ * @author Christoph Strobl
+ * @since 3.2
+ */
+class GeoConverters {
+
+	static Collection<? extends Object> getConvertersToRegister() {
+
+		return Arrays.asList(PointToMapConverter.INSTANCE, MapToPointConverter.INSTANCE, GeoPointToMapConverter.INSTANCE,
+				MapToGeoPointConverter.INSTANCE);
+	}
+
+	@WritingConverter
+	enum PointToMapConverter implements Converter<Point, Map> {
+
+		INSTANCE;
+
+		@Override
+		public Map convert(Point source) {
+
+			Map<String, Object> target = new LinkedHashMap<>();
+			target.put("lat", source.getX());
+			target.put("lon", source.getY());
+			return target;
+		}
+	}
+
+	@WritingConverter
+	enum GeoPointToMapConverter implements Converter<GeoPoint, Map> {
+
+		INSTANCE;
+
+		@Override
+		public Map convert(GeoPoint source) {
+			Map<String, Object> target = new LinkedHashMap<>();
+			target.put("lat", source.getLat());
+			target.put("lon", source.getLon());
+			return target;
+		}
+	}
+
+	@ReadingConverter
+	enum MapToPointConverter implements Converter<Map, Point> {
+
+		INSTANCE;
+
+		@Override
+		public Point convert(Map source) {
+			Double x = NumberUtils.convertNumberToTargetClass((Number) source.get("lat"), Double.class);
+			Double y = NumberUtils.convertNumberToTargetClass((Number) source.get("lon"), Double.class);
+
+			return new Point(x, y);
+		}
+	}
+
+	@ReadingConverter
+	enum MapToGeoPointConverter implements Converter<Map, GeoPoint> {
+
+		INSTANCE;
+
+		@Override
+		public GeoPoint convert(Map source) {
+			Double x = NumberUtils.convertNumberToTargetClass((Number) source.get("lat"), Double.class);
+			Double y = NumberUtils.convertNumberToTargetClass((Number) source.get("lon"), Double.class);
+
+			return new GeoPoint(x, y);
+		}
+	}
+}

src/main/java/org/springframework/data/elasticsearch/core/mapping/ElasticsearchSimpleTypes.java

@@ -0,0 +1,46 @@
+/*
+ * Copyright 2019 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
+ *
+ *      http://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.mapping;
+
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Set;
+
+import org.springframework.data.mapping.model.SimpleTypeHolder;
+
+/**
+ * @author Christoph Strobl
+ * @since 3.2
+ */
+public class ElasticsearchSimpleTypes {
+
+	static final Set<Class<?>> AUTOGENERATED_ID_TYPES;
+
+	static {
+		Set<Class<?>> classes = new HashSet<>();
+		classes.add(String.class);
+		AUTOGENERATED_ID_TYPES = Collections.unmodifiableSet(classes);
+
+		Set<Class<?>> simpleTypes = new HashSet<>();
+		ELASTICSEARCH_SIMPLE_TYPES = Collections.unmodifiableSet(simpleTypes);
+	}
+
+	private static final Set<Class<?>> ELASTICSEARCH_SIMPLE_TYPES;
+	public static final SimpleTypeHolder HOLDER = new SimpleTypeHolder(ELASTICSEARCH_SIMPLE_TYPES, true);
+
+	private ElasticsearchSimpleTypes() {}
+
+}

src/test/java/org/springframework/data/elasticsearch/config/ElasticsearchConfigurationSupportUnitTests.java

@@ -25,12 +25,15 @@ import org.apache.commons.lang.ClassUtils;
 import org.elasticsearch.client.RestHighLevelClient;
 import org.junit.Test;
 import org.springframework.context.annotation.AnnotationConfigApplicationContext;
+import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.support.AbstractApplicationContext;
+import org.springframework.core.convert.support.DefaultConversionService;
 import org.springframework.data.elasticsearch.annotations.Document;
 import org.springframework.data.elasticsearch.client.reactive.ReactiveElasticsearchClient;
+import org.springframework.data.elasticsearch.core.ElasticsearchEntityMapper;
 import org.springframework.data.elasticsearch.core.ElasticsearchRestTemplate;
-import org.springframework.data.elasticsearch.core.ElasticsearchTemplate;
+import org.springframework.data.elasticsearch.core.EntityMapper;
 import org.springframework.data.elasticsearch.core.ReactiveElasticsearchTemplate;
 import org.springframework.data.elasticsearch.core.convert.ElasticsearchConverter;
 import org.springframework.data.elasticsearch.core.mapping.SimpleElasticsearchMappingContext;
@@ -89,6 +92,13 @@ public class ElasticsearchConfigurationSupportUnitTests {
 		assertThat(context.getBean(ReactiveElasticsearchTemplate.class)).isNotNull();
 	}
 
+	@Test // DATAES-530
+	public void usesConfiguredEntityMapper() {
+
+		AbstractApplicationContext context = new AnnotationConfigApplicationContext(EntityMapperConfig.class);
+		assertThat(context.getBean(EntityMapper.class)).isInstanceOf(ElasticsearchEntityMapper.class);
+	}
+
 	@Configuration
 	static class StubConfig extends ElasticsearchConfigurationSupport {
 
@@ -112,6 +122,21 @@ public class ElasticsearchConfigurationSupportUnitTests {
 		}
 	}
 
+	@Configuration
+	static class EntityMapperConfig extends ElasticsearchConfigurationSupport {
+
+		@Bean
+		@Override
+		public EntityMapper entityMapper() {
+
+			ElasticsearchEntityMapper entityMapper = new ElasticsearchEntityMapper(elasticsearchMappingContext(),
+					new DefaultConversionService());
+			entityMapper.setConversions(elasticsearchCustomConversions());
+
+			return entityMapper;
+		}
+	}
+
 	@Document(indexName = "config-support-tests")
 	static class Entity {}
 }

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

@@ -16,6 +16,7 @@
 package org.springframework.data.elasticsearch.core;
 
 import java.io.IOException;
+import java.util.Map;
 
 /**
  * @author Artur Konczak
@@ -38,4 +39,14 @@ public class CustomEntityMapper implements EntityMapper {
 		//mapping text to Object
 		return null;
 	}
+
+	@Override
+	public Map<String, Object> mapObject(Object source) {
+		return null;
+	}
+
+	@Override
+	public <T> T readObject(Map<String, Object> source, Class<T> targetType) {
+		return null;
+	}
 }

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

@@ -15,14 +15,20 @@
  */
 package org.springframework.data.elasticsearch.core;
 
+import static java.util.Arrays.*;
+import static org.hamcrest.Matchers.*;
+import static org.junit.Assert.*;
+import static org.mockito.Mockito.*;
+
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+
 import java.util.Arrays;
+import java.util.Collection;
 import java.util.HashMap;
 import java.util.LinkedList;
 import java.util.Map;
 
-import com.fasterxml.jackson.databind.util.ArrayIterator;
-import lombok.Getter;
-import lombok.NoArgsConstructor;
 import org.elasticsearch.action.get.GetResponse;
 import org.elasticsearch.action.get.MultiGetItemResponse;
 import org.elasticsearch.action.get.MultiGetResponse;
@@ -36,8 +42,12 @@ import org.elasticsearch.search.aggregations.bucket.terms.Terms;
 import org.junit.Before;
 import org.junit.Ignore;
 import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.junit.runners.Parameterized.Parameters;
 import org.mockito.Mock;
 import org.mockito.MockitoAnnotations;
+import org.springframework.core.convert.support.DefaultConversionService;
 import org.springframework.data.domain.Page;
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.elasticsearch.annotations.Document;
@@ -46,11 +56,7 @@ import org.springframework.data.elasticsearch.core.mapping.SimpleElasticsearchMa
 import org.springframework.data.elasticsearch.entities.Car;
 import org.springframework.data.elasticsearch.entities.SampleEntity;
 
-import static java.util.Arrays.asList;
-import static org.assertj.core.api.Assertions.*;
-import static org.hamcrest.Matchers.*;
-import static org.junit.Assert.*;
-import static org.mockito.Mockito.*;
+import com.fasterxml.jackson.databind.util.ArrayIterator;
 
 /**
  * @author Artur Konczak
@@ -58,24 +64,43 @@ import static org.mockito.Mockito.*;
  * @author Chris White
  * @author Mark Paluch
  * @author Ilkang Na
+ * @author Christoph Strobl
  */
+@RunWith(Parameterized.class)
 public class DefaultResultMapperTests {
 
 	private DefaultResultMapper resultMapper;
+	private SimpleElasticsearchMappingContext context;
+	private EntityMapper entityMapper;
 
-	@Mock
-	private SearchResponse response;
+	@Mock private SearchResponse response;
+
+	public DefaultResultMapperTests(SimpleElasticsearchMappingContext context, EntityMapper entityMapper) {
+
+		this.context = context;
+		this.entityMapper = entityMapper;
+	}
+
+	@Parameters
+	public static Collection<Object[]> data() {
+
+		SimpleElasticsearchMappingContext context = new SimpleElasticsearchMappingContext();
+
+		return Arrays.asList(new Object[] { context, new DefaultEntityMapper(context) },
+				new Object[] { context, new ElasticsearchEntityMapper(context, new DefaultConversionService()) });
+	}
 
 	@Before
 	public void init() {
+
 		MockitoAnnotations.initMocks(this);
-		resultMapper = new DefaultResultMapper(new SimpleElasticsearchMappingContext());
+		resultMapper = new DefaultResultMapper(context, entityMapper);
 	}
 
 	@Test
 	public void shouldMapAggregationsToPage() {
-		//Given
-		SearchHit[] hits = {createCarHit("Ford", "Grat"), createCarHit("BMW", "Arrow")};
+		// Given
+		SearchHit[] hits = { createCarHit("Ford", "Grat"), createCarHit("BMW", "Arrow") };
 		SearchHits searchHits = mock(SearchHits.class);
 		when(searchHits.getTotalHits()).thenReturn(2L);
 		when(searchHits.iterator()).thenReturn(new ArrayIterator(hits));
@@ -84,10 +109,10 @@ public class DefaultResultMapperTests {
 		Aggregations aggregations = new Aggregations(asList(createCarAggregation()));
 		when(response.getAggregations()).thenReturn(aggregations);
 
-		//When
+		// When
 		AggregatedPage<Car> page = (AggregatedPage<Car>) resultMapper.mapResults(response, Car.class, Pageable.unpaged());
 
-		//Then
+		// Then
 		page.hasFacets();
 		assertThat(page.hasAggregations(), is(true));
 		assertThat(page.getAggregation("Diesel").getName(), is("Diesel"));
@@ -95,17 +120,17 @@ public class DefaultResultMapperTests {
 
 	@Test
 	public void shouldMapSearchRequestToPage() {
-		//Given
-		SearchHit[] hits = {createCarHit("Ford", "Grat"), createCarHit("BMW", "Arrow")};
+		// Given
+		SearchHit[] hits = { createCarHit("Ford", "Grat"), createCarHit("BMW", "Arrow") };
 		SearchHits searchHits = mock(SearchHits.class);
 		when(searchHits.getTotalHits()).thenReturn(2L);
 		when(searchHits.iterator()).thenReturn(new ArrayIterator(hits));
 		when(response.getHits()).thenReturn(searchHits);
 
-		//When
+		// When
 		Page<Car> page = resultMapper.mapResults(response, Car.class, Pageable.unpaged());
 
-		//Then
+		// Then
 		assertThat(page.hasContent(), is(true));
 		assertThat(page.getTotalElements(), is(2L));
 		assertThat(page.getContent().get(0).getName(), is("Ford"));
@@ -113,17 +138,17 @@ public class DefaultResultMapperTests {
 
 	@Test
 	public void shouldMapPartialSearchRequestToObject() {
-		//Given
-		SearchHit[] hits = {createCarPartialHit("Ford", "Grat"), createCarPartialHit("BMW", "Arrow")};
+		// Given
+		SearchHit[] hits = { createCarPartialHit("Ford", "Grat"), createCarPartialHit("BMW", "Arrow") };
 		SearchHits searchHits = mock(SearchHits.class);
 		when(searchHits.getTotalHits()).thenReturn(2L);
 		when(searchHits.iterator()).thenReturn(new ArrayIterator(hits));
 		when(response.getHits()).thenReturn(searchHits);
 
-		//When
+		// When
 		Page<Car> page = resultMapper.mapResults(response, Car.class, Pageable.unpaged());
 
-		//Then
+		// Then
 		assertThat(page.hasContent(), is(true));
 		assertThat(page.getTotalElements(), is(2L));
 		assertThat(page.getContent().get(0).getName(), is("Ford"));
@@ -131,14 +156,14 @@ public class DefaultResultMapperTests {
 
 	@Test
 	public void shouldMapGetRequestToObject() {
-		//Given
+		// Given
 		GetResponse response = mock(GetResponse.class);
 		when(response.getSourceAsString()).thenReturn(createJsonCar("Ford", "Grat"));
 
-		//When
+		// When
 		Car result = resultMapper.mapResult(response, Car.class);
 
-		//Then
+		// Then
 		assertThat(result, notNullValue());
 		assertThat(result.getModel(), is("Grat"));
 		assertThat(result.getName(), is("Ford"));

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

@@ -0,0 +1,677 @@
+/*
+ * Copyright 2019 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
+ *
+ *      http://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 org.assertj.core.api.Assertions.*;
+
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.springframework.core.convert.converter.Converter;
+import org.springframework.core.convert.support.GenericConversionService;
+import org.springframework.data.annotation.Id;
+import org.springframework.data.annotation.ReadOnlyProperty;
+import org.springframework.data.annotation.Transient;
+import org.springframework.data.annotation.TypeAlias;
+import org.springframework.data.convert.ReadingConverter;
+import org.springframework.data.convert.WritingConverter;
+import org.springframework.data.elasticsearch.core.convert.ElasticsearchCustomConversions;
+import org.springframework.data.elasticsearch.core.geo.GeoPoint;
+import org.springframework.data.elasticsearch.core.mapping.SimpleElasticsearchMappingContext;
+import org.springframework.data.elasticsearch.entities.Car;
+import org.springframework.data.elasticsearch.entities.GeoEntity;
+import org.springframework.data.geo.Point;
+
+/**
+ * @author Christoph Strobl
+ */
+public class ElasticsearchEntityMapperUnitTests {
+
+	static final String JSON_STRING = "{\"_class\":\"org.springframework.data.elasticsearch.entities.Car\",\"name\":\"Grat\",\"model\":\"Ford\"}";
+	static final String CAR_MODEL = "Ford";
+	static final String CAR_NAME = "Grat";
+	ElasticsearchEntityMapper entityMapper;
+
+	Person sarahConnor;
+	Person kyleReese;
+	Person t800;
+
+	Inventory gun = new Gun("Glock 19", 33);
+	Inventory grenade = new Grenade("40 mm");
+	Inventory rifle = new Rifle("AR-18 Assault Rifle", 3.17, 40);
+	Inventory shotGun = new ShotGun("Ithaca 37 Pump Shotgun");
+
+	Address observatoryRoad;
+	Place bigBunsCafe;
+
+	Map<String, Object> sarahAsMap;
+	Map<String, Object> t800AsMap;
+	Map<String, Object> kyleAsMap;
+	Map<String, Object> gratiotAveAsMap;
+	Map<String, Object> locationAsMap;
+	Map<String, Object> gunAsMap;
+	Map<String, Object> grenadeAsMap;
+	Map<String, Object> rifleAsMap;
+	Map<String, Object> shotGunAsMap;
+	Map<String, Object> bigBunsCafeAsMap;
+
+	@Before
+	public void init() {
+
+		SimpleElasticsearchMappingContext mappingContext = new SimpleElasticsearchMappingContext();
+		mappingContext.setInitialEntitySet(Collections.singleton(Rifle.class));
+		mappingContext.afterPropertiesSet();
+
+		entityMapper = new ElasticsearchEntityMapper(mappingContext, new GenericConversionService());
+		entityMapper.setConversions(
+				new ElasticsearchCustomConversions(Arrays.asList(new ShotGunToMapConverter(), new MapToShotGunConverter())));
+		entityMapper.afterPropertiesSet();
+
+		sarahConnor = new Person();
+		sarahConnor.id = "sarah";
+		sarahConnor.name = "Sarah Connor";
+		sarahConnor.gender = Gender.MAN;
+
+		kyleReese = new Person();
+		kyleReese.id = "kyle";
+		kyleReese.gender = Gender.MAN;
+		kyleReese.name = "Kyle Reese";
+
+		t800 = new Person();
+		t800.id = "t800";
+		t800.name = "T-800";
+		t800.gender = Gender.MACHINE;
+
+		t800AsMap = new LinkedHashMap<>();
+		t800AsMap.put("id", "t800");
+		t800AsMap.put("name", "T-800");
+		t800AsMap.put("gender", "MACHINE");
+		t800AsMap.put("_class", "org.springframework.data.elasticsearch.core.ElasticsearchEntityMapperUnitTests$Person");
+
+		observatoryRoad = new Address();
+		observatoryRoad.city = "Los Angeles";
+		observatoryRoad.street = "2800 East Observatory Road";
+		observatoryRoad.location = new Point(34.118347D, -118.3026284D);
+
+		bigBunsCafe = new Place();
+		bigBunsCafe.name = "Big Buns Cafe";
+		bigBunsCafe.city = "Los Angeles";
+		bigBunsCafe.street = "15 South Fremont Avenue";
+		bigBunsCafe.location = new Point(34.0945637D, -118.1545845D);
+
+		sarahAsMap = new LinkedHashMap<>();
+		sarahAsMap.put("id", "sarah");
+		sarahAsMap.put("name", "Sarah Connor");
+		sarahAsMap.put("gender", "MAN");
+		sarahAsMap.put("_class", "org.springframework.data.elasticsearch.core.ElasticsearchEntityMapperUnitTests$Person");
+
+		kyleAsMap = new LinkedHashMap<>();
+		kyleAsMap.put("id", "kyle");
+		kyleAsMap.put("gender", "MAN");
+		kyleAsMap.put("name", "Kyle Reese");
+
+		locationAsMap = new LinkedHashMap<>();
+		locationAsMap.put("lat", 34.118347D);
+		locationAsMap.put("lon", -118.3026284D);
+
+		gratiotAveAsMap = new LinkedHashMap<>();
+		gratiotAveAsMap.put("city", "Los Angeles");
+		gratiotAveAsMap.put("street", "2800 East Observatory Road");
+		gratiotAveAsMap.put("location", locationAsMap);
+
+		bigBunsCafeAsMap = new LinkedHashMap<>();
+		bigBunsCafeAsMap.put("name", "Big Buns Cafe");
+		bigBunsCafeAsMap.put("city", "Los Angeles");
+		bigBunsCafeAsMap.put("street", "15 South Fremont Avenue");
+		bigBunsCafeAsMap.put("location", new LinkedHashMap<>());
+		((HashMap<String, Object>) bigBunsCafeAsMap.get("location")).put("lat", 34.0945637D);
+		((HashMap<String, Object>) bigBunsCafeAsMap.get("location")).put("lon", -118.1545845D);
+		bigBunsCafeAsMap.put("_class",
+				"org.springframework.data.elasticsearch.core.ElasticsearchEntityMapperUnitTests$Place");
+
+		gunAsMap = new LinkedHashMap<>();
+		gunAsMap.put("label", "Glock 19");
+		gunAsMap.put("shotsPerMagazine", 33);
+		gunAsMap.put("_class", Gun.class.getName());
+
+		grenadeAsMap = new LinkedHashMap<>();
+		grenadeAsMap.put("label", "40 mm");
+		grenadeAsMap.put("_class", Grenade.class.getName());
+
+		rifleAsMap = new LinkedHashMap<>();
+		rifleAsMap.put("label", "AR-18 Assault Rifle");
+		rifleAsMap.put("weight", 3.17D);
+		rifleAsMap.put("maxShotsPerMagazine", 40);
+		rifleAsMap.put("_class", "rifle");
+
+		shotGunAsMap = new LinkedHashMap<>();
+		shotGunAsMap.put("model", "Ithaca 37 Pump Shotgun");
+		shotGunAsMap.put("_class", ShotGun.class.getName());
+	}
+
+	@Test // DATAES-530
+	public void shouldMapObjectToJsonString() throws IOException {
+		// Given
+
+		// When
+		String jsonResult = entityMapper.mapToString(Car.builder().model(CAR_MODEL).name(CAR_NAME).build());
+
+		// Then
+		assertThat(jsonResult).isEqualTo(JSON_STRING);
+	}
+
+	@Test // DATAES-530
+	public void shouldMapJsonStringToObject() throws IOException {
+		// Given
+
+		// When
+		Car result = entityMapper.mapToObject(JSON_STRING, Car.class);
+
+		// Then
+		assertThat(result.getName()).isEqualTo(CAR_NAME);
+		assertThat(result.getModel()).isEqualTo(CAR_MODEL);
+	}
+
+	@Test // DATAES-530
+	public void shouldMapGeoPointElasticsearchNames() throws IOException {
+		// given
+		final Point point = new Point(10, 20);
+		final String pointAsString = point.getX() + "," + point.getY();
+		final double[] pointAsArray = { point.getX(), point.getY() };
+		final GeoEntity geoEntity = GeoEntity.builder().pointA(point).pointB(GeoPoint.fromPoint(point))
+				.pointC(pointAsString).pointD(pointAsArray).build();
+		// when
+		String jsonResult = entityMapper.mapToString(geoEntity);
+
+		// then
+		assertThat(jsonResult).contains(pointTemplate("pointA", point));
+		assertThat(jsonResult).contains(pointTemplate("pointB", point));
+		assertThat(jsonResult).contains(String.format(Locale.ENGLISH, "\"%s\":\"%s\"", "pointC", pointAsString));
+		assertThat(jsonResult)
+				.contains(String.format(Locale.ENGLISH, "\"%s\":[%.1f,%.1f]", "pointD", pointAsArray[0], pointAsArray[1]));
+	}
+
+	@Test // DATAES-530
+	public void ignoresReadOnlyProperties() throws IOException {
+
+		// given
+		Sample sample = new Sample();
+		sample.readOnly = "readOnly";
+		sample.property = "property";
+		sample.javaTransientProperty = "javaTransient";
+		sample.annotatedTransientProperty = "transient";
+
+		// when
+		String result = entityMapper.mapToString(sample);
+
+		// then
+		assertThat(result).contains("\"property\"");
+		assertThat(result).contains("\"javaTransient\"");
+
+		assertThat(result).doesNotContain("readOnly");
+		assertThat(result).doesNotContain("annotatedTransientProperty");
+	}
+
+	@Test // DATAES-530
+	public void writesNestedEntity() {
+
+		Person person = new Person();
+		person.birthdate = new Date();
+		person.gender = Gender.MAN;
+		person.address = observatoryRoad;
+
+		LinkedHashMap<String, Object> sink = writeToMap(person);
+
+		assertThat(sink.get("address")).isEqualTo(gratiotAveAsMap);
+	}
+
+	@Test // DATAES-530
+	public void writesConcreteList() throws IOException {
+
+		Person ginger = new Person();
+		ginger.id = "ginger";
+		ginger.gender = Gender.MAN;
+
+		sarahConnor.coWorkers = Arrays.asList(kyleReese, ginger);
+
+		LinkedHashMap<String, Object> target = writeToMap(sarahConnor);
+		assertThat((List) target.get("coWorkers")).hasSize(2).contains(kyleAsMap);
+	}
+
+	@Test // DATAES-530
+	public void writesInterfaceList() throws IOException {
+
+		Inventory gun = new Gun("Glock 19", 33);
+		Inventory grenade = new Grenade("40 mm");
+
+		sarahConnor.inventoryList = Arrays.asList(gun, grenade);
+
+		LinkedHashMap<String, Object> target = writeToMap(sarahConnor);
+		assertThat((List) target.get("inventoryList")).containsExactly(gunAsMap, grenadeAsMap);
+	}
+
+	@Test // DATAES-530
+	public void readTypeCorrectly() {
+
+		Person target = entityMapper.read(Person.class, sarahAsMap);
+
+		assertThat(target).isEqualTo(sarahConnor);
+	}
+
+	@Test // DATAES-530
+	public void readListOfConcreteTypesCorrectly() {
+
+		sarahAsMap.put("coWorkers", Arrays.asList(kyleAsMap));
+
+		Person target = entityMapper.read(Person.class, sarahAsMap);
+
+		assertThat(target.getCoWorkers()).contains(kyleReese);
+	}
+
+	@Test // DATAES-530
+	public void readListOfInterfacesTypesCorrectly() {
+
+		sarahAsMap.put("inventoryList", Arrays.asList(gunAsMap, grenadeAsMap));
+
+		Person target = entityMapper.read(Person.class, sarahAsMap);
+
+		assertThat(target.getInventoryList()).containsExactly(gun, grenade);
+	}
+
+	@Test // DATAES-530
+	public void writeMapOfConcreteType() {
+
+		sarahConnor.shippingAddresses = new LinkedHashMap<>();
+		sarahConnor.shippingAddresses.put("home", observatoryRoad);
+
+		LinkedHashMap<String, Object> target = writeToMap(sarahConnor);
+		assertThat(target.get("shippingAddresses")).isInstanceOf(Map.class);
+		assertThat(target.get("shippingAddresses")).isEqualTo(Collections.singletonMap("home", gratiotAveAsMap));
+	}
+
+	@Test // DATAES-530
+	public void writeMapOfInterfaceType() {
+
+		sarahConnor.inventoryMap = new LinkedHashMap<>();
+		sarahConnor.inventoryMap.put("glock19", gun);
+		sarahConnor.inventoryMap.put("40 mm grenade", grenade);
+
+		LinkedHashMap<String, Object> target = writeToMap(sarahConnor);
+		assertThat(target.get("inventoryMap")).isInstanceOf(Map.class);
+		assertThat((Map) target.get("inventoryMap")).containsEntry("glock19", gunAsMap).containsEntry("40 mm grenade",
+				grenadeAsMap);
+	}
+
+	@Test // DATAES-530
+	public void readConcreteMapCorrectly() {
+
+		sarahAsMap.put("shippingAddresses", Collections.singletonMap("home", gratiotAveAsMap));
+
+		Person target = entityMapper.read(Person.class, sarahAsMap);
+
+		assertThat(target.getShippingAddresses()).hasSize(1).containsEntry("home", observatoryRoad);
+	}
+
+	@Test // DATAES-530
+	public void readInterfaceMapCorrectly() {
+
+		sarahAsMap.put("inventoryMap", Collections.singletonMap("glock19", gunAsMap));
+
+		Person target = entityMapper.read(Person.class, sarahAsMap);
+
+		assertThat(target.getInventoryMap()).hasSize(1).containsEntry("glock19", gun);
+	}
+
+	@Test // DATAES-530
+	public void genericWriteList() {
+
+		Skynet skynet = new Skynet();
+		skynet.objectList = new ArrayList<>();
+		skynet.objectList.add(t800);
+		skynet.objectList.add(gun);
+
+		LinkedHashMap<String, Object> target = writeToMap(skynet);
+
+		assertThat((List<Object>) target.get("objectList")).containsExactly(t800AsMap, gunAsMap);
+	}
+
+	@Test // DATAES-530
+	public void readGenericList() {
+
+		LinkedHashMap<String, Object> source = new LinkedHashMap<>();
+		source.put("objectList", Arrays.asList(t800AsMap, gunAsMap));
+
+		Skynet target = entityMapper.read(Skynet.class, source);
+
+		assertThat(target.getObjectList()).containsExactly(t800, gun);
+	}
+
+	@Test // DATAES-530
+	public void genericWriteListWithList() {
+
+		Skynet skynet = new Skynet();
+		skynet.objectList = new ArrayList<>();
+		skynet.objectList.add(Arrays.asList(t800, gun));
+
+		LinkedHashMap<String, Object> target = writeToMap(skynet);
+
+		assertThat((List<Object>) target.get("objectList")).containsExactly(Arrays.asList(t800AsMap, gunAsMap));
+	}
+
+	@Test // DATAES-530
+	public void readGenericListList() {
+
+		LinkedHashMap<String, Object> source = new LinkedHashMap<>();
+		source.put("objectList", Arrays.asList(Arrays.asList(t800AsMap, gunAsMap)));
+
+		Skynet target = entityMapper.read(Skynet.class, source);
+
+		assertThat(target.getObjectList()).containsExactly(Arrays.asList(t800, gun));
+	}
+
+	@Test // DATAES-530
+	public void writeGenericMap() {
+
+		Skynet skynet = new Skynet();
+		skynet.objectMap = new LinkedHashMap<>();
+		skynet.objectMap.put("gun", gun);
+		skynet.objectMap.put("grenade", grenade);
+
+		LinkedHashMap<String, Object> target = writeToMap(skynet);
+
+		assertThat((Map<String, Object>) target.get("objectMap")).containsEntry("gun", gunAsMap).containsEntry("grenade",
+				grenadeAsMap);
+	}
+
+	@Test // DATAES-530
+	public void readGenericMap() {
+
+		LinkedHashMap<String, Object> source = new LinkedHashMap<>();
+		source.put("objectMap", Collections.singletonMap("glock19", gunAsMap));
+
+		Skynet target = entityMapper.read(Skynet.class, source);
+
+		assertThat(target.getObjectMap()).containsEntry("glock19", gun);
+	}
+
+	@Test // DATAES-530
+	public void writeGenericMapMap() {
+
+		Skynet skynet = new Skynet();
+		skynet.objectMap = new LinkedHashMap<>();
+		skynet.objectMap.put("inventory", Collections.singletonMap("glock19", gun));
+
+		LinkedHashMap<String, Object> target = writeToMap(skynet);
+
+		assertThat((Map<String, Object>) target.get("objectMap")).containsEntry("inventory",
+				Collections.singletonMap("glock19", gunAsMap));
+	}
+
+	@Test // DATAES-530
+	public void readGenericMapMap() {
+
+		LinkedHashMap<String, Object> source = new LinkedHashMap<>();
+		source.put("objectMap", Collections.singletonMap("inventory", Collections.singletonMap("glock19", gunAsMap)));
+
+		Skynet target = entityMapper.read(Skynet.class, source);
+
+		assertThat(target.getObjectMap()).containsEntry("inventory", Collections.singletonMap("glock19", gun));
+	}
+
+	@Test // DATAES-530
+	public void readsNestedEntity() {
+
+		sarahAsMap.put("address", gratiotAveAsMap);
+
+		Person target = entityMapper.read(Person.class, sarahAsMap);
+
+		assertThat(target.getAddress()).isEqualTo(observatoryRoad);
+	}
+
+	@Test // DATAES-530
+	public void readsNestedObjectEntity() {
+
+		LinkedHashMap<String, Object> source = new LinkedHashMap<>();
+		source.put("object", t800AsMap);
+
+		Skynet target = entityMapper.read(Skynet.class, source);
+
+		assertThat(target.getObject()).isEqualTo(t800);
+	}
+
+	@Test // DATAES-530
+	public void writesAliased() {
+		assertThat(writeToMap(rifle)).containsEntry("_class", "rifle").doesNotContainValue(Rifle.class.getName());
+	}
+
+	@Test // DATAES-530
+	public void writesNestedAliased() {
+
+		t800.inventoryList = Collections.singletonList(rifle);
+		LinkedHashMap<String, Object> target = writeToMap(t800);
+
+		assertThat((List) target.get("inventoryList")).contains(rifleAsMap);
+	}
+
+	@Test // DATAES-530
+	public void readsAliased() {
+		assertThat(entityMapper.read(Inventory.class, rifleAsMap)).isEqualTo(rifle);
+	}
+
+	@Test // DATAES-530
+	public void readsNestedAliased() {
+
+		t800AsMap.put("inventoryList", Collections.singletonList(rifleAsMap));
+
+		assertThat(entityMapper.read(Person.class, t800AsMap).getInventoryList()).containsExactly(rifle);
+	}
+
+	@Test // DATAES-530
+	public void appliesCustomConverterForWrite() {
+		assertThat(writeToMap(shotGun)).isEqualTo(shotGunAsMap);
+	}
+
+	@Test // DATAES-530
+	public void appliesCustomConverterForRead() {
+		assertThat(entityMapper.read(Inventory.class, shotGunAsMap)).isEqualTo(shotGun);
+	}
+
+	@Test // DATAES-530
+	public void writeSubTypeCorrectly() {
+
+		sarahConnor.address = bigBunsCafe;
+
+		LinkedHashMap<String, Object> target = writeToMap(sarahConnor);
+
+		assertThat(target.get("address")).isEqualTo(bigBunsCafeAsMap);
+	}
+
+	@Test // DATAES-530
+	public void readSubTypeCorrectly() {
+
+		sarahAsMap.put("address", bigBunsCafeAsMap);
+
+		Person target = entityMapper.read(Person.class, sarahAsMap);
+
+		assertThat(target.address).isEqualTo(bigBunsCafe);
+	}
+
+	private String pointTemplate(String name, Point point) {
+		return String.format(Locale.ENGLISH, "\"%s\":{\"lat\":%.1f,\"lon\":%.1f}", name, point.getX(), point.getY());
+	}
+
+	private LinkedHashMap<String, Object> writeToMap(Object source) {
+
+		LinkedHashMap<String, Object> sink = new LinkedHashMap<>();
+		entityMapper.write(source, sink);
+		return sink;
+	}
+
+	public static class Sample {
+
+		public @ReadOnlyProperty String readOnly;
+		public @Transient String annotatedTransientProperty;
+		public transient String javaTransientProperty;
+		public String property;
+	}
+
+	@Data
+	static class Person {
+
+		@Id String id;
+		String name;
+		Date birthdate;
+		Gender gender;
+		Address address;
+
+		List<Person> coWorkers;
+		List<Inventory> inventoryList;
+		Map<String, Address> shippingAddresses;
+		Map<String, Inventory> inventoryMap;
+	}
+
+	enum Gender {
+
+		MAN("1"), MACHINE("0");
+
+		String theValue;
+
+		Gender(String theValue) {
+			this.theValue = theValue;
+		}
+
+		public String getTheValue() {
+			return theValue;
+		}
+	}
+
+	interface Inventory {
+
+		String getLabel();
+	}
+
+	@Getter
+	@RequiredArgsConstructor
+	@EqualsAndHashCode
+	static class Gun implements Inventory {
+
+		final String label;
+		final int shotsPerMagazine;
+
+		@Override
+		public String getLabel() {
+			return label;
+		}
+	}
+
+	@RequiredArgsConstructor
+	@EqualsAndHashCode
+	static class Grenade implements Inventory {
+
+		final String label;
+
+		@Override
+		public String getLabel() {
+			return label;
+		}
+	}
+
+	@TypeAlias("rifle")
+	@EqualsAndHashCode
+	@RequiredArgsConstructor
+	static class Rifle implements Inventory {
+
+		final String label;
+		final double weight;
+		final int maxShotsPerMagazine;
+
+		@Override
+		public String getLabel() {
+			return label;
+		}
+	}
+
+	@EqualsAndHashCode
+	@RequiredArgsConstructor
+	static class ShotGun implements Inventory {
+
+		final String label;
+
+		@Override
+		public String getLabel() {
+			return label;
+		}
+	}
+
+	@Data
+	static class Address {
+
+		Point location;
+		String street;
+		String city;
+	}
+
+	@Data
+	static class Place extends Address {
+
+		String name;
+	}
+
+	@Data
+	static class Skynet {
+
+		Object object;
+		List<Object> objectList;
+		Map<String, Object> objectMap;
+	}
+
+	@WritingConverter
+	static class ShotGunToMapConverter implements Converter<ShotGun, Map<String, Object>> {
+
+		@Override
+		public Map<String, Object> convert(ShotGun source) {
+
+			LinkedHashMap<String, Object> target = new LinkedHashMap<>();
+			target.put("model", source.getLabel());
+			target.put("_class", ShotGun.class.getName());
+			return target;
+		}
+	}
+
+	@ReadingConverter
+	static class MapToShotGunConverter implements Converter<Map<String, Object>, ShotGun> {
+
+		@Override
+		public ShotGun convert(Map<String, Object> source) {
+			return new ShotGun(source.get("model").toString());
+		}
+	}
+}

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

@@ -817,6 +817,11 @@ public class ElasticsearchTemplateTests {
 				}
 				return new AggregatedPageImpl<>((List<T>) values);
 			}
+
+			@Override
+			public <T> T mapSearchHit(SearchHit searchHit, Class<T> type) {
+				return null;
+			}
 		});
 		// then
 		assertThat(page, is(notNullValue()));
@@ -956,6 +961,11 @@ public class ElasticsearchTemplateTests {
 			}
 			return new AggregatedPageImpl<T>(Collections.EMPTY_LIST, response.getScrollId());
 		}
+
+		@Override
+		public <T> T mapSearchHit(SearchHit searchHit, Class<T> type) {
+			return null;
+		}
 	};
 
 	/*
@@ -1356,6 +1366,11 @@ public class ElasticsearchTemplateTests {
 				}
 				return null;
 			}
+
+			@Override
+			public <T> T mapSearchHit(SearchHit searchHit, Class<T> type) {
+				return null;
+			}
 		});
 
 		assertThat(sampleEntities.getContent().get(0).getHighlightedMessage(), is(highlightedMessage));
@@ -1410,6 +1425,11 @@ public class ElasticsearchTemplateTests {
 				}
 				return null;
 			}
+
+			@Override
+			public <T> T mapSearchHit(SearchHit searchHit, Class<T> type) {
+				return null;
+			}
 		});
 	}
 
@@ -1451,6 +1471,11 @@ public class ElasticsearchTemplateTests {
 				}
 				return null;
 			}
+
+			@Override
+			public <T> T mapSearchHit(SearchHit searchHit, Class<T> type) {
+				return null;
+			}
 		});
 	}
 
@@ -1491,6 +1516,11 @@ public class ElasticsearchTemplateTests {
 				}
 				return null;
 			}
+
+			@Override
+			public <T> T mapSearchHit(SearchHit searchHit, Class<T> type) {
+				return null;
+			}
 		});
 	}
 
@@ -1603,6 +1633,11 @@ public class ElasticsearchTemplateTests {
 				}
 				return new AggregatedPageImpl<>((List<T>) values);
 			}
+
+			@Override
+			public <T> T mapSearchHit(SearchHit searchHit, Class<T> type) {
+				return null;
+			}
 		});
 		assertThat(page, is(notNullValue()));
 		assertThat(page.getContent().size(), is(1));
@@ -1812,6 +1847,11 @@ public class ElasticsearchTemplateTests {
 				}
 				return null;
 			}
+
+			@Override
+			public <T> T mapSearchHit(SearchHit searchHit, Class<T> type) {
+				return null;
+			}
 		});
 		assertThat(sampleEntities.getTotalElements(), is(equalTo(2L)));
 		assertThat(sampleEntities.getContent().get(0).get("userId"), is(person1.get("userId")));
@@ -2372,6 +2412,11 @@ public class ElasticsearchTemplateTests {
 				}
 				return new AggregatedPageImpl<>((List<T>) values);
 			}
+
+			@Override
+			public <T> T mapSearchHit(SearchHit searchHit, Class<T> type) {
+				return null;
+			}
 		});
 
 		assertThat(page.getTotalElements(), is(2l));

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

@@ -101,7 +101,7 @@ public class MappingBuilderTests {
 		assertThat(xContentBuilderToString(xContentBuilder), is(expected));
 	}
 
-	@Test
+	@Test // DATAES-530
 	public void shouldAddStockPriceDocumentToIndex() throws IOException {
 		// Given
 
@@ -112,8 +112,9 @@ public class MappingBuilderTests {
 		String symbol = "AU";
 		double price = 2.34;
 		String id = "abc";
+
 		elasticsearchTemplate
-				.index(buildIndex(StockPrice.builder().id(id).symbol(symbol).price(new BigDecimal(price)).build()));
+				.index(buildIndex(StockPrice.builder().id(id).symbol(symbol).price(BigDecimal.valueOf(price)).build()));
 		elasticsearchTemplate.refresh(StockPrice.class);
 
 		SearchQuery searchQuery = new NativeSearchQueryBuilder().withQuery(matchAllQuery()).build();
@@ -122,7 +123,7 @@ public class MappingBuilderTests {
 		assertThat(result.size(), is(1));
 		StockPrice entry = result.get(0);
 		assertThat(entry.getSymbol(), is(symbol));
-		assertThat(entry.getPrice(), is(new BigDecimal(price)));
+		assertThat(entry.getPrice(), is(BigDecimal.valueOf(price)));
 	}
 
 	@Test

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

@@ -38,6 +38,7 @@ import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.runner.RunWith;
+import org.springframework.core.convert.support.DefaultConversionService;
 import org.springframework.dao.DataAccessResourceFailureException;
 import org.springframework.data.annotation.Id;
 import org.springframework.data.domain.PageRequest;
@@ -86,7 +87,9 @@ public class ReactiveElasticsearchTemplateTests {
 		restTemplate.putMapping(SampleEntity.class);
 		restTemplate.refresh(SampleEntity.class);
 
-		template = new ReactiveElasticsearchTemplate(TestUtils.reactiveClient());
+		template = new ReactiveElasticsearchTemplate(TestUtils.reactiveClient(), restTemplate.getElasticsearchConverter(),
+				new DefaultResultMapper(new ElasticsearchEntityMapper(
+						restTemplate.getElasticsearchConverter().getMappingContext(), new DefaultConversionService())));
 	}
 
 	@Test // DATAES-504