Allow disabling TypeHints.

Original Pull Request #1788
Closes #1788

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

@@ -34,8 +34,6 @@ The following annotations are available:
 The most important attributes are:
 ** `indexName`: the name of the index to store this entity in.
 This can contain a SpEL template expression like `"log-#{T(java.time.LocalDate).now().toString()}"`
-** `type`: [line-through]#the mapping type.
-If not set, the lowercased simple name of the class is used.# (deprecated since version 4.0)
 ** `createIndex`: flag whether to create an index on repository bootstrapping.
 Default value is _true_.
 See <<elasticsearch.repositories.autocreation>>
@@ -170,6 +168,22 @@ public class Person {
 
 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.
 
+===== Disabling Type Hints
+
+It may be necessary to disable writing of type hints when the index that should be used already exists without having the type hints defined in its mapping and with the mapping mode set to strict. In this case, writing the type hint will produce an error, as the field cannot be added automatically.
+
+Type hints can be disabled for the whole application by overriding the method `writeTypeHints()` in a configuration class derived from `AbstractElasticsearchConfiguration` (see <<elasticsearch.clients>>).
+
+As an alternativ they can be disabled for a single index with the `@Document` annotation:
+====
+[source,java]
+----
+@Document(indexName = "index", writeTypeHint = WriteTypeHint.FALSE)
+----
+====
+
+WARNING: We strongly advise against disabling Type Hints. Only do this if you are forced to. Disabling type hints can lead to documents not being retrieved correctly from Elasticsearch in case of polymorphic data or document retrieval may fail completely.
+
 ==== Geospatial Types
 
 Geospatial types like `Point` & `GeoPoint` are converted into _lat/lon_ pairs.

src/main/java/org/springframework/data/elasticsearch/annotations/Document.java

@@ -105,4 +105,11 @@ public @interface Document {
 	 * Configuration of version management.
 	 */
 	VersionType versionType() default VersionType.EXTERNAL;
+
+	/**
+	 * Defines if type hints should be written. {@see WriteTypeHint}.
+	 *
+	 * @since 4.3
+	 */
+	WriteTypeHint writeTypeHint() default WriteTypeHint.DEFAULT;
 }

src/main/java/org/springframework/data/elasticsearch/annotations/WriteTypeHint.java

@@ -0,0 +1,40 @@
+/*
+ * Copyright 2021 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.elasticsearch.annotations;
+
+import org.springframework.data.mapping.context.MappingContext;
+
+/**
+ * Defines if type hints should be written. Used by {@link Document} annotation.
+ *
+ * @author Peter-Josef Meisch
+ * @since 4.3
+ */
+public enum WriteTypeHint {
+
+	/**
+	 * Use the global settings from the {@link MappingContext}.
+	 */
+	DEFAULT,
+	/**
+	 * Always write type hints for the entity.
+	 */
+	TRUE,
+	/**
+	 * Never write type hints for the entity.
+	 */
+	FALSE
+}

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

@@ -26,7 +26,6 @@ import org.springframework.context.annotation.ClassPathScanningCandidateComponen
 import org.springframework.context.annotation.Configuration;
 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.RefreshPolicy;
 import org.springframework.data.elasticsearch.core.convert.ElasticsearchConverter;
@@ -72,6 +71,7 @@ public class ElasticsearchConfigurationSupport {
 		mappingContext.setInitialEntitySet(getInitialEntitySet());
 		mappingContext.setSimpleTypeHolder(elasticsearchCustomConversions.getSimpleTypeHolder());
 		mappingContext.setFieldNamingStrategy(fieldNamingStrategy());
+		mappingContext.setWriteTypeHints(writeTypeHints());
 
 		return mappingContext;
 	}
@@ -171,4 +171,17 @@ public class ElasticsearchConfigurationSupport {
 	protected FieldNamingStrategy fieldNamingStrategy() {
 		return PropertyNameFieldNamingStrategy.INSTANCE;
 	}
+
+	/**
+	 * Flag specifiying if type hints (_class fields) should be written in the index. It is strongly advised to keep the
+	 * default value of {@literal true}. If you need to write to an existing index that does not have a mapping defined
+	 * for these fields and that has a strict mapping set, then it might be necessary to disable type hints. But notice
+	 * that in this case reading polymorphic types may fail.
+	 *
+	 * @return flag if type hints should be written
+	 * @since 4.3
+	 */
+	protected boolean writeTypeHints() {
+		return true;
+	}
 }

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

@@ -95,15 +95,8 @@ public class MappingElasticsearchConverter
 
 	private final MappingContext<? extends ElasticsearchPersistentEntity<?>, ElasticsearchPersistentProperty> mappingContext;
 	private final GenericConversionService conversionService;
-
 	// don't access directly, use getConversions(). to prevent null access
 	private CustomConversions conversions = new ElasticsearchCustomConversions(Collections.emptyList());
-	private final EntityInstantiators instantiators = new EntityInstantiators();
-
-	private final ElasticsearchTypeMapper typeMapper;
-
-	private final ConcurrentHashMap<String, Integer> propertyWarnings = new ConcurrentHashMap<>();
-	private final SpELContext spELContext;
 
 	public MappingElasticsearchConverter(
 			MappingContext<? extends ElasticsearchPersistentEntity<?>, ElasticsearchPersistentProperty> mappingContext) {
@@ -118,8 +111,6 @@ public class MappingElasticsearchConverter
 
 		this.mappingContext = mappingContext;
 		this.conversionService = conversionService != null ? conversionService : new DefaultConversionService();
-		this.typeMapper = ElasticsearchTypeMapper.create(mappingContext);
-		this.spELContext = new SpELContext(new MapAccessor());
 	}
 
 	@Override
@@ -157,22 +148,71 @@ public class MappingElasticsearchConverter
 		return conversions;
 	}
 
-	/*
-	 * (non-Javadoc)
-	 * @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet()
-	 */
 	@Override
 	public void afterPropertiesSet() {
 		DateFormatterRegistrar.addDateConverters(conversionService);
 		getConversions().registerConvertersIn(conversionService);
 	}
 
-	// region read
+	// region read/write
 
