diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/CassandraMappingContext.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/CassandraMappingContext.java
index 77649a95d..28757ea51 100644
--- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/CassandraMappingContext.java
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/CassandraMappingContext.java
@@ -57,6 +57,7 @@ import org.springframework.util.ClassUtils;
 import org.springframework.util.StringUtils;
 
 import com.datastax.driver.core.DataType;
+import com.datastax.driver.core.DataType.Name;
 
 /**
  * Default implementation of a {@link MappingContext} for Cassandra using {@link CassandraPersistentEntity} and
@@ -489,6 +490,25 @@ public class CassandraMappingContext
 			DataTypeProvider dataTypeProvider) {
 
 		if (property.isAnnotationPresent(CassandraType.class)) {
+
+			CassandraType annotation = property.getRequiredAnnotation(CassandraType.class);
+
+			if (annotation.type() == Name.UDT) {
+
+				CqlIdentifier userTypeName = CqlIdentifier.of(annotation.userTypeName());
+				DataType userType = dataTypeProvider.getUserType(userTypeName, userTypeResolver);
+
+				if (userType == null) {
+					throw new MappingException(String.format("User type [%s] not found", userTypeName));
+				}
+
+				DataType dataType = getUserDataType(property, userType);
+
+				if (dataType != null) {
+					return dataType;
+				}
+			}
+
 			return property.getDataType();
 		}
 
@@ -496,7 +516,7 @@ public class CassandraMappingContext
 
 		if (persistentEntity != null && persistentEntity.isUserDefinedType()) {
 
-			DataType dataType = getUserDataType(property, dataTypeProvider, persistentEntity);
+			DataType dataType = getUserDataType(property, dataTypeProvider.getDataType(persistentEntity));
 
 			if (dataType != null) {
 				return dataType;
@@ -542,10 +562,7 @@ public class CassandraMappingContext
 	}
 
 	@Nullable
-	private DataType getUserDataType(CassandraPersistentProperty property, DataTypeProvider dataTypeProvider,
-			CassandraPersistentEntity<?> persistentEntity) {
-
-		DataType elementType = dataTypeProvider.getDataType(persistentEntity);
+	private DataType getUserDataType(CassandraPersistentProperty property, DataType elementType) {
 
 		if (property.isCollectionLike()) {
 
@@ -583,6 +600,7 @@ public class CassandraMappingContext
 
 	/**
 	 * @author Jens Schauder
+	 * @author Mark Paluch
 	 * @since 1.5.1
 	 */
 	enum DataTypeProvider {
@@ -593,6 +611,11 @@ public class CassandraMappingContext
 			public DataType getDataType(CassandraPersistentEntity<?> entity) {
 				return entity.getUserType();
 			}
+
+			@Override
+			DataType getUserType(CqlIdentifier userTypeName, UserTypeResolver userTypeResolver) {
+				return userTypeResolver.resolveType(userTypeName);
+			}
 		},
 
 		FrozenLiteral {
@@ -601,6 +624,11 @@ public class CassandraMappingContext
 			public DataType getDataType(CassandraPersistentEntity<?> entity) {
 				return new FrozenLiteralDataType(entity.getTableName());
 			}
+
+			@Override
+			DataType getUserType(CqlIdentifier userTypeName, UserTypeResolver userTypeResolver) {
+				return new FrozenLiteralDataType(userTypeName);
+			}
 		};
 
 		/**
@@ -609,6 +637,18 @@ public class CassandraMappingContext
 		 * @param entity must not be {@literal null}.
 		 * @return the {@link DataType}.
 		 */
+		@Nullable
 		abstract DataType getDataType(CassandraPersistentEntity<?> entity);
+
+		/**
+		 * Return the user-defined type {@code userTypeName}.
+		 *
+		 * @param userTypeName must not be {@literal null}.
+		 * @param userTypeResolver must not be {@literal null}.
+		 * @return the {@link DataType}.
+		 * @since 2.0.1
+		 */
+		@Nullable
+		abstract DataType getUserType(CqlIdentifier userTypeName, UserTypeResolver userTypeResolver);
 	}
 }
diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/mapping/CassandraMappingContextUnitTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/mapping/CassandraMappingContextUnitTests.java
index 5afb48338..7330e8bdc 100755
--- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/mapping/CassandraMappingContextUnitTests.java
+++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/mapping/CassandraMappingContextUnitTests.java
@@ -304,8 +304,8 @@ public class CassandraMappingContextUnitTests {
 	@Test // DATACASS-213
 	public void createIndexShouldConsiderAnnotatedProperties() {
 
-		List<CreateIndexSpecification> specifications = mappingContext.getCreateIndexSpecificationsFor(
-				mappingContext.getRequiredPersistentEntity(IndexedType.class));
+		List<CreateIndexSpecification> specifications = mappingContext
+				.getCreateIndexSpecificationsFor(mappingContext.getRequiredPersistentEntity(IndexedType.class));
 
 		CreateIndexSpecification firstname = getSpecificationFor("first_name", specifications);
 
@@ -473,6 +473,17 @@ public class CassandraMappingContextUnitTests {
 		assertThat(mappingContext.usesUserType(CqlIdentifier.of("mappedudt"))).isTrue();
 	}
 
+	@Test // DATACASS-506
+	public void shouldCreatedUserTypeSpecificationsWithAnnotatedTypeName() {
+
+		assertThat(
+				mappingContext.getCreateUserTypeSpecificationFor(mappingContext.getRequiredPersistentEntity(WithUdt.class)))
+						.isNotNull();
+		assertThat(
+				mappingContext.getCreateUserTypeSpecificationFor(mappingContext.getRequiredPersistentEntity(Nested.class)))
+						.isNotNull();
+	}
+
 	@Test // DATACASS-172
 	public void createTableForComplexPrimaryKeyShouldFail() {
 
@@ -573,6 +584,8 @@ public class CassandraMappingContextUnitTests {
 		@Id String id;
 
 		@CassandraType(type = DataType.Name.UDT, userTypeName = "mappedudt") UDTValue udtValue;
+
+		@CassandraType(type = DataType.Name.UDT, userTypeName = "NestedType") Nested nested;
 	}
 
 	enum HumanToStringConverter implements Converter<Human, String> {
@@ -611,4 +624,16 @@ public class CassandraMappingContextUnitTests {
 			return "serialized";
 		}
 	}
+
+	@UserDefinedType(value = "NestedType")
+	public static class Nested {
+		String s1;
+
+		@CassandraType(type = Name.UDT, userTypeName = "AnotherNestedType") AnotherNested anotherNested;
+	}
+
+	@UserDefinedType(value = "AnotherNestedType")
+	public static class AnotherNested {
+		String str;
+	}
 }
diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/query/ConvertingParameterAccessorUnitTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/query/ConvertingParameterAccessorUnitTests.java
index 3ef469d6f..55d39395f 100755
--- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/query/ConvertingParameterAccessorUnitTests.java
+++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/query/ConvertingParameterAccessorUnitTests.java
@@ -118,12 +118,13 @@ public class ConvertingParameterAccessorUnitTests {
 		assertThat(list.get(0)).isInstanceOf(com.datastax.driver.core.LocalDate.class);
 	}
 
-	@Test // DATACASS-7
+	@Test // DATACASS-7, DATACASS-506
 	@SuppressWarnings({ "rawtypes", "unchecked" })
 	public void shouldProvideTypeBasedOnPropertyType() {
 
 		when(mockProperty.getDataType()).thenReturn(DataType.varchar());
 		when(mockProperty.isAnnotationPresent(CassandraType.class)).thenReturn(true);
+		when(mockProperty.getRequiredAnnotation(CassandraType.class)).thenReturn(mock(CassandraType.class));
 		when(mockParameterAccessor.getParameterType(0)).thenReturn((Class) String.class);
 
 		assertThat(convertingParameterAccessor.getDataType(0, mockProperty)).isEqualTo(DataType.varchar());