DATAREDIS-543 - Allow customized type aliasing.

We now support customization of type aliasing for types using through Redis Repositories. Type aliasing can be customized by either annotating types with @TypeAlias or by providing a RedisTypeMapper to MappingRedisConverter.

Map<Class<?>, String> mapping = new HashMap<>();

mapping.put(Person.class, "person");

ConfigurableTypeInformationMapper mapper = new ConfigurableTypeInformationMapper(mapping);
RedisTypeMapper typeMapper = new DefaultRedisTypeMapper(DefaultRedisTypeMapper.DEFAULT_TYPE_KEY, Collections.singletonList(mapper))

alternatively:

@RedisHash
@TypeAlias("person")
class Person {
  // …
}

Original Pull Request: #299

src/main/asciidoc/reference/redis-repositories.adoc

@@ -254,6 +254,61 @@ ciudad = "emond's field"
 
 NOTE: Custom conversions have no effect on index resolution. <<redis.repositories.indexes>> will still be created even for custom converted types.
 
+=== Customizing type mapping
+
+In case you want to avoid writing the entire Java class name as type information but rather like to use some key you can use the `@TypeAlias` annotation at the entity class being persisted. If you need to customize the mapping even more have a look at the `TypeInformationMapper` interface. An instance of that interface can be configured at the `DefaultRedisTypeMapper` which can be configured on `MappingRedisConverter`.
+
+.Defining `@TypeAlias` for an Entity
+====
+[source,java]
+----
+@TypeAlias("pers")
+class Person {
+
+}
+----
+====
+
+Note that the resulting document will contain `pers` as the value in a `_class` field.
+
+==== Configuring custom type mapping
+
+The following example demonstrates how to configure a custom `RedisTypeMapper` in `MappingRedisConverter`.
+
+.Configuring a custom `RedisTypeMapper` via Spring Java Config
+====
+[source,java]
+----
+class CustomRedisTypeMapper extends DefaultRedisTypeMapper {
+  //implement custom type mapping here
+}
+----
+====
+
+[source,java]
+----
+@Configuration
+class SampleRedisConfiguration {
+
+  @Bean
+  public MappingRedisConverter redisConverter(RedisMappingContext mappingContext,
+        RedisCustomConversions customConversions, ReferenceResolver referenceResolver) {
+
+    MappingRedisConverter mappingRedisConverter = new MappingRedisConverter(mappingContext, null, referenceResolver,
+            customTypeMapper());
+
+    mappingRedisConverter.setCustomConversions(customConversions);
+
+    return mappingRedisConverter;
+  }
+
+  @Bean
+  public RedisTypeMapper customTypeMapper() {
+    return new CustomRedisTypeMapper();
+  }
+}
+----
+
 [[redis.repositories.keyspaces]]
 == Keyspaces
 Keyspaces define prefixes used to create the actual _key_ for the Redis Hash.

src/main/java/org/springframework/data/redis/core/convert/BucketPropertyPath.java

@@ -0,0 +1,86 @@
+/*
+ * Copyright 2017 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.redis.core.convert;
+
+import lombok.AccessLevel;
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.NonNull;
+
+import org.springframework.lang.Nullable;
+import org.springframework.util.StringUtils;
+
+/**
+ * Value object representing a path within a {@link Bucket}. Paths can be either top-level (if the {@code prefix} is
+ * {@literal null} or empty) or nested with a given {@code prefix}.
+ *
+ * @author Mark Paluch
+ * @since 2.1
+ */
+@AllArgsConstructor(access = AccessLevel.PRIVATE)
+@Getter
+public class BucketPropertyPath {
+
+	private final @NonNull Bucket bucket;
+	private final @Nullable String prefix;
+
+	/**
+	 * Creates a top-level {@link BucketPropertyPath} given {@link Bucket}.
+	 *
+	 * @param bucket the bucket, must not be {@literal null}.
+	 * @return {@link BucketPropertyPath} within the given {@link Bucket}.
+	 */
+	public static BucketPropertyPath from(Bucket bucket) {
+		return new BucketPropertyPath(bucket, null);
+	}
+
+	/**
+	 * Creates a {@link BucketPropertyPath} given {@link Bucket} and {@code prefix}. The resulting path is top-level if
+	 * {@code prefix} is empty or nested, if {@code prefix} is not empty.
+	 *
+	 * @param bucket the bucket, must not be {@literal null}.
+	 * @param prefix the prefix. Property path is top-level if {@code prefix} is {@literal null} or empty.
+	 * @return {@link BucketPropertyPath} within the given {@link Bucket} using {@code prefix}.
+	 */
+	public static BucketPropertyPath from(Bucket bucket, @Nullable String prefix) {
+		return new BucketPropertyPath(bucket, prefix);
+	}
+
+	/**
+	 * Retrieve a value at {@code key} considering top-level/nesting.
+	 *
+	 * @param key must not be {@literal null} or empty.
+	 * @return the resulting value, may be {@literal null}.
+	 */
+	@Nullable
+	public byte[] get(String key) {
+		return bucket.get(getPath(key));
+	}
+
+	/**
+	 * Write a {@code value} at {@code key} considering top-level/nesting.
+	 *
+	 * @param key must not be {@literal null} or empty.
+	 * @param value the value.
+	 */
+	public void put(String key, byte[] value) {
+		bucket.put(getPath(key), value);
+	}
+
+	private String getPath(String key) {
+		return StringUtils.hasText(prefix) ? prefix + "." + key : key;
+	}
+}