-	@SuppressWarnings("unchecked")
 	@Override
 	public <R> R read(Class<R> type, Document source) {
 
+		Reader reader = new Reader(mappingContext, conversionService, getConversions());
+		return reader.read(type, source);
+	}
+
+	@Override
+	public void write(Object source, Document sink) {
+
+		Assert.notNull(source, "source to map must not be null");
+
+		Writer writer = new Writer(mappingContext, conversionService, getConversions());
+		writer.write(source, sink);
+	}
+
+	/**
+	 * base class for {@link Reader} and {@link Writer} keeping the common properties
+	 */
+	private static class Base {
+
+		protected final MappingContext<? extends ElasticsearchPersistentEntity<?>, ElasticsearchPersistentProperty> mappingContext;
+		protected final ElasticsearchTypeMapper typeMapper;
+		protected final GenericConversionService conversionService;
+		protected final CustomConversions conversions;
+		protected final ConcurrentHashMap<String, Integer> propertyWarnings = new ConcurrentHashMap<>();
+
+		private Base(
+				MappingContext<? extends ElasticsearchPersistentEntity<?>, ElasticsearchPersistentProperty> mappingContext,
+				GenericConversionService conversionService, CustomConversions conversions) {
+			this.mappingContext = mappingContext;
+			this.conversionService = conversionService;
+			this.conversions = conversions;
+			this.typeMapper = ElasticsearchTypeMapper.create(mappingContext);
+		}
+	}
+
+	/**
+	 * Class to do the actual writing. The methods originally were in the MappingElasticsearchConverter class, but are
+	 * refactored to allow for keeping state during the conversion of an object.
+	 */
+	private static class Reader extends Base {
+
+		private final SpELContext spELContext;
+		private final EntityInstantiators instantiators = new EntityInstantiators();
+
+		public Reader(
+				MappingContext<? extends ElasticsearchPersistentEntity<?>, ElasticsearchPersistentProperty> mappingContext,
+				GenericConversionService conversionService, CustomConversions conversions) {
+
+			super(mappingContext, conversionService, conversions);
+			this.spELContext = new SpELContext(new MapAccessor());
+		}
+
+		@SuppressWarnings("unchecked")
+		<R> R read(Class<R> type, Document source) {
+
 			TypeInformation<R> typeHint = ClassTypeInformation.from((Class<R>) ClassUtils.getUserClass(type));
 			R r = read(typeHint, source);
 
@@ -183,7 +223,85 @@ public class MappingElasticsearchConverter
 			return r;
 		}
 
-	protected <R> R readEntity(ElasticsearchPersistentEntity<?> entity, Map<String, Object> source) {
+		@Nullable
+		@SuppressWarnings("unchecked")
+		private <R> R read(TypeInformation<R> type, Map<String, Object> source) {
+
+			Assert.notNull(source, "Source must not be null!");
+
+			TypeInformation<? extends R> typeToUse = typeMapper.readType(source, type);
+			Class<? extends R> rawType = typeToUse.getType();
+
+			if (conversions.hasCustomReadTarget(source.getClass(), rawType)) {
+				return conversionService.convert(source, rawType);
+			}
+
+			if (Document.class.isAssignableFrom(rawType)) {
+				return (R) source;
+			}
+
+			if (typeToUse.isMap()) {
+				return readMap(typeToUse, source);
+			}
+
+			if (typeToUse.equals(ClassTypeInformation.OBJECT)) {
+				return (R) source;
+			}
+			// Retrieve persistent entity info
+
+			ElasticsearchPersistentEntity<?> entity = mappingContext.getPersistentEntity(typeToUse);
+
+			if (entity == null) {
+				throw new MappingException(String.format(INVALID_TYPE_TO_READ, source, typeToUse.getType()));
+			}
+
+			return readEntity(entity, source);
+		}
+
+		@SuppressWarnings("unchecked")
+		private <R> R readMap(TypeInformation<?> type, Map<String, Object> source) {
+
+			Assert.notNull(source, "Document must not be null!");
+
+			Class<?> mapType = typeMapper.readType(source, type).getType();
+
+			TypeInformation<?> keyType = type.getComponentType();
+			TypeInformation<?> valueType = type.getMapValueType();
+
+			Class<?> rawKeyType = keyType != null ? keyType.getType() : null;
+			Class<?> rawValueType = valueType != null ? valueType.getType() : null;
+
+			Map<Object, Object> map = CollectionFactory.createMap(mapType, rawKeyType, source.keySet().size());
+
+			for (Entry<String, Object> entry : source.entrySet()) {
+
+				if (typeMapper.isTypeKey(entry.getKey())) {
+					continue;
+				}
+
+				Object key = entry.getKey();
+
+				if (rawKeyType != null && !rawKeyType.isAssignableFrom(key.getClass())) {
+					key = conversionService.convert(key, rawKeyType);
+				}
+
+				Object value = entry.getValue();
+				TypeInformation<?> defaultedValueType = valueType != null ? valueType : ClassTypeInformation.OBJECT;
+
+				if (value instanceof Map) {
+					map.put(key, read(defaultedValueType, (Map<String, Object>) value));
+				} else if (value instanceof List) {
+					map.put(key,
+							readCollectionOrArray(valueType != null ? valueType : ClassTypeInformation.LIST, (List<Object>) value));
+				} else {
+					map.put(key, getPotentiallyConvertedSimpleRead(value, rawValueType));
+				}
+			}
+
+			return (R) map;
+		}
+
+		private <R> R readEntity(ElasticsearchPersistentEntity<?> entity, Map<String, Object> source) {
 
 			ElasticsearchPersistentEntity<?> targetEntity = computeClosestEntity(entity, source);
 
@@ -294,7 +412,8 @@ public class MappingElasticsearchConverter
 		}
 
 		@Nullable
-	protected <R> R readValue(@Nullable Object value, ElasticsearchPersistentProperty property, TypeInformation<?> type) {
+		protected <R> R readValue(@Nullable Object value, ElasticsearchPersistentProperty property,
+				TypeInformation<?> type) {
 
 			if (value == null) {
 				return null;
@@ -305,7 +424,7 @@ public class MappingElasticsearchConverter
 			if (property.hasPropertyConverter()) {
 				value = propertyConverterRead(property, value);
 			} else if (TemporalAccessor.class.isAssignableFrom(property.getType())
-				&& !getConversions().hasCustomReadTarget(value.getClass(), rawType)) {
+					&& !conversions.hasCustomReadTarget(value.getClass(), rawType)) {
 
 				// log at most 5 times
 				String propertyName = property.getOwner().getType().getSimpleName() + '.' + property.getName();
@@ -342,41 +461,6 @@ public class MappingElasticsearchConverter
 			}
 		}
 
-	@Nullable
-	@SuppressWarnings("unchecked")
-	private <R> R read(TypeInformation<R> type, Map<String, Object> source) {
-
-		Assert.notNull(source, "Source must not be null!");
-
-		TypeInformation<? extends R> typeToUse = typeMapper.readType(source, type);
-		Class<? extends R> rawType = typeToUse.getType();
-
-		if (conversions.hasCustomReadTarget(source.getClass(), rawType)) {
-			return conversionService.convert(source, rawType);
-		}
-
-		if (Document.class.isAssignableFrom(rawType)) {
-			return (R) source;
-		}
-
-		if (typeToUse.isMap()) {
-			return readMap(typeToUse, source);
-		}
-
-		if (typeToUse.equals(ClassTypeInformation.OBJECT)) {
-			return (R) source;
-		}
-		// Retrieve persistent entity info
-
-		ElasticsearchPersistentEntity<?> entity = mappingContext.getPersistentEntity(typeToUse);
-
-		if (entity == null) {
-			throw new MappingException(String.format(INVALID_TYPE_TO_READ, source, typeToUse.getType()));
-		}
-
-		return readEntity(entity, source);
-	}
-
 		private Object propertyConverterRead(ElasticsearchPersistentProperty property, Object source) {
 			ElasticsearchPersistentPropertyConverter propertyConverter = Objects
 					.requireNonNull(property.getPropertyConverter());
@@ -387,7 +471,8 @@ public class MappingElasticsearchConverter
 			}
 
 			if (source instanceof List) {
-			source = ((List<?>) source).stream().map(it -> convertOnRead(propertyConverter, it)).collect(Collectors.toList());
+				source = ((List<?>) source).stream().map(it -> convertOnRead(propertyConverter, it))
+						.collect(Collectors.toList());
 			} else if (source instanceof Set) {
 				source = ((Set<?>) source).stream().map(it -> convertOnRead(propertyConverter, it)).collect(Collectors.toSet());
 			} else {
@@ -456,49 +541,6 @@ public class MappingElasticsearchConverter
 			return getPotentiallyConvertedSimpleRead(items, targetType.getType());
 		}
 
-	@SuppressWarnings("unchecked")
-	private <R> R readMap(TypeInformation<?> type, Map<String, Object> source) {
-
-		Assert.notNull(source, "Document must not be null!");
-
-		Class<?> mapType = typeMapper.readType(source, type).getType();
-
-		TypeInformation<?> keyType = type.getComponentType();
-		TypeInformation<?> valueType = type.getMapValueType();
-
-		Class<?> rawKeyType = keyType != null ? keyType.getType() : null;
-		Class<?> rawValueType = valueType != null ? valueType.getType() : null;
-
-		Map<Object, Object> map = CollectionFactory.createMap(mapType, rawKeyType, source.keySet().size());
-
-		for (Entry<String, Object> entry : source.entrySet()) {
-
-			if (typeMapper.isTypeKey(entry.getKey())) {
-				continue;
-			}
-
-			Object key = entry.getKey();
-
-			if (rawKeyType != null && !rawKeyType.isAssignableFrom(key.getClass())) {
-				key = conversionService.convert(key, rawKeyType);
-			}
-
-			Object value = entry.getValue();
-			TypeInformation<?> defaultedValueType = valueType != null ? valueType : ClassTypeInformation.OBJECT;
-
-			if (value instanceof Map) {
-				map.put(key, read(defaultedValueType, (Map<String, Object>) value));
-			} else if (value instanceof List) {
-				map.put(key,
-						readCollectionOrArray(valueType != null ? valueType : ClassTypeInformation.LIST, (List<Object>) value));
-			} else {
-				map.put(key, getPotentiallyConvertedSimpleRead(value, rawValueType));
-			}
-		}
-
-		return (R) map;
-	}
-
 		@Nullable
 		private Object getPotentiallyConvertedSimpleRead(@Nullable Object value, TypeInformation<?> targetType) {
 			return getPotentiallyConvertedSimpleRead(value, targetType.getType());
@@ -512,7 +554,7 @@ public class MappingElasticsearchConverter
 				return value;
 			}
 
-		if (getConversions().hasCustomReadTarget(value.getClass(), target)) {
+			if (conversions.hasCustomReadTarget(value.getClass(), target)) {
 				return conversionService.convert(value, target);
 			}
 
@@ -545,13 +587,111 @@ public class MappingElasticsearchConverter
 				}
 			}
 		}
-	// endregion
 
-	// region write
+		/**
+		 * Compute the type to use by checking the given entity against the store type;
+		 */
+		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);
+		}
+
+		class ElasticsearchPropertyValueProvider implements PropertyValueProvider<ElasticsearchPersistentProperty> {
+
+			final MapValueAccessor accessor;
+			final SpELExpressionEvaluator evaluator;
+
+			ElasticsearchPropertyValueProvider(MapValueAccessor accessor, SpELExpressionEvaluator evaluator) {
+				this.accessor = accessor;
+				this.evaluator = evaluator;
+			}
+
 			@Override
-	public void write(Object source, Document sink) {
+			public <T> T getPropertyValue(ElasticsearchPersistentProperty property) {
 
-		Assert.notNull(source, "source to map must not be null");
+				String expression = property.getSpelExpression();
+				Object value = expression != null ? evaluator.evaluate(expression) : accessor.get(property);
+
+				if (value == null) {
+					return null;
+				}
+
+				return readValue(value, property, property.getTypeInformation());
+			}
+		}
+
+		/**
+		 * Extension of {@link SpELExpressionParameterValueProvider} to recursively trigger value conversion on the raw
+		 * resolved SpEL value.
+		 *
+		 * @author Mark Paluch
+		 */
+		private class ConverterAwareSpELExpressionParameterValueProvider
+				extends SpELExpressionParameterValueProvider<ElasticsearchPersistentProperty> {
+
+			/**
+			 * Creates a new {@link ConverterAwareSpELExpressionParameterValueProvider}.
+			 *
+			 * @param evaluator must not be {@literal null}.
+			 * @param conversionService must not be {@literal null}.
+			 * @param delegate must not be {@literal null}.
+			 */
+			public ConverterAwareSpELExpressionParameterValueProvider(SpELExpressionEvaluator evaluator,
+					ConversionService conversionService, ParameterValueProvider<ElasticsearchPersistentProperty> delegate) {
+
+				super(evaluator, conversionService, delegate);
+			}
+
+			/*
+			 * (non-Javadoc)
+			 * @see org.springframework.data.mapping.model.SpELExpressionParameterValueProvider#potentiallyConvertSpelValue(java.lang.Object, org.springframework.data.mapping.PreferredConstructor.Parameter)
+			 */
+			@Override
+			protected <T> T potentiallyConvertSpelValue(Object object,
+					PreferredConstructor.Parameter<T, ElasticsearchPersistentProperty> parameter) {
+				return readValue(object, parameter.getType());
+			}
+		}
+
+		enum NoOpParameterValueProvider implements ParameterValueProvider<ElasticsearchPersistentProperty> {
+
+			INSTANCE;
+
+			@Override
+			public <T> T getParameterValue(PreferredConstructor.Parameter<T, ElasticsearchPersistentProperty> parameter) {
+				return null;
+			}
+		}
+	}
+
+	/**
+	 * Class to do the actual writing. The methods originally were in the MappingElasticsearchConverter class, but are
+	 * refactored to allow for keeping state during the conversion of an object.
+	 */
+	static private class Writer extends Base {
+
+		private boolean writeTypeHints = true;
+
+		public Writer(
+				MappingContext<? extends ElasticsearchPersistentEntity<?>, ElasticsearchPersistentProperty> mappingContext,
+				GenericConversionService conversionService, CustomConversions conversions) {
+			super(mappingContext, conversionService, conversions);
+		}
+
+		void write(Object source, Document sink) {
 
 			if (source instanceof Map) {
 				// noinspection unchecked
@@ -560,9 +700,15 @@ public class MappingElasticsearchConverter
 			}
 
 			Class<?> entityType = ClassUtils.getUserClass(source.getClass());
+			ElasticsearchPersistentEntity<?> entity = mappingContext.getPersistentEntity(entityType);
+
+			if (entity != null) {
+				writeTypeHints = entity.writeTypeHints();
+			}
+
 			TypeInformation<?> typeInformation = ClassTypeInformation.from(entityType);
 
-		if (requiresTypeHint(entityType)) {
+			if (writeTypeHints && requiresTypeHint(entityType)) {
 				typeMapper.writeType(typeInformation, sink);
 			}
 
@@ -577,7 +723,7 @@ public class MappingElasticsearchConverter
 		 * @param typeInformation type information for the source
 		 */
 		@SuppressWarnings("unchecked")
-	protected void writeInternal(@Nullable Object source, Map<String, Object> sink,
+		private void writeInternal(@Nullable Object source, Map<String, Object> sink,
 				@Nullable TypeInformation<?> typeInformation) {
 
 			if (null == source) {
@@ -618,7 +764,7 @@ public class MappingElasticsearchConverter
 		 * @param sink the write destination
 		 * @param entity entity for the source
 		 */
-	protected void writeInternal(@Nullable Object source, Map<String, Object> sink,
+		private void writeInternal(@Nullable Object source, Map<String, Object> sink,
 				@Nullable ElasticsearchPersistentEntity<?> entity) {
 
 			if (source == null) {
@@ -633,138 +779,24 @@ public class MappingElasticsearchConverter
 			writeProperties(entity, accessor, new MapValueAccessor(sink));
 		}
 
-	protected void writeProperties(ElasticsearchPersistentEntity<?> entity, PersistentPropertyAccessor<?> accessor,
-			MapValueAccessor sink) {
-
-		for (ElasticsearchPersistentProperty property : entity) {
-
-			if (!property.isWritable()) {
-				continue;
-			}
-
-			Object value = accessor.getProperty(property);
-
-			if (value == null) {
-
-				if (property.storeNullValue()) {
-					sink.set(property, null);
-				}
-
-				continue;
-			}
-
-			if (property.hasPropertyConverter()) {
-				value = propertyConverterWrite(property, value);
-				sink.set(property, value);
-			} else if (TemporalAccessor.class.isAssignableFrom(property.getActualType())
-					&& !getConversions().hasCustomWriteTarget(value.getClass())) {
-
-				// log at most 5 times
-				String propertyName = entity.getType().getSimpleName() + '.' + property.getName();
-				String key = propertyName + "-write";
-				int count = propertyWarnings.computeIfAbsent(key, k -> 0);
-				if (count < 5) {
-					LOGGER.warn(
-							"Type {} of property {} is a TemporalAccessor class but has neither a @Field annotation defining the date type nor a registered converter for writing!"
-									+ " It will be mapped to a complex object in Elasticsearch!",
-							property.getType().getSimpleName(), propertyName);
-					propertyWarnings.put(key, count + 1);
-				}
-			} else if (!isSimpleType(value)) {
-				writeProperty(property, value, sink);
-			} else {
-				Object writeSimpleValue = getPotentiallyConvertedSimpleWrite(value, Object.class);
-				if (writeSimpleValue != null) {
-					sink.set(property, writeSimpleValue);
-				}
-			}
-		}
-	}
-
-	private Object propertyConverterWrite(ElasticsearchPersistentProperty property, Object value) {
-		ElasticsearchPersistentPropertyConverter propertyConverter = Objects
-				.requireNonNull(property.getPropertyConverter());
-
-		if (value instanceof List) {
-			value = ((List<?>) value).stream().map(propertyConverter::write).collect(Collectors.toList());
-		} else if (value instanceof Set) {
-			value = ((Set<?>) value).stream().map(propertyConverter::write).collect(Collectors.toSet());
-		} else {
-			value = propertyConverter.write(value);
-		}
-		return value;
-	}
-
-	@SuppressWarnings("unchecked")
-	protected void writeProperty(ElasticsearchPersistentProperty property, Object value, MapValueAccessor sink) {
-
-		Optional<Class<?>> customWriteTarget = getConversions().getCustomWriteTarget(value.getClass());
-
-		if (customWriteTarget.isPresent()) {
-			Class<?> writeTarget = customWriteTarget.get();
-			sink.set(property, conversionService.convert(value, writeTarget));
-			return;
-		}
-
-		TypeInformation<?> valueType = ClassTypeInformation.from(value.getClass());
-		TypeInformation<?> type = property.getTypeInformation();
-
-		if (valueType.isCollectionLike()) {
-			List<Object> collectionInternal = createCollection(asCollection(value), property);
-			sink.set(property, collectionInternal);
-			return;
-		}
-
-		if (valueType.isMap()) {
-			Map<String, Object> mapDbObj = createMap((Map<?, ?>) value, property);
-			sink.set(property, mapDbObj);
-			return;
-		}
-
-		// Lookup potential custom target type
-		Optional<Class<?>> basicTargetType = conversions.getCustomWriteTarget(value.getClass());
-
-		if (basicTargetType.isPresent()) {
-
-			sink.set(property, conversionService.convert(value, basicTargetType.get()));
-			return;
-		}
-
-		ElasticsearchPersistentEntity<?> entity = valueType.isSubTypeOf(property.getType())
-				? mappingContext.getRequiredPersistentEntity(value.getClass())
-				: mappingContext.getRequiredPersistentEntity(type);
-
-		Object existingValue = sink.get(property);
-		Map<String, Object> document = existingValue instanceof Map ? (Map<String, Object>) existingValue
-				: Document.create();
-
-		addCustomTypeKeyIfNecessary(value, document, ClassTypeInformation.from(property.getRawType()));
-		writeInternal(value, document, entity);
-		sink.set(property, document);
-	}
-
 		/**
-	 * Writes the given {@link Collection} using the given {@link ElasticsearchPersistentProperty} information.
+		 * Check if a given type requires a type hint (aka {@literal _class} attribute) when writing to the document.
 		 *
-	 * @param collection must not be {@literal null}.
+		 * @param type must not be {@literal null}.
-	 * @param property must not be {@literal null}.
+		 * @return {@literal true} if not a simple type, {@link Collection} or type with custom write target.
 		 */
-	protected List<Object> createCollection(Collection<?> collection, ElasticsearchPersistentProperty property) {
+		private boolean requiresTypeHint(Class<?> type) {
-		return writeCollectionInternal(collection, property.getTypeInformation(), new ArrayList<>(collection.size()));
+
+			return !isSimpleType(type) && !ClassUtils.isAssignable(Collection.class, type)
+					&& !conversions.hasCustomWriteTarget(type, Document.class);
 		}
 
-	/**
+		private boolean isSimpleType(Object value) {
-	 * Writes the given {@link Map} using the given {@link ElasticsearchPersistentProperty} information.
+			return isSimpleType(value.getClass());
-	 *
+		}
-	 * @param map must not {@literal null}.
-	 * @param property must not be {@literal null}.
-	 */
-	protected Map<String, Object> createMap(Map<?, ?> map, ElasticsearchPersistentProperty property) {
 
-		Assert.notNull(map, "Given map must not be null!");
+		private boolean isSimpleType(Class<?> type) {
-		Assert.notNull(property, "PersistentProperty must not be null!");
+			return !Map.class.isAssignableFrom(type) && conversions.isSimpleType(type);
-
-		return writeMapInternal(map, new LinkedHashMap<>(map.size()), property.getTypeInformation());
 		}
 
 		/**
@@ -774,7 +806,7 @@ public class MappingElasticsearchConverter
 		 * @param sink must not be {@literal null}.
 		 * @param propertyType must not be {@literal null}.
 		 */
-	protected Map<String, Object> writeMapInternal(Map<?, ?> source, Map<String, Object> sink,
+		private Map<String, Object> writeMapInternal(Map<?, ?> source, Map<String, Object> sink,
 				TypeInformation<?> propertyType) {
 
 			for (Map.Entry<?, ?> entry : source.entrySet()) {
@@ -844,6 +876,134 @@ public class MappingElasticsearchConverter
 			return collection;
 		}
 
+		private void writeProperties(ElasticsearchPersistentEntity<?> entity, PersistentPropertyAccessor<?> accessor,
+				MapValueAccessor sink) {
+
+			for (ElasticsearchPersistentProperty property : entity) {
+
+				if (!property.isWritable()) {
+					continue;
+				}
+
+				Object value = accessor.getProperty(property);
+
+				if (value == null) {
+
+					if (property.storeNullValue()) {
+						sink.set(property, null);
+					}
+
+					continue;
+				}
+
+				if (property.hasPropertyConverter()) {
+					value = propertyConverterWrite(property, value);
+					sink.set(property, value);
+				} else if (TemporalAccessor.class.isAssignableFrom(property.getActualType())
+						&& !conversions.hasCustomWriteTarget(value.getClass())) {
+
+					// log at most 5 times
+					String propertyName = entity.getType().getSimpleName() + '.' + property.getName();
+					String key = propertyName + "-write";
+					int count = propertyWarnings.computeIfAbsent(key, k -> 0);
+					if (count < 5) {
+						LOGGER.warn(
+								"Type {} of property {} is a TemporalAccessor class but has neither a @Field annotation defining the date type nor a registered converter for writing!"
+										+ " It will be mapped to a complex object in Elasticsearch!",
+								property.getType().getSimpleName(), propertyName);
+						propertyWarnings.put(key, count + 1);
+					}
+				} else if (!isSimpleType(value)) {
+					writeProperty(property, value, sink);
+				} else {
+					Object writeSimpleValue = getPotentiallyConvertedSimpleWrite(value, Object.class);
+					if (writeSimpleValue != null) {
+						sink.set(property, writeSimpleValue);
+					}
+				}
+			}
+		}
+
+		@SuppressWarnings("unchecked")
+		protected void writeProperty(ElasticsearchPersistentProperty property, Object value, MapValueAccessor sink) {
+
+			Optional<Class<?>> customWriteTarget = conversions.getCustomWriteTarget(value.getClass());
+
+			if (customWriteTarget.isPresent()) {
+				Class<?> writeTarget = customWriteTarget.get();
+				sink.set(property, conversionService.convert(value, writeTarget));
+				return;
+			}
+
+			TypeInformation<?> valueType = ClassTypeInformation.from(value.getClass());
+			TypeInformation<?> type = property.getTypeInformation();
+
+			if (valueType.isCollectionLike()) {
+				List<Object> collectionInternal = createCollection(asCollection(value), property);
+				sink.set(property, collectionInternal);
+				return;
+			}
+
+			if (valueType.isMap()) {
+				Map<String, Object> mapDbObj = createMap((Map<?, ?>) value, property);
+				sink.set(property, mapDbObj);
+				return;
+			}
+
+			// Lookup potential custom target type
+			Optional<Class<?>> basicTargetType = conversions.getCustomWriteTarget(value.getClass());
+
+			if (basicTargetType.isPresent()) {
+
+				sink.set(property, conversionService.convert(value, basicTargetType.get()));
+				return;
+			}
+
+			ElasticsearchPersistentEntity<?> entity = valueType.isSubTypeOf(property.getType())
+					? mappingContext.getRequiredPersistentEntity(value.getClass())
+					: mappingContext.getRequiredPersistentEntity(type);
+
+			Object existingValue = sink.get(property);
+			Map<String, Object> document = existingValue instanceof Map ? (Map<String, Object>) existingValue
+					: Document.create();
+
+			addCustomTypeKeyIfNecessary(value, document, ClassTypeInformation.from(property.getRawType()));
+			writeInternal(value, document, entity);
+			sink.set(property, document);
+		}
+
+		/**
+		 * Adds custom typeInformation information to the given {@link Map} if necessary. That is if the value is not the
+		 * same as the one given. This is usually the case if you store a subtype of the actual declared typeInformation of
+		 * the property.
+		 *
+		 * @param source must not be {@literal null}.
+		 * @param sink must not be {@literal null}.
+		 * @param type type to compare to
+		 */
+		private void addCustomTypeKeyIfNecessary(Object source, Map<String, Object> sink,
+				@Nullable TypeInformation<?> type) {
+
+			if (!writeTypeHints) {
+				return;
+			}
+
+			Class<?> reference;
+
+			if (type == null) {
+				reference = Object.class;
+			} else {
+				TypeInformation<?> actualType = type.getActualType();
+				reference = actualType == null ? Object.class : actualType.getType();
+			}
+			Class<?> valueType = ClassUtils.getUserClass(source.getClass());
+
+			boolean notTheSameClass = !valueType.equals(reference);
+			if (notTheSameClass) {
+				typeMapper.writeType(valueType, sink);
+			}
+		}
+
 		/**
 		 * Returns a {@link String} representation of the given {@link Map} key
 		 *
@@ -868,7 +1028,8 @@ public class MappingElasticsearchConverter
 
 		/**
 		 * Checks whether we have a custom conversion registered for the given value into an arbitrary simple Elasticsearch
-	 * type. Returns the converted value if so. If not, we perform special enum handling or simply return the value as is.
+		 * type. Returns the converted value if so. If not, we perform special enum handling or simply return the value as
+		 * is.
 		 *
 		 * @param value value to convert
 		 */
@@ -907,13 +1068,51 @@ public class MappingElasticsearchConverter
 			return Enum.class.isAssignableFrom(value.getClass()) ? ((Enum<?>) value).name() : value;
 		}
 
+		private Object propertyConverterWrite(ElasticsearchPersistentProperty property, Object value) {
+			ElasticsearchPersistentPropertyConverter propertyConverter = Objects
+					.requireNonNull(property.getPropertyConverter());
+
+			if (value instanceof List) {
+				value = ((List<?>) value).stream().map(propertyConverter::write).collect(Collectors.toList());
+			} else if (value instanceof Set) {
+				value = ((Set<?>) value).stream().map(propertyConverter::write).collect(Collectors.toSet());
+			} else {
+				value = propertyConverter.write(value);
+			}
+			return value;
+		}
+
+		/**
+		 * Writes the given {@link Collection} using the given {@link ElasticsearchPersistentProperty} information.
+		 *
+		 * @param collection must not be {@literal null}.
+		 * @param property must not be {@literal null}.
+		 */
+		protected List<Object> createCollection(Collection<?> collection, ElasticsearchPersistentProperty property) {
+			return writeCollectionInternal(collection, property.getTypeInformation(), new ArrayList<>(collection.size()));
+		}
+
+		/**
+		 * Writes the given {@link Map} using the given {@link ElasticsearchPersistentProperty} information.
+		 *
+		 * @param map must not {@literal null}.
+		 * @param property must not be {@literal null}.
+		 */
+		protected Map<String, Object> createMap(Map<?, ?> map, ElasticsearchPersistentProperty property) {
+
+			Assert.notNull(map, "Given map must not be null!");
+			Assert.notNull(property, "PersistentProperty must not be null!");
+
+			return writeMapInternal(map, new LinkedHashMap<>(map.size()), property.getTypeInformation());
+		}
+
 		/**
 		 * @deprecated since 4.2, use {@link #getPotentiallyConvertedSimpleWrite(Object, Class)} instead.
 		 */
 		@Nullable
 		@Deprecated
 		protected Object getWriteSimpleValue(Object value) {
-		Optional<Class<?>> customTarget = getConversions().getCustomWriteTarget(value.getClass());
+			Optional<Class<?>> customTarget = conversions.getCustomWriteTarget(value.getClass());
 
 			if (customTarget.isPresent()) {
 				return conversionService.convert(value, customTarget.get());
@@ -935,79 +1134,6 @@ public class MappingElasticsearchConverter
 			return document;
 		}
 
-	// endregion
-
-	// region helper methods
-
-	/**
-	 * Adds custom typeInformation information to the given {@link Map} if necessary. That is if the value is not the same
-	 * as the one given. This is usually the case if you store a subtype of the actual declared typeInformation of the
-	 * property.
-	 *
-	 * @param source must not be {@literal null}.
-	 * @param sink must not be {@literal null}.
-	 * @param type type to compare to
-	 */
-	protected void addCustomTypeKeyIfNecessary(Object source, Map<String, Object> sink,
-			@Nullable TypeInformation<?> type) {
-
-		Class<?> reference;
-
-		if (type == null) {
-			reference = Object.class;
-		} else {
-			TypeInformation<?> actualType = type.getActualType();
-			reference = actualType == null ? Object.class : actualType.getType();
-		}
-		Class<?> valueType = ClassUtils.getUserClass(source.getClass());
-
-		boolean notTheSameClass = !valueType.equals(reference);
-		if (notTheSameClass) {
-			typeMapper.writeType(valueType, sink);
-		}
-	}
-
-	/**
-	 * Check if a given type requires a type hint (aka {@literal _class} attribute) when writing to the document.
-	 *
-	 * @param type must not be {@literal null}.
-	 * @return {@literal true} if not a simple type, {@link Collection} or type with custom write target.
-	 */
-	public boolean requiresTypeHint(Class<?> type) {
-
-		return !isSimpleType(type) && !ClassUtils.isAssignable(Collection.class, type)
-				&& !conversions.hasCustomWriteTarget(type, Document.class);
-	}
-
-	/**
-	 * Compute the type to use by checking the given entity against the store type;
-	 */
-	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 boolean isSimpleType(Object value) {
-		return isSimpleType(value.getClass());
-	}
-
-	private boolean isSimpleType(Class<?> type) {
-		return !Map.class.isAssignableFrom(type) && getConversions().isSimpleType(type);
-	}
-
 		/**
 		 * Returns given object as {@link Collection}. Will return the {@link Collection} as is if the source is a
 		 * {@link Collection} already, will convert an array into a {@link Collection} or simply create a single element
@@ -1023,6 +1149,7 @@ public class MappingElasticsearchConverter
 
 			return source.getClass().isArray() ? CollectionUtils.arrayToList(source) : Collections.singleton(source);
 		}
+	}
 	// endregion
 
 	// region queries
@@ -1260,71 +1387,4 @@ public class MappingElasticsearchConverter
 		}
 	}
 
-	class ElasticsearchPropertyValueProvider implements PropertyValueProvider<ElasticsearchPersistentProperty> {
-
-		final MapValueAccessor accessor;
-		final SpELExpressionEvaluator evaluator;
-
-		ElasticsearchPropertyValueProvider(MapValueAccessor accessor, SpELExpressionEvaluator evaluator) {
-			this.accessor = accessor;
-			this.evaluator = evaluator;
-		}
-
-		@Override
-		public <T> T getPropertyValue(ElasticsearchPersistentProperty property) {
-
-			String expression = property.getSpelExpression();
-			Object value = expression != null ? evaluator.evaluate(expression) : accessor.get(property);
-
-			if (value == null) {
-				return null;
-			}
-
-			return readValue(value, property, property.getTypeInformation());
-		}
-	}
-
-	/**
-	 * Extension of {@link SpELExpressionParameterValueProvider} to recursively trigger value conversion on the raw
-	 * resolved SpEL value.
-	 *
-	 * @author Mark Paluch
-	 */
-	private class ConverterAwareSpELExpressionParameterValueProvider
-			extends SpELExpressionParameterValueProvider<ElasticsearchPersistentProperty> {
-
-		/**
-		 * Creates a new {@link ConverterAwareSpELExpressionParameterValueProvider}.
-		 *
-		 * @param evaluator must not be {@literal null}.
-		 * @param conversionService must not be {@literal null}.
-		 * @param delegate must not be {@literal null}.
-		 */
-		public ConverterAwareSpELExpressionParameterValueProvider(SpELExpressionEvaluator evaluator,
-				ConversionService conversionService, ParameterValueProvider<ElasticsearchPersistentProperty> delegate) {
-
-			super(evaluator, conversionService, delegate);
-		}
-
-		/*
-		 * (non-Javadoc)
-		 * @see org.springframework.data.mapping.model.SpELExpressionParameterValueProvider#potentiallyConvertSpelValue(java.lang.Object, org.springframework.data.mapping.PreferredConstructor.Parameter)
-		 */
-		@Override
-		protected <T> T potentiallyConvertSpelValue(Object object,
-				PreferredConstructor.Parameter<T, ElasticsearchPersistentProperty> parameter) {
-			return readValue(object, parameter.getType());
-		}
-	}
-
-	enum NoOpParameterValueProvider implements ParameterValueProvider<ElasticsearchPersistentProperty> {
-
-		INSTANCE;
-
-		@Override
-		public <T> T getParameterValue(PreferredConstructor.Parameter<T, ElasticsearchPersistentProperty> parameter) {
-			return null;
-		}
-	}
-
 }

src/main/java/org/springframework/data/elasticsearch/core/index/MappingBuilder.java

@@ -95,6 +95,8 @@ public class MappingBuilder {
 
 	private final ElasticsearchConverter elasticsearchConverter;
 
+	private boolean writeTypeHints = true;
+
 	public MappingBuilder(ElasticsearchConverter elasticsearchConverter) {
 		this.elasticsearchConverter = elasticsearchConverter;
 	}
@@ -111,6 +113,8 @@ public class MappingBuilder {
 			ElasticsearchPersistentEntity<?> entity = elasticsearchConverter.getMappingContext()
 					.getRequiredPersistentEntity(clazz);
 
+			writeTypeHints = entity.writeTypeHints();
+
 			XContentBuilder builder = jsonBuilder().startObject();
 
 			// Dynamic templates
@@ -128,12 +132,15 @@ public class MappingBuilder {
 	}
 
 	private void writeTypeHintMapping(XContentBuilder builder) throws IOException {
+
+		if (writeTypeHints) {
 			builder.startObject(TYPEHINT_PROPERTY) //
 					.field(FIELD_PARAM_TYPE, TYPE_VALUE_KEYWORD) //
 					.field(FIELD_PARAM_INDEX, false) //
 					.field(FIELD_PARAM_DOC_VALUES, false) //
 					.endObject();
 		}
+	}
 
 	private void mapEntity(XContentBuilder builder, @Nullable ElasticsearchPersistentEntity<?> entity,
 			boolean isRootObject, String nestedObjectFieldName, boolean nestedOrObjectField, FieldType fieldType,

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

@@ -17,11 +17,11 @@ package org.springframework.data.elasticsearch.core.mapping;
 
 import org.elasticsearch.index.VersionType;
 import org.springframework.data.elasticsearch.annotations.Field;
-import org.springframework.data.elasticsearch.core.document.Document;
 import org.springframework.data.elasticsearch.core.index.Settings;
 import org.springframework.data.elasticsearch.core.join.JoinField;
 import org.springframework.data.elasticsearch.core.query.SeqNoPrimaryTerm;
 import org.springframework.data.mapping.PersistentEntity;
+import org.springframework.data.mapping.model.FieldNamingStrategy;
 import org.springframework.lang.Nullable;
 
 /**
@@ -148,4 +148,16 @@ public interface ElasticsearchPersistentEntity<T> extends PersistentEntity<T, El
 	 */
 	@Nullable
 	String resolveRouting(T bean);
+
+	/**
+	 * @return the {@link FieldNamingStrategy} for the entity
+	 * @since 4.3
+	 */
+	FieldNamingStrategy getFieldNamingStrategy();
+
+	/**
+	 * @return true if type hints on this entity should be written.
+	 * @since 4.3
+	 */
+	boolean writeTypeHints();
 }

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

@@ -16,6 +16,7 @@
 package org.springframework.data.elasticsearch.core.mapping;
 
 import org.springframework.core.convert.converter.Converter;
+import org.springframework.data.elasticsearch.core.query.SeqNoPrimaryTerm;
 import org.springframework.data.mapping.PersistentProperty;
 import org.springframework.lang.Nullable;
 

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

@@ -37,6 +37,7 @@ public class SimpleElasticsearchMappingContext
 	private static final FieldNamingStrategy DEFAULT_NAMING_STRATEGY = PropertyNameFieldNamingStrategy.INSTANCE;
 
 	private FieldNamingStrategy fieldNamingStrategy = DEFAULT_NAMING_STRATEGY;
+	private boolean writeTypeHints = true;
 
 	/**
 	 * Configures the {@link FieldNamingStrategy} to be used to determine the field name if no manual mapping is applied.
@@ -50,6 +51,15 @@ public class SimpleElasticsearchMappingContext
 		this.fieldNamingStrategy = fieldNamingStrategy == null ? DEFAULT_NAMING_STRATEGY : fieldNamingStrategy;
 	}
 
+	/**
+	 * Sets the flag if type hints should be written in Entities created by this instance.
+	 *
+	 * @since 4.3
+	 */
+	public void setWriteTypeHints(boolean writeTypeHints) {
+		this.writeTypeHints = writeTypeHints;
+	}
+
 	@Override
 	protected boolean shouldCreatePersistentEntityFor(TypeInformation<?> type) {
 		return !ElasticsearchSimpleTypes.HOLDER.isSimpleType(type.getType());
@@ -57,12 +67,13 @@ public class SimpleElasticsearchMappingContext
 
 	@Override
 	protected <T> SimpleElasticsearchPersistentEntity<?> createPersistentEntity(TypeInformation<T> typeInformation) {
-		return new SimpleElasticsearchPersistentEntity<>(typeInformation);
+		return new SimpleElasticsearchPersistentEntity<>(typeInformation,
+				new SimpleElasticsearchPersistentEntity.ContextConfiguration(fieldNamingStrategy, writeTypeHints));
 	}
 
 	@Override
 	protected ElasticsearchPersistentProperty createPersistentProperty(Property property,
 			SimpleElasticsearchPersistentEntity<?> owner, SimpleTypeHolder simpleTypeHolder) {
-		return new SimpleElasticsearchPersistentProperty(property, owner, simpleTypeHolder, fieldNamingStrategy);
+		return new SimpleElasticsearchPersistentProperty(property, owner, simpleTypeHolder);
 	}
 }

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

@@ -34,6 +34,7 @@ import org.springframework.data.elasticsearch.core.join.JoinField;
 import org.springframework.data.mapping.MappingException;
 import org.springframework.data.mapping.PropertyHandler;
 import org.springframework.data.mapping.model.BasicPersistentEntity;
+import org.springframework.data.mapping.model.FieldNamingStrategy;
 import org.springframework.data.mapping.model.PersistentPropertyAccessorFactory;
 import org.springframework.data.spel.ExpressionDependencies;
 import org.springframework.data.util.Lazy;
@@ -66,10 +67,9 @@ public class SimpleElasticsearchPersistentEntity<T> extends BasicPersistentEntit
 	private static final Logger LOGGER = LoggerFactory.getLogger(SimpleElasticsearchPersistentEntity.class);
 	private static final SpelExpressionParser PARSER = new SpelExpressionParser();
 
+	private @Nullable final Document document;
 	private @Nullable String indexName;
 	private final Lazy<SettingsParameter> settingsParameter;
-	@Deprecated private @Nullable String parentType;
-	@Deprecated private @Nullable ElasticsearchPersistentProperty parentIdProperty;
 	private @Nullable ElasticsearchPersistentProperty seqNoPrimaryTermProperty;
 	private @Nullable ElasticsearchPersistentProperty joinFieldProperty;
 	private @Nullable VersionType versionType;
@@ -77,18 +77,21 @@ public class SimpleElasticsearchPersistentEntity<T> extends BasicPersistentEntit
 	private final Map<String, ElasticsearchPersistentProperty> fieldNamePropertyCache = new ConcurrentHashMap<>();
 	private final ConcurrentHashMap<String, Expression> routingExpressions = new ConcurrentHashMap<>();
 	private @Nullable String routing;
+	private final ContextConfiguration contextConfiguration;
 
 	private final ConcurrentHashMap<String, Expression> indexNameExpressions = new ConcurrentHashMap<>();
 	private final Lazy<EvaluationContext> indexNameEvaluationContext = Lazy.of(this::getIndexNameEvaluationContext);
 
-	public SimpleElasticsearchPersistentEntity(TypeInformation<T> typeInformation) {
+	public SimpleElasticsearchPersistentEntity(TypeInformation<T> typeInformation,
+			ContextConfiguration contextConfiguration) {
 
 		super(typeInformation);
+		this.contextConfiguration = contextConfiguration;
 
 		Class<T> clazz = typeInformation.getType();
 
-		org.springframework.data.elasticsearch.annotations.Document document = AnnotatedElementUtils
+		document = AnnotatedElementUtils.findMergedAnnotation(clazz,
-				.findMergedAnnotation(clazz, org.springframework.data.elasticsearch.annotations.Document.class);
+				org.springframework.data.elasticsearch.annotations.Document.class);
 
 		// need a Lazy here, because we need the persistent properties available
 		this.settingsParameter = Lazy.of(() -> buildSettingsParameter(clazz));
@@ -159,7 +162,31 @@ public class SimpleElasticsearchPersistentEntity<T> extends BasicPersistentEntit
 		return createIndexAndMapping;
 	}
 
-	// endregion
+	@Override
+	public FieldNamingStrategy getFieldNamingStrategy() {
+		return contextConfiguration.getFieldNamingStrategy();
+	}
+
+	@Override
+	public boolean writeTypeHints() {
+
+		boolean writeTypeHints = contextConfiguration.writeTypeHints;
+
+		if (document != null) {
+			switch (document.writeTypeHint()) {
+				case TRUE:
+					writeTypeHints = true;
+					break;
+				case FALSE:
+					writeTypeHints = false;
+					break;
+				case DEFAULT:
+					break;
+			}
+		}
+
+		return writeTypeHints;
+	}
 
 	@Override
 	public void addPersistentProperty(ElasticsearchPersistentProperty property) {
@@ -215,6 +242,7 @@ public class SimpleElasticsearchPersistentEntity<T> extends BasicPersistentEntit
 	 * (non-Javadoc)
 	 * @see org.springframework.data.mapping.model.BasicPersistentEntity#setPersistentPropertyAccessorFactory(org.springframework.data.mapping.model.PersistentPropertyAccessorFactory)
 	 */
+	@SuppressWarnings("SpellCheckingInspection")
 	@Override
 	public void setPersistentPropertyAccessorFactory(PersistentPropertyAccessorFactory factory) {
 
@@ -327,6 +355,7 @@ public class SimpleElasticsearchPersistentEntity<T> extends BasicPersistentEntit
 		ExpressionDependencies expressionDependencies = expression != null ? ExpressionDependencies.discover(expression)
 				: ExpressionDependencies.none();
 
+		// noinspection ConstantConditions
 		return getEvaluationContext(null, expressionDependencies);
 	}
 
@@ -350,6 +379,7 @@ public class SimpleElasticsearchPersistentEntity<T> extends BasicPersistentEntit
 			Expression expression = routingExpressions.computeIfAbsent(routing, PARSER::parseExpression);
 			ExpressionDependencies expressionDependencies = ExpressionDependencies.discover(expression);
 
+			// noinspection ConstantConditions
 			EvaluationContext context = getEvaluationContext(null, expressionDependencies);
 			context.setVariable("entity", bean);
 
@@ -525,4 +555,22 @@ public class SimpleElasticsearchPersistentEntity<T> extends BasicPersistentEntit
 	}
 
 	// endregion
+
+	/**
+	 * Configuration settings passed in from the creating {@link SimpleElasticsearchMappingContext}.
+	 */
+	static class ContextConfiguration {
+
+		private final FieldNamingStrategy fieldNamingStrategy;
+		private final boolean writeTypeHints;
+
+		ContextConfiguration(FieldNamingStrategy fieldNamingStrategy, boolean writeTypeHints) {
+			this.fieldNamingStrategy = fieldNamingStrategy;
+			this.writeTypeHints = writeTypeHints;
+		}
+
+		public FieldNamingStrategy getFieldNamingStrategy() {
+			return fieldNamingStrategy;
+		}
+	}
 }

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

@@ -64,23 +64,20 @@ public class SimpleElasticsearchPersistentProperty extends
 	private static final Logger LOGGER = LoggerFactory.getLogger(SimpleElasticsearchPersistentProperty.class);
 
 	private static final List<String> SUPPORTED_ID_PROPERTY_NAMES = Arrays.asList("id", "document");
+	private static final PropertyNameFieldNamingStrategy DEFAULT_FIELD_NAMING_STRATEGY = PropertyNameFieldNamingStrategy.INSTANCE;
 
 	private final boolean isId;
 	private final boolean isSeqNoPrimaryTerm;
 	private final @Nullable String annotatedFieldName;
 	@Nullable private ElasticsearchPersistentPropertyConverter propertyConverter;
 	private final boolean storeNullValue;
-	private final FieldNamingStrategy fieldNamingStrategy;
 
 	public SimpleElasticsearchPersistentProperty(Property property,
-			PersistentEntity<?, ElasticsearchPersistentProperty> owner, SimpleTypeHolder simpleTypeHolder,
+			PersistentEntity<?, ElasticsearchPersistentProperty> owner, SimpleTypeHolder simpleTypeHolder) {
-			@Nullable FieldNamingStrategy fieldNamingStrategy) {
 
 		super(property, owner, simpleTypeHolder);
 
 		this.annotatedFieldName = getAnnotatedFieldName();
-		this.fieldNamingStrategy = fieldNamingStrategy == null ? PropertyNameFieldNamingStrategy.INSTANCE
-				: fieldNamingStrategy;
 		this.isId = super.isIdProperty()
 				|| (SUPPORTED_ID_PROPERTY_NAMES.contains(getFieldName()) && !hasExplicitFieldName());
 		this.isSeqNoPrimaryTerm = SeqNoPrimaryTerm.class.isAssignableFrom(getRawType());
@@ -248,6 +245,7 @@ public class SimpleElasticsearchPersistentProperty extends
 	public String getFieldName() {
 
 		if (annotatedFieldName == null) {
+			FieldNamingStrategy fieldNamingStrategy = getFieldNamingStrategy();
 			String fieldName = fieldNamingStrategy.getFieldName(this);
 
 			if (!StringUtils.hasText(fieldName)) {
@@ -261,6 +259,16 @@ public class SimpleElasticsearchPersistentProperty extends
 		return annotatedFieldName;
 	}
 
+	private FieldNamingStrategy getFieldNamingStrategy() {
+		PersistentEntity<?, ElasticsearchPersistentProperty> owner = getOwner();
+
+		if (owner instanceof ElasticsearchPersistentEntity) {
+			return ((ElasticsearchPersistentEntity<?>) owner).getFieldNamingStrategy();
+		}
+
+		return DEFAULT_FIELD_NAMING_STRATEGY;
+	}
+
 	@Override
 	public boolean isIdProperty() {
 		return isId;

src/test/java/org/springframework/data/elasticsearch/core/convert/MappingElasticsearchConverterUnitTests.java

@@ -26,7 +26,6 @@ import java.util.Collections;
 import java.util.HashMap;
 import java.util.LinkedHashMap;
 import java.util.List;
-import java.util.Locale;
 import java.util.Map;
 
 import org.json.JSONException;
@@ -1176,8 +1175,74 @@ public class MappingElasticsearchConverterUnitTests {
 		}
 	}
 
-	private String pointTemplate(String name, Point point) {
+	@Test // #1454
-		return String.format(Locale.ENGLISH, "\"%s\":{\"lat\":%.1f,\"lon\":%.1f}", name, point.getY(), point.getX());
+	@DisplayName("should write type hints if configured")
+	void shouldWriteTypeHintsIfConfigured() throws JSONException {
+
+		((SimpleElasticsearchMappingContext) mappingElasticsearchConverter.getMappingContext()).setWriteTypeHints(true);
+		PersonWithCars person = new PersonWithCars();
+		person.setId("42");
+		person.setName("Smith");
+		Car car1 = new Car();
+		car1.setModel("Ford Mustang");
+		Car car2 = new ElectricCar();
+		car2.setModel("Porsche Taycan");
+		person.setCars(Arrays.asList(car1, car2));
+
+		String expected = "{\n" + //
+				"  \"_class\": \"org.springframework.data.elasticsearch.core.convert.MappingElasticsearchConverterUnitTests$PersonWithCars\",\n"
+				+ "  \"id\": \"42\",\n" + //
+				"  \"name\": \"Smith\",\n" + //
+				"  \"cars\": [\n" + //
+				"    {\n" + //
+				"      \"model\": \"Ford Mustang\"\n" + //
+				"    },\n" + //
+				"    {\n" + //
+				"      \"_class\": \"org.springframework.data.elasticsearch.core.convert.MappingElasticsearchConverterUnitTests$ElectricCar\",\n"
+				+ "      \"model\": \"Porsche Taycan\"\n" + //
+				"    }\n" + //
+				"  ]\n" + //
+				"}\n"; //
+
+		Document document = Document.create();
+
+		mappingElasticsearchConverter.write(person, document);
+
+		assertEquals(expected, document.toJson(), true);
+	}
+
+	@Test // #1454
+	@DisplayName("should not write type hints if configured")
+	void shouldNotWriteTypeHintsIfNotConfigured() throws JSONException {
+
+		((SimpleElasticsearchMappingContext) mappingElasticsearchConverter.getMappingContext()).setWriteTypeHints(false);
+		PersonWithCars person = new PersonWithCars();
+		person.setId("42");
+		person.setName("Smith");
+		Car car1 = new Car();
+		car1.setModel("Ford Mustang");
+		Car car2 = new ElectricCar();
+		car2.setModel("Porsche Taycan");
+		person.setCars(Arrays.asList(car1, car2));
+
+		String expected = "{\n" + //
+				"  \"id\": \"42\",\n" + //
+				"  \"name\": \"Smith\",\n" + //
+				"  \"cars\": [\n" + //
+				"    {\n" + //
+				"      \"model\": \"Ford Mustang\"\n" + //
+				"    },\n" + //
+				"    {\n" + //
+				"      \"model\": \"Porsche Taycan\"\n" + //
+				"    }\n" + //
+				"  ]\n" + //
+				"}\n"; //
+
+		Document document = Document.create();
+
+		mappingElasticsearchConverter.write(person, document);
+
+		assertEquals(expected, document.toJson(), true);
 	}
 
 	private Map<String, Object> writeToMap(Object source) {
@@ -1187,6 +1252,7 @@ public class MappingElasticsearchConverterUnitTests {
 		return sink;
 	}
 
+	// region entities
 	public static class Sample {
 		@Nullable public @ReadOnlyProperty String readOnly;
 		@Nullable public @Transient String annotatedTransientProperty;
@@ -2008,4 +2074,39 @@ public class MappingElasticsearchConverterUnitTests {
 		}
 	}
 
+	private static class ElectricCar extends Car {}
+
+	private static class PersonWithCars {
+		@Id @Nullable String id;
+		@Field(type = FieldType.Text) @Nullable private String name;
+		@Field(type = FieldType.Nested) @Nullable private List<? extends Car> cars;
+
+		@Nullable
+		public String getId() {
+			return id;
+		}
+
+		public void setId(@Nullable String id) {
+			this.id = id;
+		}
+
+		@Nullable
+		public String getName() {
+			return name;
+		}
+
+		public void setName(@Nullable String name) {
+			this.name = name;
+		}
+
+		@Nullable
+		public List<? extends Car> getCars() {
+			return cars;
+		}
+
+		public void setCars(@Nullable List<Car> cars) {
+			this.cars = cars;
+		}
+	}
+	// endregion
 }

src/test/java/org/springframework/data/elasticsearch/core/index/MappingBuilderUnitTests.java

@@ -44,6 +44,7 @@ import org.springframework.data.elasticsearch.annotations.*;
 import org.springframework.data.elasticsearch.core.MappingContextBaseTests;
 import org.springframework.data.elasticsearch.core.completion.Completion;
 import org.springframework.data.elasticsearch.core.geo.GeoPoint;
+import org.springframework.data.elasticsearch.core.mapping.SimpleElasticsearchMappingContext;
 import org.springframework.data.elasticsearch.core.query.SeqNoPrimaryTerm;
 import org.springframework.data.geo.Box;
 import org.springframework.data.geo.Circle;
@@ -661,6 +662,119 @@ public class MappingBuilderUnitTests extends MappingContextBaseTests {
 		assertEquals(expected, mapping, false);
 	}
 
+	@Test // #1454
+	@DisplayName("should write type hints when context is configured to do so")
+	void shouldWriteTypeHintsWhenContextIsConfiguredToDoSo() throws JSONException {
+
+		((SimpleElasticsearchMappingContext) (elasticsearchConverter.get().getMappingContext())).setWriteTypeHints(true);
+		String expected = "{\n" + //
+				"  \"properties\": {\n" + //
+				"    \"_class\": {\n" + //
+				"      \"type\": \"keyword\",\n" + //
+				"      \"index\": false,\n" + //
+				"      \"doc_values\": false\n" + //
+				"    },\n" + //
+				"    \"title\": {\n" + //
+				"      \"type\": \"text\"\n" + //
+				"    },\n" + //
+				"    \"authors\": {\n" + //
+				"      \"type\": \"nested\",\n" + //
+				"      \"properties\": {\n" + //
+				"        \"_class\": {\n" + //
+				"          \"type\": \"keyword\",\n" + //
+				"          \"index\": false,\n" + //
+				"          \"doc_values\": false\n" + //
+				"        }\n" + //
+				"      }\n" + //
+				"    }\n" + //
+				"  }\n" + //
+				"}\n"; //
+
+		String mapping = getMappingBuilder().buildPropertyMapping(Magazine.class);
+
+		assertEquals(expected, mapping, true);
+	}
+
+	@Test // #1454
+	@DisplayName("should not write type hints when context is configured to not do so")
+	void shouldNotWriteTypeHintsWhenContextIsConfiguredToNotDoSo() throws JSONException {
+
+		((SimpleElasticsearchMappingContext) (elasticsearchConverter.get().getMappingContext())).setWriteTypeHints(false);
+		String expected = "{\n" + //
+				"  \"properties\": {\n" + //
+				"    \"title\": {\n" + //
+				"      \"type\": \"text\"\n" + //
+				"    },\n" + //
+				"    \"authors\": {\n" + //
+				"      \"type\": \"nested\",\n" + //
+				"      \"properties\": {\n" + //
+				"      }\n" + //
+				"    }\n" + //
+				"  }\n" + //
+				"}\n"; //
+
+		String mapping = getMappingBuilder().buildPropertyMapping(Magazine.class);
+
+		assertEquals(expected, mapping, true);
+	}
+
+	@Test // #1454
+	@DisplayName("should write type hints when context is configured to not do so but entity should")
+	void shouldWriteTypeHintsWhenContextIsConfiguredToNotDoSoButEntityShould() throws JSONException {
+
+		((SimpleElasticsearchMappingContext) (elasticsearchConverter.get().getMappingContext())).setWriteTypeHints(false);
+		String expected = "{\n" + //
+				"  \"properties\": {\n" + //
+				"    \"_class\": {\n" + //
+				"      \"type\": \"keyword\",\n" + //
+				"      \"index\": false,\n" + //
+				"      \"doc_values\": false\n" + //
+				"    },\n" + //
+				"    \"title\": {\n" + //
+				"      \"type\": \"text\"\n" + //
+				"    },\n" + //
+				"    \"authors\": {\n" + //
+				"      \"type\": \"nested\",\n" + //
+				"      \"properties\": {\n" + //
+				"        \"_class\": {\n" + //
+				"          \"type\": \"keyword\",\n" + //
+				"          \"index\": false,\n" + //
+				"          \"doc_values\": false\n" + //
+				"        }\n" + //
+				"      }\n" + //
+				"    }\n" + //
+				"  }\n" + //
+				"}\n"; //
+
+		String mapping = getMappingBuilder().buildPropertyMapping(MagazineWithTypeHints.class);
+
+		assertEquals(expected, mapping, true);
+	}
+
+	@Test // #1454
+	@DisplayName("should not write type hints when context is configured to do so but entity should not")
+	void shouldNotWriteTypeHintsWhenContextIsConfiguredToDoSoButEntityShouldNot() throws JSONException {
+
+		((SimpleElasticsearchMappingContext) (elasticsearchConverter.get().getMappingContext())).setWriteTypeHints(true);
+		String expected = "{\n" + //
+				"  \"properties\": {\n" + //
+				"    \"title\": {\n" + //
+				"      \"type\": \"text\"\n" + //
+				"    },\n" + //
+				"    \"authors\": {\n" + //
+				"      \"type\": \"nested\",\n" + //
+				"      \"properties\": {\n" + //
+				"      }\n" + //
+				"    }\n" + //
+				"  }\n" + //
+				"}\n"; //
+
+		String mapping = getMappingBuilder().buildPropertyMapping(MagazineWithoutTypeHints.class);
+
+		assertEquals(expected, mapping, true);
+	}
+
+	// region entities
 	@Document(indexName = "ignore-above-index")
 	static class IgnoreAboveEntity {
 		@Nullable @Id private String id;
@@ -1555,4 +1669,26 @@ public class MappingBuilderUnitTests extends MappingContextBaseTests {
 			this.field5 = field5;
 		}
 	}
+
+	@Document(indexName = "magazine")
+	private static class Magazine {
+		@Id @Nullable private String id;
+		@Field(type = Text) @Nullable private String title;
+		@Field(type = Nested) @Nullable private List<Author> authors;
+	}
+
+	@Document(indexName = "magazine-without-type-hints", writeTypeHint = WriteTypeHint.FALSE)
+	private static class MagazineWithoutTypeHints {
+		@Id @Nullable private String id;
+		@Field(type = Text) @Nullable private String title;
+		@Field(type = Nested) @Nullable private List<Author> authors;
+	}
+
+	@Document(indexName = "magazine-with-type-hints", writeTypeHint = WriteTypeHint.TRUE)
+	private static class MagazineWithTypeHints {
+		@Id @Nullable private String id;
+		@Field(type = Text) @Nullable private String title;
+		@Field(type = Nested) @Nullable private List<Author> authors;
+	}
+	// endregion
 }

src/test/java/org/springframework/data/elasticsearch/core/mapping/SimpleElasticsearchPersistentEntityTests.java

@@ -28,10 +28,14 @@ import org.springframework.data.elasticsearch.annotations.Document;
 import org.springframework.data.elasticsearch.annotations.Field;
 import org.springframework.data.elasticsearch.annotations.FieldType;
 import org.springframework.data.elasticsearch.annotations.Setting;
+import org.springframework.data.elasticsearch.annotations.WriteTypeHint;
 import org.springframework.data.elasticsearch.core.MappingContextBaseTests;
 import org.springframework.data.elasticsearch.core.query.SeqNoPrimaryTerm;
 import org.springframework.data.mapping.MappingException;
+import org.springframework.data.mapping.PersistentProperty;
+import org.springframework.data.mapping.model.FieldNamingStrategy;
 import org.springframework.data.mapping.model.Property;
+import org.springframework.data.mapping.model.PropertyNameFieldNamingStrategy;
 import org.springframework.data.mapping.model.SimpleTypeHolder;
 import org.springframework.data.util.ClassTypeInformation;
 import org.springframework.data.util.TypeInformation;
@@ -52,13 +56,16 @@ public class SimpleElasticsearchPersistentEntityTests extends MappingContextBase
 	@DisplayName("properties setup")
 	class PropertiesTests {
 
+		private final SimpleElasticsearchPersistentEntity.ContextConfiguration contextConfiguration = new SimpleElasticsearchPersistentEntity.ContextConfiguration(
+				PropertyNameFieldNamingStrategy.INSTANCE, true);
+
 		@Test
 		public void shouldThrowExceptionGivenVersionPropertyIsNotLong() {
 
 			TypeInformation<EntityWithWrongVersionType> typeInformation = ClassTypeInformation
 					.from(EntityWithWrongVersionType.class);
 			SimpleElasticsearchPersistentEntity<EntityWithWrongVersionType> entity = new SimpleElasticsearchPersistentEntity<>(
-					typeInformation);
+					typeInformation, contextConfiguration);
 
 			assertThatThrownBy(() -> createProperty(entity, "version")).isInstanceOf(MappingException.class);
 		}
@@ -69,7 +76,7 @@ public class SimpleElasticsearchPersistentEntityTests extends MappingContextBase
 			TypeInformation<EntityWithMultipleVersionField> typeInformation = ClassTypeInformation
 					.from(EntityWithMultipleVersionField.class);
 			SimpleElasticsearchPersistentEntity<EntityWithMultipleVersionField> entity = new SimpleElasticsearchPersistentEntity<>(
-					typeInformation);
+					typeInformation, contextConfiguration);
 			SimpleElasticsearchPersistentProperty persistentProperty1 = createProperty(entity, "version1");
 			SimpleElasticsearchPersistentProperty persistentProperty2 = createProperty(entity, "version2");
 			entity.addPersistentProperty(persistentProperty1);
@@ -98,7 +105,7 @@ public class SimpleElasticsearchPersistentEntityTests extends MappingContextBase
 			TypeInformation<EntityWithoutSeqNoPrimaryTerm> typeInformation = ClassTypeInformation
 					.from(EntityWithoutSeqNoPrimaryTerm.class);
 			SimpleElasticsearchPersistentEntity<EntityWithoutSeqNoPrimaryTerm> entity = new SimpleElasticsearchPersistentEntity<>(
-					typeInformation);
+					typeInformation, contextConfiguration);
 
 			assertThat(entity.hasSeqNoPrimaryTermProperty()).isFalse();
 		}
@@ -109,7 +116,7 @@ public class SimpleElasticsearchPersistentEntityTests extends MappingContextBase
 			TypeInformation<EntityWithSeqNoPrimaryTerm> typeInformation = ClassTypeInformation
 					.from(EntityWithSeqNoPrimaryTerm.class);
 			SimpleElasticsearchPersistentEntity<EntityWithSeqNoPrimaryTerm> entity = new SimpleElasticsearchPersistentEntity<>(
-					typeInformation);
+					typeInformation, contextConfiguration);
 
 			entity.addPersistentProperty(createProperty(entity, "seqNoPrimaryTerm"));
 
@@ -123,7 +130,7 @@ public class SimpleElasticsearchPersistentEntityTests extends MappingContextBase
 			TypeInformation<EntityWithSeqNoPrimaryTerm> typeInformation = ClassTypeInformation
 					.from(EntityWithSeqNoPrimaryTerm.class);
 			SimpleElasticsearchPersistentEntity<EntityWithSeqNoPrimaryTerm> entity = new SimpleElasticsearchPersistentEntity<>(
-					typeInformation);
+					typeInformation, contextConfiguration);
 			entity.addPersistentProperty(createProperty(entity, "seqNoPrimaryTerm"));
 			EntityWithSeqNoPrimaryTerm instance = new EntityWithSeqNoPrimaryTerm();
 			SeqNoPrimaryTerm seqNoPrimaryTerm = new SeqNoPrimaryTerm(1, 2);
@@ -142,7 +149,7 @@ public class SimpleElasticsearchPersistentEntityTests extends MappingContextBase
 			TypeInformation<EntityWithSeqNoPrimaryTerm> typeInformation = ClassTypeInformation
 					.from(EntityWithSeqNoPrimaryTerm.class);
 			SimpleElasticsearchPersistentEntity<EntityWithSeqNoPrimaryTerm> entity = new SimpleElasticsearchPersistentEntity<>(
-					typeInformation);
+					typeInformation, contextConfiguration);
 			entity.addPersistentProperty(createProperty(entity, "seqNoPrimaryTerm"));
 
 			assertThatThrownBy(() -> entity.addPersistentProperty(createProperty(entity, "seqNoPrimaryTerm2")))
@@ -165,10 +172,9 @@ public class SimpleElasticsearchPersistentEntityTests extends MappingContextBase
 		@DisplayName("should error if index sorting parameters do not have the same number of arguments")
 		void shouldErrorIfIndexSortingParametersDoNotHaveTheSameNumberOfArguments() {
 
-			assertThatThrownBy(() -> {
+			assertThatThrownBy(() -> elasticsearchConverter.get().getMappingContext()
-				elasticsearchConverter.get().getMappingContext()
+					.getRequiredPersistentEntity(SettingsInvalidSortParameterSizes.class).getDefaultSettings())
-						.getRequiredPersistentEntity(SettingsInvalidSortParameterSizes.class).getDefaultSettings();
+							.isInstanceOf(IllegalArgumentException.class);
-			}).isInstanceOf(IllegalArgumentException.class);
 		}
 
 		@Test // #1719
@@ -190,6 +196,75 @@ public class SimpleElasticsearchPersistentEntityTests extends MappingContextBase
 		}
 	}
 
+	@Nested
+	@DisplayName("configuration")
+	class ConfigurationTests {
+
+		@Test // #1454
+		@DisplayName("should return FieldNamingStrategy from context configuration")
+		void shouldReturnFieldNamingStrategyFromContextConfiguration() {
+
+			SimpleElasticsearchMappingContext context = new SimpleElasticsearchMappingContext();
+			FieldNamingStrategy fieldNamingStrategy = new FieldNamingStrategy() {
+				@Override
+				public String getFieldName(PersistentProperty<?> property) {
+					return property.getName() + "foo";
+				}
+			};
+			context.setFieldNamingStrategy(fieldNamingStrategy);
+			SimpleElasticsearchPersistentEntity<?> persistentEntity = context
+					.getRequiredPersistentEntity(FieldNameEntity.class);
+
+			assertThat(persistentEntity.getFieldNamingStrategy()).isSameAs(fieldNamingStrategy);
+		}
+
+		@Test // #1454
+		@DisplayName("should write type hints on default context settings")
+		void shouldWriteTypeHintsOnDefaultContextSettings() {
+
+			SimpleElasticsearchMappingContext context = new SimpleElasticsearchMappingContext();
+			SimpleElasticsearchPersistentEntity<?> entity = context
+					.getRequiredPersistentEntity(DisableTypeHintNoSetting.class);
+
+			assertThat(entity.writeTypeHints()).isTrue();
+		}
+
+		@Test // #1454
+		@DisplayName("should not write type hints when configured in context settings")
+		void shouldNotWriteTypeHintsWhenConfiguredInContextSettings() {
+
+			SimpleElasticsearchMappingContext context = new SimpleElasticsearchMappingContext();
+			context.setWriteTypeHints(false);
+			SimpleElasticsearchPersistentEntity<?> entity = context
+					.getRequiredPersistentEntity(DisableTypeHintNoSetting.class);
+
+			assertThat(entity.writeTypeHints()).isFalse();
+		}
+
+		@Test // #1454
+		@DisplayName("should not write type hints when configured explicitly on entity")
+		void shouldNotWriteTypeHintsWhenConfiguredExplicitlyOnEntity() {
+
+			SimpleElasticsearchMappingContext context = new SimpleElasticsearchMappingContext();
+			SimpleElasticsearchPersistentEntity<?> entity = context
+					.getRequiredPersistentEntity(DisableTypeHintExplicitSetting.class);
+
+			assertThat(entity.writeTypeHints()).isFalse();
+		}
+
+		@Test // #1454
+		@DisplayName("should write type hints when  configured explicitly on entity and global setting is false")
+		void shouldWriteTypeHintsWhenConfiguredExplicitlyOnEntityAndGlobalSettingIsFalse() {
+
+			SimpleElasticsearchMappingContext context = new SimpleElasticsearchMappingContext();
+			context.setWriteTypeHints(false);
+			SimpleElasticsearchPersistentEntity<?> entity = context
+					.getRequiredPersistentEntity(EnableTypeHintExplicitSetting.class);
+
+			assertThat(entity.writeTypeHints()).isTrue();
+		}
+	}
+
 	// region helper functions
 	private static SimpleElasticsearchPersistentProperty createProperty(SimpleElasticsearchPersistentEntity<?> entity,
 			String fieldName) {
@@ -198,7 +273,7 @@ public class SimpleElasticsearchPersistentEntityTests extends MappingContextBase
 		java.lang.reflect.Field field = ReflectionUtils.findField(entity.getType(), fieldName);
 		assertThat(field).isNotNull();
 		Property property = Property.of(type, field);
-		return new SimpleElasticsearchPersistentProperty(property, entity, SimpleTypeHolder.DEFAULT, null);
+		return new SimpleElasticsearchPersistentProperty(property, entity, SimpleTypeHolder.DEFAULT);
 
 	}
 	// endregion
@@ -286,5 +361,18 @@ private static class SettingsValidSortParameterSizes {
 		@Nullable @Field(name = "second_field", type = FieldType.Keyword) private String secondField;
 	}
 
+	private static class DisableTypeHintNoSetting {
+		@Nullable @Id String id;
+	}
+
+	@Document(indexName = "foo", writeTypeHint = WriteTypeHint.FALSE)
+	private static class DisableTypeHintExplicitSetting {
+		@Nullable @Id String id;
+	}
+
+	@Document(indexName = "foo", writeTypeHint = WriteTypeHint.TRUE)
+	private static class EnableTypeHintExplicitSetting {
+		@Nullable @Id String id;
+	}
 	// endregion
 }

src/test/java/org/springframework/data/elasticsearch/core/mapping/SimpleElasticsearchPersistentPropertyUnitTests.java

@@ -36,6 +36,7 @@ import org.springframework.data.elasticsearch.core.query.SeqNoPrimaryTerm;
 import org.springframework.data.mapping.MappingException;
 import org.springframework.data.mapping.model.FieldNamingStrategy;
 import org.springframework.data.mapping.model.Property;
+import org.springframework.data.mapping.model.PropertyNameFieldNamingStrategy;
 import org.springframework.data.mapping.model.SimpleTypeHolder;
 import org.springframework.data.mapping.model.SnakeCaseFieldNamingStrategy;
 import org.springframework.data.util.ClassTypeInformation;
@@ -200,20 +201,22 @@ public class SimpleElasticsearchPersistentPropertyUnitTests {
 	@DisplayName("should use default FieldNamingStrategy")
 	void shouldUseDefaultFieldNamingStrategy() {
 
+		SimpleElasticsearchPersistentEntity.ContextConfiguration contextConfiguration = new SimpleElasticsearchPersistentEntity.ContextConfiguration(
+				PropertyNameFieldNamingStrategy.INSTANCE, true);
+
 		ElasticsearchPersistentEntity<FieldNamingStrategyEntity> entity = new SimpleElasticsearchPersistentEntity<>(
-				ClassTypeInformation.from(FieldNamingStrategyEntity.class));
+				ClassTypeInformation.from(FieldNamingStrategyEntity.class), contextConfiguration);
 		ClassTypeInformation<FieldNamingStrategyEntity> type = ClassTypeInformation.from(FieldNamingStrategyEntity.class);
 
 		java.lang.reflect.Field field = ReflectionUtils.findField(FieldNamingStrategyEntity.class,
 				"withoutCustomFieldName");
 		SimpleElasticsearchPersistentProperty property = new SimpleElasticsearchPersistentProperty(Property.of(type, field),
-				entity, SimpleTypeHolder.DEFAULT, null);
+				entity, SimpleTypeHolder.DEFAULT);
 
 		assertThat(property.getFieldName()).isEqualTo("withoutCustomFieldName");
 
 		field = ReflectionUtils.findField(FieldNamingStrategyEntity.class, "withCustomFieldName");
-		property = new SimpleElasticsearchPersistentProperty(Property.of(type, field), entity, SimpleTypeHolder.DEFAULT,
+		property = new SimpleElasticsearchPersistentProperty(Property.of(type, field), entity, SimpleTypeHolder.DEFAULT);
-				null);
 
 		assertThat(property.getFieldName()).isEqualTo("CUStomFIEldnAME");
 	}
@@ -223,25 +226,27 @@ public class SimpleElasticsearchPersistentPropertyUnitTests {
 	void shouldUseCustomFieldNamingStrategy() {
 
 		FieldNamingStrategy fieldNamingStrategy = new SnakeCaseFieldNamingStrategy();
+		SimpleElasticsearchPersistentEntity.ContextConfiguration contextConfiguration = new SimpleElasticsearchPersistentEntity.ContextConfiguration(
+				fieldNamingStrategy, true);
 
 		ElasticsearchPersistentEntity<FieldNamingStrategyEntity> entity = new SimpleElasticsearchPersistentEntity<>(
-				ClassTypeInformation.from(FieldNamingStrategyEntity.class));
+				ClassTypeInformation.from(FieldNamingStrategyEntity.class), contextConfiguration);
 		ClassTypeInformation<FieldNamingStrategyEntity> type = ClassTypeInformation.from(FieldNamingStrategyEntity.class);
 
 		java.lang.reflect.Field field = ReflectionUtils.findField(FieldNamingStrategyEntity.class,
 				"withoutCustomFieldName");
 		SimpleElasticsearchPersistentProperty property = new SimpleElasticsearchPersistentProperty(Property.of(type, field),
-				entity, SimpleTypeHolder.DEFAULT, fieldNamingStrategy);
+				entity, SimpleTypeHolder.DEFAULT);
 
 		assertThat(property.getFieldName()).isEqualTo("without_custom_field_name");
 
 		field = ReflectionUtils.findField(FieldNamingStrategyEntity.class, "withCustomFieldName");
-		property = new SimpleElasticsearchPersistentProperty(Property.of(type, field), entity, SimpleTypeHolder.DEFAULT,
+		property = new SimpleElasticsearchPersistentProperty(Property.of(type, field), entity, SimpleTypeHolder.DEFAULT);
-				fieldNamingStrategy);
 
 		assertThat(property.getFieldName()).isEqualTo("CUStomFIEldnAME");
 	}
 
+	// region entities
 	static class FieldNameProperty {
 		@Nullable @Field(name = "by-name") String fieldProperty;
 	}
@@ -319,4 +324,5 @@ public class SimpleElasticsearchPersistentPropertyUnitTests {
 			this.withCustomFieldName = withCustomFieldName;
 		}
 	}
+	// endregion
 }