diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/mapping/BasicCassandraMappingContext.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/mapping/BasicCassandraMappingContext.java
index 460104c83..8ef2c186d 100644
--- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/mapping/BasicCassandraMappingContext.java
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/mapping/BasicCassandraMappingContext.java
@@ -257,6 +257,51 @@ public class BasicCassandraMappingContext
 		return getTableEntities();
 	}
 
+	/* (non-Javadoc)
+	 * @see org.springframework.data.mapping.context.AbstractMappingContext#addPersistentEntity(org.springframework.data.util.TypeInformation)
+	 */
+	@Override
+	protected Optional<CassandraPersistentEntity<?>> addPersistentEntity(TypeInformation<?> typeInformation) {
+
+		// Prevent conversion types created as CassandraPersistentEntity
+		Optional<CassandraPersistentEntity<?>> optional = shouldCreatePersistentEntityFor(typeInformation)
+				? super.addPersistentEntity(typeInformation) : Optional.empty();
+
+		optional.ifPresent(entity -> {
+
+			if (entity.isUserDefinedType()) {
+				userDefinedTypes.add(entity);
+			}
+			// now do some caching of the entity
+
+			Set<CassandraPersistentEntity<?>> entities = entitySetsByTableName.computeIfAbsent(entity.getTableName(),
+					cqlIdentifier -> new HashSet<>());
+
+			entities.add(entity);
+
+			if (!entity.isUserDefinedType()) {
+				if (entity.isCompositePrimaryKey()) {
+					primaryKeyEntities.add(entity);
+				}
+
+				entity.findAnnotation(Table.class).ifPresent(table -> tableEntities.add(entity));
+			}
+
+			entitiesByType.put(entity.getType(), entity);
+		});
+
+		return optional;
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.mapping.context.AbstractMappingContext#shouldCreatePersistentEntityFor(org.springframework.data.util.TypeInformation)
+	 */
+	@Override
+	protected boolean shouldCreatePersistentEntityFor(TypeInformation<?> typeInfo) {
+		return (!customConversions.hasCustomWriteTarget(typeInfo.getType())
+				&& super.shouldCreatePersistentEntityFor(typeInfo));
+	}
+
 	/* (non-Javadoc)
 	 * @see org.springframework.data.mapping.context.AbstractMappingContext#createPersistentEntity(org.springframework.data.util.TypeInformation)
 	 */
@@ -270,7 +315,7 @@ public class BasicCassandraMappingContext
 
 		if (userDefinedType != null) {
 			entity = new CassandraUserTypePersistentEntity<>(typeInformation, this, verifier, userTypeResolver);
-			userDefinedTypes.add(entity);
+
 		} else {
 			entity = new BasicCassandraPersistentEntity<>(typeInformation, this, verifier);
 		}
@@ -279,23 +324,6 @@ public class BasicCassandraMappingContext
 			entity.setApplicationContext(context);
 		}
 
-		// now do some caching of the entity
-
-		Set<CassandraPersistentEntity<?>> entities = entitySetsByTableName.computeIfAbsent(entity.getTableName(),
-				cqlIdentifier -> new HashSet<>());
-
-		entities.add(entity);
-
-		if (!entity.isUserDefinedType()) {
-			if (entity.isCompositePrimaryKey()) {
-				primaryKeyEntities.add(entity);
-			}
-
-			entity.findAnnotation(Table.class).ifPresent(table -> tableEntities.add(entity));
-		}
-
-		entitiesByType.put(entity.getType(), entity);
-
 		return entity;
 	}
 
@@ -415,25 +443,6 @@ public class BasicCassandraMappingContext
 		return specification;
 	}
 
-	/* (non-Javadoc)
-	 * @see org.springframework.data.mapping.context.AbstractMappingContext#shouldCreatePersistentEntityFor(org.springframework.data.util.TypeInformation)
-	 */
-	@Override
-	protected boolean shouldCreatePersistentEntityFor(TypeInformation<?> typeInfo) {
-		return (!customConversions.hasCustomWriteTarget(typeInfo.getType())
-				&& super.shouldCreatePersistentEntityFor(typeInfo));
-	}
-
-	/* (non-Javadoc)
-	 * @see org.springframework.data.mapping.context.AbstractMappingContext#addPersistentEntity(org.springframework.data.util.TypeInformation)
-	 */
-	@Override
-	protected Optional<CassandraPersistentEntity<?>> addPersistentEntity(TypeInformation<?> typeInformation) {
-		// Prevent conversion types created as CassandraPersistentEntity
-		return (shouldCreatePersistentEntityFor(typeInformation) ? super.addPersistentEntity(typeInformation)
-				: Optional.empty());
-	}
-
 	/* (non-Javadoc)
 	 * @see org.springframework.data.cassandra.mapping.CassandraMappingContext#getDataType(org.springframework.data.cassandra.mapping.CassandraPersistentProperty)
 	 */
diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/mapping/BasicCassandraMappingContextUnitTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/mapping/BasicCassandraMappingContextUnitTests.java
index af21dc638..a61fa3821 100755
--- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/mapping/BasicCassandraMappingContextUnitTests.java
+++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/mapping/BasicCassandraMappingContextUnitTests.java
@@ -37,10 +37,12 @@ import org.springframework.dao.InvalidDataAccessApiUsageException;
 import org.springframework.data.annotation.Id;
 import org.springframework.data.cassandra.convert.CassandraCustomConversions;
 import org.springframework.data.convert.WritingConverter;
+import org.springframework.data.mapping.model.MappingException;
 import org.springframework.data.util.ClassTypeInformation;
 
 import com.datastax.driver.core.DataType;
 import com.datastax.driver.core.DataType.Name;
+import com.datastax.driver.core.TableMetadata;
 import com.datastax.driver.core.UDTValue;
 import com.datastax.driver.core.UserType;
 
@@ -67,14 +69,22 @@ public class BasicCassandraMappingContextUnitTests {
 
 	private static class Transient {}
 
-	@Test
+	@Test // DATACASS-282
 	public void testGetExistingPersistentEntityHappyPath() {
 
+		TableMetadata tableMetadata = mock(TableMetadata.class);
+		when(tableMetadata.getName()).thenReturn(X.class.getSimpleName().toLowerCase());
+
 		mappingContext.getRequiredPersistentEntity(X.class);
 
-		assertThat(mappingContext.contains(X.class)).isTrue();
 		assertThat(mappingContext.getExistingPersistentEntity(X.class)).isNotNull();
 		assertThat(mappingContext.contains(Y.class)).isFalse();
+		assertThat(mappingContext.getNonPrimaryKeyEntities()).hasSize(1);
+		assertThat(mappingContext.getPrimaryKeyEntities()).isEmpty();
+		assertThat(mappingContext.getUserDefinedTypeEntities()).isEmpty();
+		assertThat(mappingContext.getTableEntities()).hasSize(1);
+		assertThat(mappingContext.contains(X.class)).isTrue();
+		assertThat(mappingContext.usesTable(tableMetadata)).isTrue();
 	}
 
 	@Test // DATACASS-248
@@ -353,6 +363,13 @@ public class BasicCassandraMappingContextUnitTests {
 		CassandraPersistentEntity<?> persistentEntity = mappingContext.getRequiredPersistentEntity(MappedUdt.class);
 
 		assertThat(persistentEntity.isUserDefinedType()).isTrue();
+
+		assertThat(mappingContext.getExistingPersistentEntity(MappedUdt.class)).isNotNull();
+		assertThat(mappingContext.getNonPrimaryKeyEntities()).isEmpty();
+		assertThat(mappingContext.getPrimaryKeyEntities()).isEmpty();
+		assertThat(mappingContext.getUserDefinedTypeEntities()).hasSize(1);
+		assertThat(mappingContext.getTableEntities()).hasSize(0);
+		assertThat(mappingContext.contains(MappedUdt.class)).isTrue();
 	}
 
 	@Test // DATACASS-172
@@ -437,6 +454,35 @@ public class BasicCassandraMappingContextUnitTests {
 		}
 	}
 
+	@Test // DATACASS-282
+	public void shouldNotRetainInvalidEntitiesInCache() {
+
+		TableMetadata tableMetadata = mock(TableMetadata.class);
+		when(tableMetadata.getName())
+				.thenReturn(InvalidEntityWithIdAndPrimaryKeyColumn.class.getSimpleName().toLowerCase());
+
+		try {
+			mappingContext.getPersistentEntity(InvalidEntityWithIdAndPrimaryKeyColumn.class);
+			fail("Missing MappingException");
+		} catch (MappingException e) {
+			assertThat(e).isInstanceOf(VerifierMappingExceptions.class);
+		}
+
+		assertThat(mappingContext.getNonPrimaryKeyEntities()).isEmpty();
+		assertThat(mappingContext.getPrimaryKeyEntities()).isEmpty();
+		assertThat(mappingContext.getUserDefinedTypeEntities()).isEmpty();
+		assertThat(mappingContext.getTableEntities()).isEmpty();
+		assertThat(mappingContext.contains(InvalidEntityWithIdAndPrimaryKeyColumn.class)).isFalse();
+		assertThat(mappingContext.usesTable(tableMetadata)).isFalse();
+	}
+
+	@Table
+	private static class InvalidEntityWithIdAndPrimaryKeyColumn {
+
+		@Id String foo;
+		@PrimaryKeyColumn String bar;
+	}
+
 	@Table
 	static class EntityWithComplexPrimaryKeyColumn {
 
@@ -520,4 +566,5 @@ public class BasicCassandraMappingContextUnitTests {
 			return "serialized";
 		}
 	}
+
 }