src/main/java/org/springframework/data/redis/core/convert/DefaultRedisTypeMapper.java

@@ -0,0 +1,170 @@
+/*
+ * Copyright 2017 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.redis.core.convert;
+
+import java.util.Collections;
+import java.util.List;
+
+import org.springframework.core.convert.ConversionService;
+import org.springframework.core.convert.support.GenericConversionService;
+import org.springframework.data.convert.DefaultTypeMapper;
+import org.springframework.data.convert.SimpleTypeInformationMapper;
+import org.springframework.data.convert.TypeAliasAccessor;
+import org.springframework.data.convert.TypeInformationMapper;
+import org.springframework.data.mapping.Alias;
+import org.springframework.data.mapping.PersistentEntity;
+import org.springframework.data.mapping.context.MappingContext;
+import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
+
+/**
+ * Default implementation of {@link RedisTypeMapper} allowing configuration of the key to lookup and store type
+ * information via {@link BucketPropertyPath} in buckets. The key defaults to {@link #DEFAULT_TYPE_KEY}. Actual
+ * type-to-{@code byte[]} conversion and back is done in {@link BucketTypeAliasAccessor}.
+ *
+ * @author Mark Paluch
+ * @since 2.1
+ */
+public class DefaultRedisTypeMapper extends DefaultTypeMapper<BucketPropertyPath> implements RedisTypeMapper {
+
+	public static final String DEFAULT_TYPE_KEY = "_class";
+
+	private final @Nullable String typeKey;
+
+	/**
+	 * Create a new {@link DefaultRedisTypeMapper} using {@link #DEFAULT_TYPE_KEY} to exchange type hints.
+	 */
+	public DefaultRedisTypeMapper() {
+		this(DEFAULT_TYPE_KEY);
+	}
+
+	/**
+	 * Create a new {@link DefaultRedisTypeMapper} given {@code typeKey} to exchange type hints. Does not consider type
+	 * hints if {@code typeKey} is {@literal null}.
+	 *
+	 * @param typeKey the type key can be {@literal null} to skip type hinting.
+	 */
+	public DefaultRedisTypeMapper(@Nullable String typeKey) {
+		this(typeKey, Collections.singletonList(new SimpleTypeInformationMapper()));
+	}
+
+	/**
+	 * Create a new {@link DefaultRedisTypeMapper} given {@code typeKey} to exchange type hints and
+	 * {@link MappingContext}. Does not consider type hints if {@code typeKey} is {@literal null}. {@link MappingContext}
+	 * is used to obtain entity-based aliases
+	 *
+	 * @param typeKey the type key can be {@literal null} to skip type hinting.
+	 * @param mappingContext must not be {@literal null}.
+	 * @see org.springframework.data.annotation.TypeAlias
+	 */
+	public DefaultRedisTypeMapper(@Nullable String typeKey,
+			MappingContext<? extends PersistentEntity<?, ?>, ?> mappingContext) {
+		this(typeKey, new BucketTypeAliasAccessor(typeKey, getConversionService()), mappingContext,
+				Collections.singletonList(new SimpleTypeInformationMapper()));
+	}
+
+	/**
+	 * Create a new {@link DefaultRedisTypeMapper} given {@code typeKey} to exchange type hints and {@link List} of
+	 * {@link TypeInformationMapper}. Does not consider type hints if {@code typeKey} is {@literal null}.
+	 * {@link MappingContext} is used to obtain entity-based aliases
+	 *
+	 * @param typeKey the type key can be {@literal null} to skip type hinting.
+	 * @param mappers must not be {@literal null}.
+	 */
+	public DefaultRedisTypeMapper(@Nullable String typeKey, List<? extends TypeInformationMapper> mappers) {
+		this(typeKey, new BucketTypeAliasAccessor(typeKey, getConversionService()), null, mappers);
+	}
+
+	private DefaultRedisTypeMapper(@Nullable String typeKey, TypeAliasAccessor<BucketPropertyPath> accessor,
+			@Nullable MappingContext<? extends PersistentEntity<?, ?>, ?> mappingContext,
+			List<? extends TypeInformationMapper> mappers) {
+
+		super(accessor, mappingContext, mappers);
+
+		this.typeKey = typeKey;
+	}
+
+	private static GenericConversionService getConversionService() {
+
+		GenericConversionService conversionService = new GenericConversionService();
+		new RedisCustomConversions().registerConvertersIn(conversionService);
+
+		return conversionService;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.convert.RedisTypeMapper#isTypeKey(java.lang.String)
+	 */
+	public boolean isTypeKey(@Nullable String key) {
+		return key != null && typeKey != null && key.endsWith(typeKey);
+	}
+
+	/**
+	 * {@link TypeAliasAccessor} to store aliases in a {@link Bucket}.
+	 *
+	 * @author Mark Paluch
+	 */
+	static final class BucketTypeAliasAccessor implements TypeAliasAccessor<BucketPropertyPath> {
+
+		private final @Nullable String typeKey;
+
+		private final ConversionService conversionService;
+
+		BucketTypeAliasAccessor(@Nullable String typeKey, ConversionService conversionService) {
+
+			Assert.notNull(conversionService, "ConversionService must not be null!");
+
+			this.typeKey = typeKey;
+			this.conversionService = conversionService;
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.convert.TypeAliasAccessor#readAliasFrom(java.lang.Object)
+		 */
+		public Alias readAliasFrom(BucketPropertyPath source) {
+
+			if (typeKey == null || source instanceof List) {
+				return Alias.NONE;
+			}
+
+			byte[] bytes = source.get(typeKey);
+
+			if (bytes != null) {
+				return Alias.ofNullable(conversionService.convert(bytes, String.class));
+			}
+
+			return Alias.NONE;
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.convert.TypeAliasAccessor#writeTypeTo(java.lang.Object, java.lang.Object)
+		 */
+		public void writeTypeTo(BucketPropertyPath sink, Object alias) {
+
+			if (typeKey != null) {
+
+				if (alias instanceof byte[]) {
+					sink.put(typeKey, (byte[]) alias);
+				} else {
+					sink.put(typeKey, conversionService.convert(alias, byte[].class));
+				}
+			}
+		}
+	}
+}

src/main/java/org/springframework/data/redis/core/convert/MappingRedisConverter.java

@@ -42,13 +42,9 @@ import org.springframework.core.convert.ConverterNotFoundException;
 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.DefaultTypeMapper;
 import org.springframework.data.convert.EntityInstantiator;
 import org.springframework.data.convert.EntityInstantiators;
-import org.springframework.data.convert.TypeAliasAccessor;
-import org.springframework.data.convert.TypeMapper;
 import org.springframework.data.keyvalue.core.mapping.KeySpaceResolver;
-import org.springframework.data.mapping.Alias;
 import org.springframework.data.mapping.AssociationHandler;
 import org.springframework.data.mapping.MappingException;
 import org.springframework.data.mapping.PersistentPropertyAccessor;
@@ -121,13 +117,12 @@ import org.springframework.util.comparator.NullSafeComparator;
  */
 public class MappingRedisConverter implements RedisConverter, InitializingBean {
 
-	private static final String TYPE_HINT_ALIAS = "_class";
 	private static final String INVALID_TYPE_ASSIGNMENT = "Value of type %s cannot be assigned to property %s of type %s.";
 
 	private final RedisMappingContext mappingContext;
 	private final GenericConversionService conversionService;
 	private final EntityInstantiators entityInstantiators;
-	private final TypeMapper<RedisData> typeMapper;
+	private final RedisTypeMapper typeMapper;
 	private final Comparator<String> listKeyComparator = new NullSafeComparator<>(NaturalOrderingKeyComparator.INSTANCE,
 			true);
 
@@ -141,7 +136,7 @@ public class MappingRedisConverter implements RedisConverter, InitializingBean {
 	 * @param context can be {@literal null}.
 	 */
 	MappingRedisConverter(RedisMappingContext context) {
-		this(context, null, null);
+		this(context, null, null, null);
 	}
 
 	/**
@@ -149,17 +144,32 @@ public class MappingRedisConverter implements RedisConverter, InitializingBean {
 	 *
 	 * @param mappingContext can be {@literal null}.
 	 * @param indexResolver can be {@literal null}.
-	 * @param referenceResolver must not be {@literal null}.
+	 * @param referenceResolver can be not be {@literal null}.
 	 */
 	public MappingRedisConverter(@Nullable RedisMappingContext mappingContext, @Nullable IndexResolver indexResolver,
 			@Nullable ReferenceResolver referenceResolver) {
+		this(mappingContext, indexResolver, referenceResolver, null);
+	}
+
+	/**
+	 * Creates new {@link MappingRedisConverter} and defaults {@link RedisMappingContext} when {@literal null}.
+	 *
+	 * @param mappingContext can be {@literal null}.
+	 * @param indexResolver can be {@literal null}.
+	 * @param referenceResolver can be {@literal null}.
+	 * @param typeMapper can be {@literal null}.
+	 * @since 2.1
+	 */
+	public MappingRedisConverter(@Nullable RedisMappingContext mappingContext, @Nullable IndexResolver indexResolver,
+			@Nullable ReferenceResolver referenceResolver, @Nullable RedisTypeMapper typeMapper) {
 
 		this.mappingContext = mappingContext != null ? mappingContext : new RedisMappingContext();
 
 		this.entityInstantiators = new EntityInstantiators();
 		this.conversionService = new DefaultConversionService();
 		this.customConversions = new RedisCustomConversions();
-		this.typeMapper = new DefaultTypeMapper<>(new RedisTypeAliasAccessor(this.conversionService));
+		this.typeMapper = typeMapper != null ? typeMapper
+				: new DefaultRedisTypeMapper(DefaultRedisTypeMapper.DEFAULT_TYPE_KEY, this.mappingContext);
 
 		this.indexResolver = indexResolver != null ? indexResolver : new PathIndexResolver(this.mappingContext);
 		this.referenceResolver = referenceResolver;
@@ -182,10 +192,8 @@ public class MappingRedisConverter implements RedisConverter, InitializingBean {
 			return null;
 		}
 
-		TypeInformation<?> readType = typeMapper.readType(source);
-		if (readType == null) {
-			readType = ClassTypeInformation.from(type);
-		}
+		TypeInformation<?> readType = typeMapper.readType(BucketPropertyPath.from(source.getBucket()),
+				ClassTypeInformation.from(type));
 
 		RedisPersistentEntity<?> entity = mappingContext.getPersistentEntity(readType);
 
@@ -267,16 +275,13 @@ public class MappingRedisConverter implements RedisConverter, InitializingBean {
 			} else if (persistentProperty.isEntity() && !conversionService.canConvert(byte[].class,
 					persistentProperty.getTypeInformation().getActualType().getType())) {
 
-				Class<R> targetType = (Class<R>) persistentProperty.getTypeInformation().getActualType().getType();
-
 				Bucket bucket = source.getBucket().extract(currentPath + ".");
 
 				RedisData newBucket = new RedisData(bucket);
+				TypeInformation<?> typeInformation = typeMapper.readType(BucketPropertyPath.from(bucket, currentPath),
+						persistentProperty.getTypeInformation().getActualType());
 
-				byte[] type1 = bucket.get(currentPath + "." + TYPE_HINT_ALIAS);
-				if (type1 != null && type1.length > 0) {
-					newBucket.getBucket().put(TYPE_HINT_ALIAS, type1);
-				}
+				Class<R> targetType = (Class<R>) typeInformation.getType();
 
 				R val = readInternal(currentPath, targetType, newBucket);
 
@@ -381,12 +386,12 @@ public class MappingRedisConverter implements RedisConverter, InitializingBean {
 		RedisPersistentEntity<?> entity = mappingContext.getPersistentEntity(source.getClass());
 
 		if (!customConversions.hasCustomWriteTarget(source.getClass())) {
-			typeMapper.writeType(ClassUtils.getUserClass(source), sink);
+			typeMapper.writeType(ClassUtils.getUserClass(source), BucketPropertyPath.from(sink.getBucket()));
 		}
 
 		if (entity == null) {
 
-			typeMapper.writeType(ClassUtils.getUserClass(source), sink);
+			typeMapper.writeType(ClassUtils.getUserClass(source), BucketPropertyPath.from(sink.getBucket()));
 			sink.getBucket().put("_raw", conversionService.convert(source, byte[].class));
 			return;
 		}
@@ -416,8 +421,12 @@ public class MappingRedisConverter implements RedisConverter, InitializingBean {
 		RedisPersistentEntity<?> entity = mappingContext.getRequiredPersistentEntity(update.getTarget());
 
 		write(update.getValue(), sink);
-		if (sink.getBucket().keySet().contains(TYPE_HINT_ALIAS)) {
-			sink.getBucket().remove(TYPE_HINT_ALIAS); // overwrite stuff in here
+
+		for (String key : sink.getBucket().keySet()) {
+			if (typeMapper.isTypeKey(key)) {
+				sink.getBucket().remove(key);
+				break;
+			}
 		}
 
 		if (update.isRefreshTtl() && !update.getPropertyUpdates().isEmpty()) {
@@ -574,8 +583,7 @@ public class MappingRedisConverter implements RedisConverter, InitializingBean {
 		}
 
 		if (value.getClass() != typeHint.getType()) {
-			sink.getBucket().put((!path.isEmpty() ? path + "." + TYPE_HINT_ALIAS : TYPE_HINT_ALIAS),
-					toBytes(value.getClass().getName()));
+			typeMapper.writeType(value.getClass(), BucketPropertyPath.from(sink.getBucket(), path));
 		}
 
 		RedisPersistentEntity<?> entity = mappingContext.getRequiredPersistentEntity(value.getClass());
@@ -735,8 +743,7 @@ public class MappingRedisConverter implements RedisConverter, InitializingBean {
 
 			if (!targetType.filter(it -> ClassUtils.isAssignable(Map.class, it)).isPresent()
 					&& customConversions.isSimpleType(value.getClass()) && value.getClass() != propertyType) {
-				sink.getBucket().put((!path.isEmpty() ? path + "." + TYPE_HINT_ALIAS : TYPE_HINT_ALIAS),
-						toBytes(value.getClass().getName()));
+				typeMapper.writeType(value.getClass(), BucketPropertyPath.from(sink.getBucket(), path));
 			}
 
 			if (targetType.filter(it -> ClassUtils.isAssignable(Map.class, it)).isPresent()) {
@@ -753,7 +760,6 @@ public class MappingRedisConverter implements RedisConverter, InitializingBean {
 						String.format("Cannot convert value '%s' of type %s to bytes.", value, value.getClass()));
 			}
 		}
-
 	}
 
 	private Object readCollectionOrArray(String path, Class<?> collectionType, Class<?> valueType, Bucket bucket) {
@@ -767,22 +773,20 @@ public class MappingRedisConverter implements RedisConverter, InitializingBean {
 
 		for (String key : keys) {
 
-			if (key.endsWith(TYPE_HINT_ALIAS)) {
+			if (typeMapper.isTypeKey(key)) {
 				continue;
 			}
 
 			Bucket elementData = bucket.extract(key);
 
-			byte[] typeInfo = elementData.get(key + "." + TYPE_HINT_ALIAS);
-			if (typeInfo != null && typeInfo.length > 0) {
-				elementData.put(TYPE_HINT_ALIAS, typeInfo);
-			}
+			TypeInformation<?> typeInformation = typeMapper.readType(BucketPropertyPath.from(elementData, key),
+					ClassTypeInformation.from(valueType));
 
-			Class<?> typeToUse = getTypeHint(key, elementData, valueType);
+			Class<?> typeToUse = typeInformation.getType();
 			if (conversionService.canConvert(byte[].class, typeToUse)) {
 				target.add(fromBytes(elementData.get(key), typeToUse));
 			} else {
-				target.add(readInternal(key, valueType, new RedisData(elementData)));
+				target.add(readInternal(key, typeToUse, new RedisData(elementData)));
 			}
 		}
 
@@ -850,7 +854,7 @@ public class MappingRedisConverter implements RedisConverter, InitializingBean {
 
 		for (Entry<String, byte[]> entry : partial.entrySet()) {
 
-			if (entry.getKey().endsWith(TYPE_HINT_ALIAS)) {
+			if (typeMapper.isTypeKey(entry.getKey())) {
 				continue;
 			}
 
@@ -881,16 +885,14 @@ public class MappingRedisConverter implements RedisConverter, InitializingBean {
 		for (String key : keys) {
 
 			Bucket partial = source.getBucket().extract(key);
-
-			byte[] typeInfo = partial.get(key + "." + TYPE_HINT_ALIAS);
-			if (typeInfo != null && typeInfo.length > 0) {
-				partial.put(TYPE_HINT_ALIAS, typeInfo);
-			}
-
-			Object value = readInternal(key, valueType, new RedisData(partial));
-
+			
 			Object mapKey = extractMapKeyForPath(path, key, keyType);
-			target.put(mapKey, value);
+
+			TypeInformation<?> typeInformation = typeMapper.readType(BucketPropertyPath.from(source.getBucket(), key),
+					ClassTypeInformation.from(valueType));
+
+			Object o = readInternal(key, typeInformation.getType(), new RedisData(partial));
+			target.put(mapKey, o);
 		}
 
 		return target.isEmpty() ? null : target;
@@ -918,18 +920,9 @@ public class MappingRedisConverter implements RedisConverter, InitializingBean {
 
 	private Class<?> getTypeHint(String path, Bucket bucket, Class<?> fallback) {
 
-		byte[] typeInfo = bucket.get(path + "." + TYPE_HINT_ALIAS);
-
-		if (typeInfo == null || typeInfo.length < 1) {
-			return fallback;
-		}
-
-		String typeName = fromBytes(typeInfo, String.class);
-		try {
-			return ClassUtils.forName(typeName, this.getClass().getClassLoader());
-		} catch (ClassNotFoundException | LinkageError e) {
-			throw new MappingException(String.format("Cannot find class for type %s. ", typeName), e);
-		}
+		TypeInformation<?> typeInformation = typeMapper.readType(BucketPropertyPath.from(bucket, path),
+				ClassTypeInformation.from(fallback));
+		return typeInformation.getType();
 	}
 
 	/**
@@ -1062,36 +1055,6 @@ public class MappingRedisConverter implements RedisConverter, InitializingBean {
 		}
 	}
 
-	/**
-	 * @author Christoph Strobl
-	 */
-	private static class RedisTypeAliasAccessor implements TypeAliasAccessor<RedisData> {
-
-		private final String typeKey;
-
-		private final ConversionService conversionService;
-
-		RedisTypeAliasAccessor(ConversionService conversionService) {
-			this(conversionService, TYPE_HINT_ALIAS);
-		}
-
-		RedisTypeAliasAccessor(ConversionService conversionService, String typeKey) {
-
-			this.conversionService = conversionService;
-			this.typeKey = typeKey;
-		}
-
-		@Override
-		public Alias readAliasFrom(RedisData source) {
-			return Alias.ofNullable(conversionService.convert(source.getBucket().get(typeKey), String.class));
-		}
-
-		@Override
-		public void writeTypeTo(RedisData sink, Object alias) {
-			sink.getBucket().put(typeKey, conversionService.convert(alias, byte[].class));
-		}
-	}
-
 	enum ClassNameKeySpaceResolver implements KeySpaceResolver {
 
 		INSTANCE;

src/main/java/org/springframework/data/redis/core/convert/RedisTypeMapper.java

@@ -0,0 +1,35 @@
+/*
+ * Copyright 2017 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.redis.core.convert;
+
+import org.springframework.data.convert.TypeMapper;
+
+/**
+ * Redis-specific {@link TypeMapper} exposing that {@link BucketPropertyPath}s might contain a type key.
+ *
+ * @author Mark Paluch
+ * @since 2.1
+ * @see BucketPropertyPath
+ */
+public interface RedisTypeMapper extends TypeMapper<BucketPropertyPath> {
+
+	/**
+	 * Returns whether the given {@code key} is the type key.
+	 *
+	 * @return {@literal true} if the given {@code key} is the type key.
+	 */
+	boolean isTypeKey(String key);
+}

src/test/java/org/springframework/data/redis/core/convert/ConversionTestEntities.java

@@ -34,6 +34,7 @@ import java.util.concurrent.TimeUnit;
 
 import org.springframework.data.annotation.Id;
 import org.springframework.data.annotation.Reference;
+import org.springframework.data.annotation.TypeAlias;
 import org.springframework.data.redis.core.RedisHash;
 import org.springframework.data.redis.core.TimeToLive;
 import org.springframework.data.redis.core.index.Indexed;
@@ -109,6 +110,7 @@ public class ConversionTestEntities {
 		}
 	}
 
+	@TypeAlias("with-post-code")
 	public static class AddressWithPostcode extends Address {
 
 		String postcode;

src/test/java/org/springframework/data/redis/core/convert/DefaultRedisTypeMapperUnitTests.java

@@ -0,0 +1,224 @@
+/*
+ * Copyright 2017 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.redis.core.convert;
+
+import static java.util.Collections.*;
+import static org.assertj.core.api.Assertions.*;
+
+import java.util.Arrays;
+import java.util.Collections;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.springframework.core.convert.support.GenericConversionService;
+import org.springframework.data.convert.ConfigurableTypeInformationMapper;
+import org.springframework.data.convert.SimpleTypeInformationMapper;
+import org.springframework.data.util.TypeInformation;
+import org.springframework.lang.Nullable;
+
+/**
+ * Unit tests for {@link DefaultRedisTypeMapper}.
+ *
+ * @author Mark Paluch
+ */
+public class DefaultRedisTypeMapperUnitTests {
+
+	GenericConversionService conversionService;
+	ConfigurableTypeInformationMapper configurableTypeInformationMapper;
+	SimpleTypeInformationMapper simpleTypeInformationMapper;
+	DefaultRedisTypeMapper typeMapper;
+
+	@Before
+	public void setUp() {
+
+		conversionService = new GenericConversionService();
+		new RedisCustomConversions().registerConvertersIn(conversionService);
+
+		configurableTypeInformationMapper = new ConfigurableTypeInformationMapper(singletonMap(String.class, "1"));
+		simpleTypeInformationMapper = new SimpleTypeInformationMapper();
+
+		typeMapper = new DefaultRedisTypeMapper(DefaultRedisTypeMapper.DEFAULT_TYPE_KEY);
+	}
+
+	@Test // DATAREDIS-543
+	public void defaultInstanceWritesClasses() {
+		writesTypeToField(new Bucket(), String.class, String.class.getName());
+	}
+
+	@Test // DATAREDIS-543
+	public void defaultInstanceReadsClasses() {
+
+		Bucket bucket = Bucket
+				.newBucketFromStringMap(singletonMap(DefaultRedisTypeMapper.DEFAULT_TYPE_KEY, String.class.getName()));
+		readsTypeFromField(bucket, String.class);
+	}
+
+	@Test // DATAREDIS-543
+	public void writesMapKeyForType() {
+
+		typeMapper = new DefaultRedisTypeMapper(DefaultRedisTypeMapper.DEFAULT_TYPE_KEY,
+				Collections.singletonList(configurableTypeInformationMapper));
+
+		writesTypeToField(new Bucket(), String.class, "1");
+		writesTypeToField(new Bucket(), Object.class, null);
+	}
+
+	@Test // DATAREDIS-543
+	public void writesClassNamesForUnmappedValuesIfConfigured() {
+
+		typeMapper = new DefaultRedisTypeMapper(DefaultRedisTypeMapper.DEFAULT_TYPE_KEY,
+				Arrays.asList(configurableTypeInformationMapper, simpleTypeInformationMapper));
+
+		writesTypeToField(new Bucket(), String.class, "1");
+		writesTypeToField(new Bucket(), Object.class, Object.class.getName());
+	}
+
+	@Test // DATAREDIS-543
+	public void readsTypeForMapKey() {
+
+		typeMapper = new DefaultRedisTypeMapper(DefaultRedisTypeMapper.DEFAULT_TYPE_KEY,
+				Collections.singletonList(configurableTypeInformationMapper));
+
+		readsTypeFromField(Bucket.newBucketFromStringMap(singletonMap(DefaultRedisTypeMapper.DEFAULT_TYPE_KEY, "1")),
+				String.class);
+		readsTypeFromField(Bucket.newBucketFromStringMap(singletonMap(DefaultRedisTypeMapper.DEFAULT_TYPE_KEY, "unmapped")),
+				null);
+	}
+
+	@Test // DATAREDIS-543
+	public void readsTypeLoadingClassesForUnmappedTypesIfConfigured() {
+
+		typeMapper = new DefaultRedisTypeMapper(DefaultRedisTypeMapper.DEFAULT_TYPE_KEY,
+				Arrays.asList(configurableTypeInformationMapper, simpleTypeInformationMapper));
+
+		readsTypeFromField(new Bucket(singletonMap(DefaultRedisTypeMapper.DEFAULT_TYPE_KEY, "1".getBytes())), String.class);
+		readsTypeFromField(
+				new Bucket(singletonMap(DefaultRedisTypeMapper.DEFAULT_TYPE_KEY, Object.class.getName().getBytes())),
+				Object.class);
+	}
+
+	@Test // DATAREDIS-543
+	public void addsFullyQualifiedClassNameUnderDefaultKeyByDefault() {
+		writesTypeToField(DefaultRedisTypeMapper.DEFAULT_TYPE_KEY, new Bucket(), String.class);
+	}
+
+	@Test // DATAREDIS-543
+	public void writesTypeToCustomFieldIfConfigured() {
+		typeMapper = new DefaultRedisTypeMapper("_custom");
+		writesTypeToField("_custom", new Bucket(), String.class);
+	}
+
+	@Test // DATAREDIS-543
+	public void doesNotWriteTypeInformationInCaseKeyIsSetToNull() {
+		typeMapper = new DefaultRedisTypeMapper(null);
+		writesTypeToField(null, new Bucket(), String.class);
+	}
+
+	@Test // DATAREDIS-543
+	public void readsTypeFromDefaultKeyByDefault() {
+		readsTypeFromField(
+				Bucket.newBucketFromStringMap(singletonMap(DefaultRedisTypeMapper.DEFAULT_TYPE_KEY, String.class.getName())),
+				String.class);
+	}
+
+	@Test // DATAREDIS-543
+	public void readsTypeFromCustomFieldConfigured() {
+
+		typeMapper = new DefaultRedisTypeMapper("_custom");
+		readsTypeFromField(Bucket.newBucketFromStringMap(singletonMap("_custom", String.class.getName())), String.class);
+	}
+
+	@Test // DATAREDIS-543
+	public void returnsListForBasicDBLists() {
+		readsTypeFromField(new Bucket(), null);
+	}
+
+	@Test // DATAREDIS-543
+	public void returnsNullIfNoTypeInfoInBucket() {
+
+		readsTypeFromField(new Bucket(), null);
+		readsTypeFromField(Bucket.newBucketFromStringMap(singletonMap(DefaultRedisTypeMapper.DEFAULT_TYPE_KEY, "")), null);
+	}
+
+	@Test // DATAREDIS-543
+	public void returnsNullIfClassCannotBeLoaded() {
+
+		readsTypeFromField(Bucket.newBucketFromStringMap(singletonMap(DefaultRedisTypeMapper.DEFAULT_TYPE_KEY, "fooBar")),
+				null);
+	}
+
+	@Test // DATAREDIS-543
+	public void returnsNullIfTypeKeySetToNull() {
+
+		typeMapper = new DefaultRedisTypeMapper(null);
+		readsTypeFromField(
+				Bucket.newBucketFromStringMap(singletonMap(DefaultRedisTypeMapper.DEFAULT_TYPE_KEY, String.class.getName())),
+				null);
+	}
+
+	@Test // DATAREDIS-543
+	public void returnsCorrectTypeKey() {
+
+		assertThat(typeMapper.isTypeKey(DefaultRedisTypeMapper.DEFAULT_TYPE_KEY)).isTrue();
+
+		typeMapper = new DefaultRedisTypeMapper("_custom");
+		assertThat(typeMapper.isTypeKey("_custom")).isTrue();
+		assertThat(typeMapper.isTypeKey(DefaultRedisTypeMapper.DEFAULT_TYPE_KEY)).isFalse();
+
+		typeMapper = new DefaultRedisTypeMapper(null);
+		assertThat(typeMapper.isTypeKey("_custom")).isFalse();
+		assertThat(typeMapper.isTypeKey(DefaultRedisTypeMapper.DEFAULT_TYPE_KEY)).isFalse();
+	}
+
+	private void readsTypeFromField(Bucket bucket, @Nullable Class<?> type) {
+
+		TypeInformation<?> typeInfo = typeMapper.readType(BucketPropertyPath.from(bucket));
+
+		if (type != null) {
+			assertThat(typeInfo).isNotNull();
+			assertThat(typeInfo.getType()).isAssignableFrom(type);
+		} else {
+			assertThat(typeInfo).isNull();
+		}
+	}
+
+	private void writesTypeToField(@Nullable String field, Bucket bucket, Class<?> type) {
+
+		typeMapper.writeType(type, BucketPropertyPath.from(bucket));
+
+		if (field == null) {
+			assertThat(bucket.keySet()).isEmpty();
+		} else {
+			assertThat(bucket.asMap()).containsKey(field);
+			assertThat(bucket.get(field)).isEqualTo(type.getName().getBytes());
+		}
+	}
+
+	private void writesTypeToField(Bucket bucket, Class<?> type, @Nullable Object value) {
+
+		typeMapper.writeType(type, BucketPropertyPath.from(bucket));
+
+		if (value == null) {
+			assertThat(bucket.keySet()).isEmpty();
+		} else {
+
+			byte[] expected = value instanceof Class ? ((Class) value).getName().getBytes() : value.toString().getBytes();
+
+			assertThat(bucket.asMap()).containsKey(DefaultRedisTypeMapper.DEFAULT_TYPE_KEY);
+			assertThat(bucket.get(DefaultRedisTypeMapper.DEFAULT_TYPE_KEY)).isEqualTo(expected);
+		}
+	}
+}

src/test/java/org/springframework/data/redis/core/convert/MappingRedisConverterUnitTests.java

@@ -32,16 +32,7 @@ import java.time.LocalTime;
 import java.time.Period;
 import java.time.ZoneId;
 import java.time.ZonedDateTime;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Calendar;
-import java.util.Collections;
-import java.util.Date;
-import java.util.HashMap;
-import java.util.LinkedHashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.UUID;
+import java.util.*;
 
 import org.hamcrest.core.IsEqual;
 import org.junit.Before;
@@ -94,6 +85,15 @@ public class MappingRedisConverterUnitTests {
 		assertThat(write(rand).getBucket(), isBucket().containingTypeHint("_class", Person.class));
 	}
 
+	@Test // DATAREDIS-543
+	public void writeSkipsTypeHintIfConfigured() {
+
+		converter = new MappingRedisConverter(new RedisMappingContext(), null, resolverMock);
+		converter.afterPropertiesSet();
+
+		assertThat(write(rand).getBucket(), isBucket().containingTypeHint("_class", Person.class));
+	}
+
 	@Test // DATAREDIS-425
 	public void writeAppendsKeyCorrectly() {
 
@@ -205,7 +205,7 @@ public class MappingRedisConverterUnitTests {
 		assertThat(target.getBucket(), isBucket().without("address._class"));
 	}
 
-	@Test // DATAREDIS-425
+	@Test // DATAREDIS-425, DATAREDIS-543
 	public void writeAddsClassTypeInformationCorrectlyForNonMatchingTypes() {
 
 		AddressWithPostcode address = new AddressWithPostcode();
@@ -216,7 +216,7 @@ public class MappingRedisConverterUnitTests {
 
 		RedisData target = write(rand);
 
-		assertThat(target.getBucket(), isBucket().containingTypeHint("address._class", AddressWithPostcode.class));
+		assertThat(target.getBucket(), isBucket().containingUtf8String("address._class", "with-post-code"));
 	}
 
 	@Test // DATAREDIS-425

src/test/java/org/springframework/data/redis/repository/RedisRepositoryIntegrationTests.java

@@ -15,19 +15,38 @@
  */
 package org.springframework.data.redis.repository;
 
+import static org.hamcrest.MatcherAssert.*;
+import static org.hamcrest.Matchers.*;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.junit.Test;
 import org.junit.runner.RunWith;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.ComponentScan;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.FilterType;
+import org.springframework.data.convert.ConfigurableTypeInformationMapper;
 import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
+import org.springframework.data.redis.core.RedisOperations;
 import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.core.convert.DefaultRedisTypeMapper;
+import org.springframework.data.redis.core.convert.MappingRedisConverter;
+import org.springframework.data.redis.core.convert.RedisCustomConversions;
+import org.springframework.data.redis.core.convert.RedisTypeMapper;
+import org.springframework.data.redis.core.convert.ReferenceResolver;
+import org.springframework.data.redis.core.mapping.RedisMappingContext;
 import org.springframework.data.redis.repository.configuration.EnableRedisRepositories;
+import org.springframework.data.redis.serializer.StringRedisSerializer;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
 /**
  * @author Christoph Strobl
+ * @author Mark Paluch
  */
 @RunWith(SpringJUnit4ClassRunner.class)
 @ContextConfiguration
@@ -45,10 +64,52 @@ public class RedisRepositoryIntegrationTests extends RedisRepositoryIntegrationT
 			JedisConnectionFactory connectionFactory = new JedisConnectionFactory();
 			connectionFactory.afterPropertiesSet();
 
-			RedisTemplate<byte[], byte[]> template = new RedisTemplate<>();
+			RedisTemplate<String, String> template = new RedisTemplate<>();
+			template.setDefaultSerializer(StringRedisSerializer.UTF_8);
 			template.setConnectionFactory(connectionFactory);
 
 			return template;
 		}
+
+		@Bean
+		public MappingRedisConverter redisConverter(RedisMappingContext mappingContext,
+				RedisCustomConversions customConversions, ReferenceResolver referenceResolver) {
+
+			MappingRedisConverter mappingRedisConverter = new MappingRedisConverter(mappingContext, null, referenceResolver,
+					customTypeMapper());
+
+			mappingRedisConverter.setCustomConversions(customConversions);
+
+			return mappingRedisConverter;
+		}
+
+		private RedisTypeMapper customTypeMapper() {
+
+			Map<Class<?>, String> mapping = new HashMap<>();
+
+			mapping.put(Person.class, "person");
+			mapping.put(City.class, "city");
+
+			ConfigurableTypeInformationMapper mapper = new ConfigurableTypeInformationMapper(mapping);
+
+			return new DefaultRedisTypeMapper(DefaultRedisTypeMapper.DEFAULT_TYPE_KEY, Collections.singletonList(mapper));
+		}
+	}
+
+	@Autowired RedisOperations<String, String> operations;
+
+	@Test // DATAREDIS-543
+	public void shouldConsiderCustomTypeMapper() {
+
+		Person rand = new Person();
+		rand.id = "rand";
+		rand.firstname = "rand";
+		rand.lastname = "al'thor";
+
+		repo.save(rand);
+
+		Map<String, String> entries = operations.<String, String> opsForHash().entries("persons:rand");
+
+		assertThat(entries.get("_class"), is(equalTo("person")));
 	}
 }