DATACASS-308 - Fix NullPointerException using MapId with unknown property names.

We now reject unknown/misspelled property names when using MapId. The Id conversion and extraction takes now place in the Cassandra Converter. The Id handling code in CassandraTemplate is deprecated.

Original pull request: #73.

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/convert/CassandraConverter.java

@@ -25,12 +25,40 @@ import org.springframework.data.convert.EntityConverter;
  * 
  * @author Alex Shvid
  * @author Matthew T. Adams
+ * @author Mark Paluch
  */
-public interface CassandraConverter extends
-		EntityConverter<CassandraPersistentEntity<?>, CassandraPersistentProperty, Object, Object> {
+public interface CassandraConverter
+		extends EntityConverter<CassandraPersistentEntity<?>, CassandraPersistentProperty, Object, Object> {
 
+	/* (non-Javadoc)
+	 * @see org.springframework.data.convert.EntityConverter#getMappingContext()
+	 */
 	@Override
 	CassandraMappingContext getMappingContext();
 
+	/**
+	 * Returns the Id for an entity. It can return:
+	 * <ul>
+	 * <li>A singular value if for a simple {@link org.springframework.data.annotation.Id} or
+	 * {@link org.springframework.data.cassandra.mapping.PrimaryKey} Id</li>
+	 * <li>A {@link org.springframework.data.cassandra.repository.MapId} for composite
+	 * {@link org.springframework.data.cassandra.mapping.PrimaryKeyColumn} Id's</li>
+	 * <li>A the composite primary key for {@link org.springframework.data.cassandra.mapping.PrimaryKey} using a
+	 * {@link org.springframework.data.cassandra.mapping.PrimaryKeyClass}</li>
+	 * </ul>
+	 * 
+	 * @param object must not be {@literal null}.
+	 * @param entity must not be {@literal null}.
+	 * @return
+	 */
 	Object getId(Object object, CassandraPersistentEntity<?> entity);
+
+	/**
+	 * Converts and writes a {@code source} object into a {@code sink} using the given {@link CassandraPersistentEntity}.
+	 * 
+	 * @param source the source, may be {@literal null}.
+	 * @param sink must not be {@literal null}.
+	 * @param entity must not be {@literal null}.
+	 */
+	void write(Object source, Object sink, CassandraPersistentEntity<?> entity);
 }

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/convert/MappingCassandraConverter.java

@@ -18,8 +18,9 @@ package org.springframework.data.cassandra.convert;
 import static org.springframework.data.cassandra.repository.support.BasicMapId.*;
 
 import java.io.Serializable;
+import java.util.ArrayList;
 import java.util.Collection;
-import java.util.Map;
+import java.util.Collections;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -29,6 +30,7 @@ import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationContextAware;
 import org.springframework.core.CollectionFactory;
 import org.springframework.core.convert.support.DefaultConversionService;
+import org.springframework.dao.InvalidDataAccessApiUsageException;
 import org.springframework.data.cassandra.mapping.BasicCassandraMappingContext;
 import org.springframework.data.cassandra.mapping.CassandraMappingContext;
 import org.springframework.data.cassandra.mapping.CassandraPersistentEntity;
@@ -50,10 +52,14 @@ import org.springframework.util.ClassUtils;
 import org.springframework.util.ObjectUtils;
 
 import com.datastax.driver.core.CodecRegistry;
+import com.datastax.driver.core.DataType;
 import com.datastax.driver.core.Row;
-import com.datastax.driver.core.querybuilder.Delete.Where;
+import com.datastax.driver.core.TypeCodec;
+import com.datastax.driver.core.querybuilder.Clause;
+import com.datastax.driver.core.querybuilder.Delete;
 import com.datastax.driver.core.querybuilder.Insert;
 import com.datastax.driver.core.querybuilder.QueryBuilder;
+import com.datastax.driver.core.querybuilder.Select;
 import com.datastax.driver.core.querybuilder.Update;
 
 /**
@@ -175,7 +181,7 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 	}
 
 	protected void readPropertyFromRow(CassandraPersistentEntity<?> entity, CassandraPersistentProperty property,
-		BasicCassandraRowValueProvider row, PersistentPropertyAccessor propertyAccessor) {
+			BasicCassandraRowValueProvider row, PersistentPropertyAccessor propertyAccessor) {
 
 		// if true then skip; property was set in constructor
 		if (entity.isConstructorArgument(property)) {
@@ -213,10 +219,10 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 
 	@SuppressWarnings("unused")
 	protected Object instantiatePrimaryKey(CassandraPersistentEntity<?> entity, CassandraPersistentProperty keyProperty,
-		BasicCassandraRowValueProvider propertyProvider) {
+			BasicCassandraRowValueProvider propertyProvider) {
 
 		return instantiators.getInstantiatorFor(entity).createInstance(entity,
-			new CassandraPersistentEntityParameterValueProvider(entity, propertyProvider, null));
+				new CassandraPersistentEntityParameterValueProvider(entity, propertyProvider, null));
 	}
 
 	/* (non-Javadoc)
@@ -237,22 +243,33 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 
 		if (source != null) {
 			Class<?> beanClassLoaderClass = transformClassToBeanClassLoaderClass(source.getClass());
-
 			CassandraPersistentEntity<?> entity = mappingContext.getPersistentEntity(beanClassLoaderClass);
 
-			if (entity == null) {
-				throw new MappingException("No mapping metadata found for " + source.getClass());
-			}
+			write(source, sink, entity);
+		}
+	}
 
-			if (sink instanceof Insert) {
-				writeInsertFromObject(source, (Insert) sink, entity);
-			} else if (sink instanceof Update) {
-				writeUpdateFromObject(source, (Update) sink, entity);
-			} else if (sink instanceof Where) {
-				writeDeleteWhereFromObject(source, (Where) sink, entity);
-			} else {
-				throw new MappingException("Unknown write target " + sink.getClass().getName());
-			}
+	@Override
+	public void write(Object source, Object sink, CassandraPersistentEntity<?> entity) {
+
+		if (source == null) {
+			return;
+		}
+
+		if (entity == null) {
+			throw new MappingException("No mapping metadata found for " + source.getClass());
+		}
+
+		if (sink instanceof Insert) {
+			writeInsertFromObject(source, (Insert) sink, entity);
+		} else if (sink instanceof Update) {
+			writeUpdateFromObject(source, (Update) sink, entity);
+		} else if (sink instanceof Select.Where) {
+			writeSelectWhereFromObject(source, (Select.Where) sink, entity);
+		} else if (sink instanceof Delete.Where) {
+			writeDeleteWhereFromObject(source, (Delete.Where) sink, entity);
+		} else {
+			throw new MappingException("Unknown write target " + sink.getClass().getName());
 		}
 	}
 
@@ -261,7 +278,7 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 	}
 
 	protected void writeInsertFromWrapper(final ConvertingPropertyAccessor accessor, final Insert insert,
-		CassandraPersistentEntity<?> entity) {
+			CassandraPersistentEntity<?> entity) {
 
 		entity.doWithProperties(new PropertyHandler<CassandraPersistentProperty>() {
 
@@ -280,7 +297,7 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 					}
 
 					writeInsertFromWrapper(getConvertingAccessor(value, property.getCompositePrimaryKeyEntity()), insert,
-						property.getCompositePrimaryKeyEntity());
+							property.getCompositePrimaryKeyEntity());
 
 					return;
 				}
@@ -323,83 +340,132 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 		});
 	}
 
-	protected void writeDeleteWhereFromObject(final Object object, final Where where,
-		CassandraPersistentEntity<?> entity) {
-		writeDeleteWhereFromWrapper(getConvertingAccessor(object, entity), where, entity);
+	protected void writeSelectWhereFromObject(final Object object, final Select.Where where,
+			CassandraPersistentEntity<?> entity) {
+
+		Collection<Clause> clauses = getWhereClauses(object, entity);
+
+		for (Clause clause : clauses) {
+			where.and(clause);
+		}
 	}
 
-	protected void writeDeleteWhereFromWrapper(final ConvertingPropertyAccessor accessor, final Where where,
-		CassandraPersistentEntity<?> entity) {
+	protected void writeDeleteWhereFromObject(final Object object, final Delete.Where where,
+			CassandraPersistentEntity<?> entity) {
 
-		// if the entity itself if a composite primary key, then we've recursed, so just add columns & return
-		if (entity.isCompositePrimaryKey()) {
-			entity.doWithProperties(new PropertyHandler<CassandraPersistentProperty>() {
-				@Override
-				public void doWithPersistentProperty(CassandraPersistentProperty prop) {
+		Collection<Clause> clauses = getWhereClauses(object, entity);
 
-					Object value = accessor.getProperty(prop,
-						CodecRegistry.DEFAULT_INSTANCE.codecFor(prop.getDataType()).getJavaType().getRawType());
-					where.and(QueryBuilder.eq(prop.getColumnName().toCql(), value));
-				}
-			});
-
-			return;
+		for (Clause clause : clauses) {
+			where.and(clause);
 		}
+	}
 
-		// else, wrapper is an entity with an id
-		Object id = getId(accessor, entity);
+	private Collection<Clause> getWhereClauses(Object source, CassandraPersistentEntity<?> entity) {
+
+		Assert.notNull(source, "Id source must not be null");
+
+		CassandraPersistentProperty idProperty = entity.getIdProperty();
+
+		Object id = extractId(source, entity);
 
 		if (id == null) {
-			String message = String.format("no id value found in object %s", accessor.getBean());
-			log.error(message);
+			String message = String.format("No Id value found in object %s", source);
 			throw new IllegalArgumentException(message);
 		}
 
 		if (id instanceof MapId) {
-
-			for (Map.Entry<String, Serializable> entry : ((MapId) id).entrySet()) {
-
-				CassandraPersistentProperty persistentProperty = entity.getPersistentProperty(entry.getKey());
-				if (persistentProperty != null) {
-					where.and(QueryBuilder.eq(persistentProperty.getColumnName().toCql(), entry.getValue()));
-				} else {
-					where.and(QueryBuilder.eq(entry.getKey(), entry.getValue()));
-				}
-			}
-			return;
+			return getWhereClauses((MapId) id, idProperty != null && idProperty.isCompositePrimaryKey() ? idProperty.getCompositePrimaryKeyEntity() : entity);
 		}
 
-		CassandraPersistentProperty idProperty = entity.getIdProperty();
+		if (idProperty == null) {
+			throw new InvalidDataAccessApiUsageException(
+					String.format("Cannot obtain where clauses for entity [%s] using [%s]", entity.getName(), source));
+		}
 
-		if (idProperty != null) {
+		if (idProperty.isCompositePrimaryKey()) {
 
-			if (idProperty.isCompositePrimaryKey()) {
+			if (ClassUtils.isAssignableValue(idProperty.getType(), id)) {
+				return getWhereClauses(getConvertingAccessor(id, idProperty.getCompositePrimaryKeyEntity()),
+						idProperty.getCompositePrimaryKeyEntity());
+			} else {
+				throw new InvalidDataAccessApiUsageException(
+						String.format("Cannot use [%s] as composite Id for [%s]", id, entity.getName()));
+			}
+		}
 
-				CassandraPersistentEntity<?> idEntity = idProperty.getCompositePrimaryKeyEntity();
+		TypeCodec<Object> codec = getCodec(idProperty);
 
-				writeDeleteWhereFromWrapper(getConvertingAccessor(id, idEntity), where,
-					idProperty.getCompositePrimaryKeyEntity());
+		if(conversionService.canConvert(id.getClass(), codec.getJavaType().getRawType())){
+			return Collections.singleton(QueryBuilder.eq(idProperty.getColumnName().toCql(), conversionService.convert(id, codec.getJavaType().getRawType())));
+		}
 
-				return;
+		return Collections.singleton(QueryBuilder.eq(idProperty.getColumnName().toCql(), id));
+
+	}
+
+	private Object extractId(Object source, CassandraPersistentEntity<?> entity) {
+
+		if (ClassUtils.isAssignableValue(entity.getType(), source)) {
+			return getId(source, entity);
+		} else if (source instanceof MapId) {
+			return source;
+		} else if (source instanceof MapIdentifiable) {
+			return ((MapIdentifiable) source).getMapId();
+		}
+		return source;
+	}
+
+	private Collection<Clause> getWhereClauses(final ConvertingPropertyAccessor accessor,
+			CassandraPersistentEntity<?> entity) {
+
+		Assert.isTrue(entity.isCompositePrimaryKey(),
+				String.format("Entity [%s] is not a composite primary key", entity.getName()));
+
+		final Collection<Clause> clauses = new ArrayList<Clause>();
+		entity.doWithProperties(new PropertyHandler<CassandraPersistentProperty>() {
+			@Override
+			public void doWithPersistentProperty(CassandraPersistentProperty property) {
+
+				TypeCodec<Object> codec = getCodec(property);
+				Object value = accessor.getProperty(property,
+						codec.getJavaType().getRawType());
+				clauses.add(QueryBuilder.eq(property.getColumnName().toCql(), value));
+			}
+		});
+
+		return clauses;
+	}
+
+	private Collection<Clause> getWhereClauses(MapId id, CassandraPersistentEntity<?> entity) {
+
+		Assert.notNull("MapId must not be null");
+
+		Collection<Clause> clauses = new ArrayList<Clause>();
+
+		for (Entry<String, Serializable> entry : id.entrySet()) {
+			CassandraPersistentProperty persistentProperty = entity.getPersistentProperty(entry.getKey());
+			if (persistentProperty == null) {
+				throw new IllegalArgumentException(String.format("MapId contains references [%s] that is an unknown property of [%s]", entry.getKey(), entity.getName()));
 			}
 
-			where.and(QueryBuilder.eq(idProperty.getColumnName().toCql(), id));
+			clauses.add(QueryBuilder.eq(persistentProperty.getColumnName().toCql(), getWriteValue(persistentProperty, entry.getValue())));
 		}
+
+		return clauses;
 	}
 
 	@Override
 	public Object getId(Object object, CassandraPersistentEntity<?> entity) {
 
-		Assert.notNull(object);
+		Assert.notNull(object, "Object instance must not be null");
+		Assert.notNull(entity, "CassandraPersistentEntity must not be null");
 
-		final ConvertingPropertyAccessor wrapper = getConvertingAccessor(object, entity);
-
-		object = wrapper.getBean();
+		final ConvertingPropertyAccessor accessor = getConvertingAccessor(object, entity);
 
 		if (!entity.getType().isAssignableFrom(object.getClass())) {
 			throw new IllegalArgumentException(
-				String.format("given instance of type [%s] is not of compatible expected type [%s]",
-					object.getClass().getName(), entity.getType().getName()));
+					String.format("Given instance of type [%s] is not of compatible expected type [%s]",
+							object.getClass().getName(), entity.getType().getName()));
 		}
 
 		if (object instanceof MapIdentifiable) {
@@ -409,8 +475,8 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 		CassandraPersistentProperty idProperty = entity.getIdProperty();
 
 		if (idProperty != null) {
-			return wrapper.getProperty(entity.getIdProperty(), idProperty.isCompositePrimaryKey() ? idProperty.getType()
-				: CodecRegistry.DEFAULT_INSTANCE.codecFor(idProperty.getDataType()).getJavaType().getRawType());
+			return accessor.getProperty(idProperty, idProperty.isCompositePrimaryKey() ? idProperty.getType()
+					: getCodec(idProperty).getJavaType().getRawType());
 		}
 
 		// if the class doesn't have an id property, then it's using MapId
@@ -419,9 +485,9 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 		entity.doWithProperties(new PropertyHandler<CassandraPersistentProperty>() {
 
 			@Override
-			public void doWithPersistentProperty(CassandraPersistentProperty p) {
-				if (p.isPrimaryKeyColumn()) {
-					id.with(p.getName(), (Serializable) wrapper.getProperty(p, p.getType()));
+			public void doWithPersistentProperty(CassandraPersistentProperty property) {
+				if (property.isPrimaryKeyColumn()) {
+					id.with(property.getName(), (Serializable) getWriteValue(property, accessor));
 				}
 			}
 		});
@@ -492,8 +558,23 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 	 */
 	@SuppressWarnings("unchecked")
 	private Object getWriteValue(CassandraPersistentProperty property, ConvertingPropertyAccessor accessor) {
+		return getWriteValue(property, accessor.getProperty(property, getTargetType(property)));
+	}
 
-		Object value = accessor.getProperty(property, getTargetType(property));
+	/**
+	 * Retrieve the value to write for the given {@link CassandraPersistentProperty} from
+	 * {@link ConvertingPropertyAccessor} and perform optionally a conversion of collection element types.
+	 *
+	 * @param property the property.
+	 * @param value the value
+	 * @return the return value, may be {@literal null}.
+	 */
+	@SuppressWarnings("unchecked")
+	private Object getWriteValue(CassandraPersistentProperty property, Object value) {
+
+		if (value == null) {
+			return value;
+		}
 
 		if (conversions.hasCustomWriteTarget(property.getActualType()) && property.isCollectionLike()) {
 
@@ -548,4 +629,10 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 
 		return obj;
 	}
+
+	private TypeCodec<Object> getCodec(CassandraPersistentProperty property) {
+
+		DataType dataType = mappingContext.getDataType(property);
+		return CodecRegistry.DEFAULT_INSTANCE.codecFor(dataType);
+	}
 }

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraTemplate.java

@@ -160,13 +160,13 @@ public class CassandraTemplate extends CqlTemplate implements CassandraOperation
 	@Override
 	public boolean exists(Class<?> type, Object id) {
 
-		Assert.notNull(type);
-		Assert.notNull(id);
+		Assert.notNull(type, "Type must not be null");
+		Assert.notNull(id, "Id must not be null");
 
-		CassandraPersistentEntity<?> entity = mappingContext.getPersistentEntity(type);
+		CassandraPersistentEntity<?> entity = getPersistentEntity(type);
 
 		Select select = QueryBuilder.select().countAll().from(entity.getTableName().toCql());
-		appendIdCriteria(select.where(), entity, id);
+		cassandraConverter.write(id, select.where(), entity);
 
 		Long count = queryForObject(select, Long.class);
 
@@ -194,10 +194,10 @@ public class CassandraTemplate extends CqlTemplate implements CassandraOperation
 		Assert.notNull(type);
 		Assert.notNull(id);
 
-		CassandraPersistentEntity<?> entity = mappingContext.getPersistentEntity(type);
+		CassandraPersistentEntity<?> entity = getPersistentEntity(type);
 
 		Delete delete = QueryBuilder.delete().from(entity.getTableName().toCql());
-		appendIdCriteria(delete.where(), entity, id);
+		cassandraConverter.write(id, delete.where(), entity);
 
 		execute(delete);
 	}
@@ -398,16 +398,13 @@ public class CassandraTemplate extends CqlTemplate implements CassandraOperation
 	@Override
 	public <T> T selectOneById(Class<T> type, Object id) {
 
-		Assert.notNull(type);
-		Assert.notNull(id);
+		Assert.notNull(type, "Type must not be null");
+		Assert.notNull(id, "Id must not be null");
 
-		CassandraPersistentEntity<?> entity = mappingContext.getPersistentEntity(type);
-		if (entity == null) {
-			throw new IllegalArgumentException(String.format("unknown entity class [%s]", type.getName()));
-		}
+		CassandraPersistentEntity<?> entity = getPersistentEntity(type);
 
 		Select select = QueryBuilder.select().all().from(entity.getTableName().toCql());
-		appendIdCriteria(select.where(), entity, id);
+		cassandraConverter.write(id, select.where(), entity);
 
 		return selectOne(select, type);
 	}
@@ -416,16 +413,24 @@ public class CassandraTemplate extends CqlTemplate implements CassandraOperation
 		void doWithClause(Clause clause);
 	}
 
+	@Deprecated
 	protected void appendIdCriteria(final ClauseCallback clauseCallback, CassandraPersistentEntity<?> entity,
 			final Map<?, ?> id) {
 
 		for (Map.Entry<?, ?> entry : id.entrySet()) {
 
 			CassandraPersistentProperty property = entity.getPersistentProperty(entry.getKey().toString());
+
+			if (property == null) {
+				throw new IllegalArgumentException(String.format("Entity class [%s] has no persistent property named [%s]",
+						entity.getType().getName(), entry.getKey()));
+			}
+
 			clauseCallback.doWithClause(QueryBuilder.eq(property.getColumnName().toCql(), entry.getValue()));
 		}
 	}
 
+	@Deprecated
 	protected void appendIdCriteria(final ClauseCallback clauseCallback, CassandraPersistentEntity<?> entity, Object id) {
 
 		if (id instanceof Map<?, ?>) {
@@ -460,6 +465,7 @@ public class CassandraTemplate extends CqlTemplate implements CassandraOperation
 		clauseCallback.doWithClause(QueryBuilder.eq(idProperty.getColumnName().toCql(), id));
 	}
 
+	@Deprecated
 	protected void appendIdCriteria(final com.datastax.driver.core.querybuilder.Select.Where where,
 			CassandraPersistentEntity<?> entity, Object id) {
 
@@ -472,6 +478,7 @@ public class CassandraTemplate extends CqlTemplate implements CassandraOperation
 		}, entity, id);
 	}
 
+	@Deprecated
 	protected void appendIdCriteria(final Where where, CassandraPersistentEntity<?> entity, Object id) {
 
 		appendIdCriteria(new ClauseCallback() {
@@ -1082,6 +1089,23 @@ public class CassandraTemplate extends CqlTemplate implements CassandraOperation
 		return doSelectOneAsync(cql, type, listener, options);
 	}
 
+	private <T> CassandraPersistentEntity<?> getPersistentEntity(Class<T> entityClass) {
+
+		if (entityClass == null) {
+			throw new InvalidDataAccessApiUsageException(
+					"No class parameter provided, entity collection can't be determined!");
+		}
+
+		CassandraPersistentEntity<?> entity = mappingContext.getPersistentEntity(entityClass);
+
+		if (entity == null) {
+			throw new InvalidDataAccessApiUsageException(
+					String.format("No Persistent Entity information found for the class [%s]", entityClass.getName()));
+		}
+
+		return entity;
+	}
+
 	protected <T> Cancellable doSelectOneAsync(final Object query, final Class<T> type,
 			final QueryForObjectListener<T> listener, QueryOptions options) {
 

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/mapping/BasicCassandraMappingContext.java

@@ -355,7 +355,7 @@ public class BasicCassandraMappingContext
 		CassandraPersistentProperty property = entity.getPersistentProperty(mapping.getPropertyName());
 
 		if (property == null) {
-			throw new IllegalArgumentException(String.format("entity class [%s] has no persistent property named [%s]",
+			throw new IllegalArgumentException(String.format("Entity class [%s] has no persistent property named [%s]",
 					entity.getType().getName(), mapping.getPropertyName()));
 		}
 

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/convert/MappingCassandraConverterUnitTests.java

@@ -23,6 +23,7 @@ import static org.hamcrest.Matchers.startsWith;
 import static org.junit.Assume.*;
 import static org.mockito.Mockito.*;
 import static org.springframework.data.cassandra.RowMockUtil.*;
+import static org.springframework.data.cassandra.repository.support.BasicMapId.*;
 
 import java.io.Serializable;
 import java.math.BigDecimal;
@@ -54,6 +55,11 @@ import org.springframework.cassandra.core.PrimaryKeyType;
 import org.springframework.core.SpringVersion;
 import org.springframework.core.convert.ConverterNotFoundException;
 import org.springframework.data.cassandra.RowMockUtil;
+import org.springframework.data.cassandra.domain.Person;
+import org.springframework.data.cassandra.domain.CompositeKey;
+import org.springframework.data.cassandra.domain.TypeWithCompositeKey;
+import org.springframework.data.cassandra.domain.TypeWithKeyClass;
+import org.springframework.data.cassandra.domain.TypeWithMapId;
 import org.springframework.data.cassandra.domain.UserToken;
 import org.springframework.data.cassandra.mapping.BasicCassandraMappingContext;
 import org.springframework.data.cassandra.mapping.CassandraMappingContext;
@@ -103,7 +109,6 @@ public class MappingCassandraConverterUnitTests {
 		mappingCassandraConverter = new MappingCassandraConverter(mappingContext);
 
 		mappingCassandraConverter.afterPropertiesSet();
-
 	}
 
 	/**
@@ -389,7 +394,7 @@ public class MappingCassandraConverterUnitTests {
 	 * @see DATACASS-280
 	 */
 	@Test
-	public void shouldReadInetAddressCorrectly() throws UnknownHostException{
+	public void shouldReadInetAddressCorrectly() throws UnknownHostException {
 
 		InetAddress localHost = InetAddress.getLocalHost();
 		when(rowMock.getInet(0)).thenReturn(localHost);
@@ -785,6 +790,214 @@ public class MappingCassandraConverterUnitTests {
 		assertThat(getWherePredicates(delete), hasEntry("user_id", (Object) userToken.getUserId()));
 	}
 
+	/**
+	 * @see DATACASS-308
+	 */
+	@Test
+	public void shouldWriteWhereConditionUsingPlainId() {
+
+		Delete delete = QueryBuilder.delete().from("table");
+		mappingCassandraConverter.write("42", delete.where(), mappingContext.getPersistentEntity(Person.class));
+
+		assertThat(getWherePredicates(delete), hasEntry("id", (Object) "42"));
+	}
+
+	/**
+	 * @see DATACASS-308
+	 */
+	@Test
+	public void shouldWriteWhereConditionUsingEntity() {
+
+		Delete delete = QueryBuilder.delete().from("table");
+		Person person = new Person();
+		person.setId("42");
+
+		mappingCassandraConverter.write(person, delete.where(), mappingContext.getPersistentEntity(Person.class));
+
+		assertThat(getWherePredicates(delete), hasEntry("id", (Object) "42"));
+	}
+
+	/**
+	 * @see DATACASS-308
+	 */
+	@Test(expected = IllegalArgumentException.class)
+	public void shouldFailWriteWhereConditionUsingEntityWithNullId() {
+
+		Delete delete = QueryBuilder.delete().from("table");
+
+		mappingCassandraConverter.write(new Person(), delete.where(), mappingContext.getPersistentEntity(Person.class));
+	}
+
+	/**
+	 * @see DATACASS-308
+	 */
+	@Test
+	public void shouldWriteWhereConditionUsingMapId() {
+
+		Delete delete = QueryBuilder.delete().from("table");
+		mappingCassandraConverter.write(id("id", "42"), delete.where(), mappingContext.getPersistentEntity(Person.class));
+
+		assertThat(getWherePredicates(delete), hasEntry("id", (Object) "42"));
+	}
+
+	/**
+	 * @see DATACASS-308
+	 */
+	@Test
+	public void shouldWriteWhereConditionForCompositeKeyUsingEntity() {
+
+		Delete delete = QueryBuilder.delete().from("table");
+
+		TypeWithCompositeKey entity = new TypeWithCompositeKey();
+		entity.setFirstname("Walter");
+		entity.setLastname("White");
+
+		mappingCassandraConverter.write(entity, delete.where(),
+				mappingContext.getPersistentEntity(TypeWithCompositeKey.class));
+
+		assertThat(getWherePredicates(delete), hasEntry("firstname", (Object) "Walter"));
+		assertThat(getWherePredicates(delete), hasEntry("lastname", (Object) "White"));
+	}
+
+	/**
+	 * @see DATACASS-308
+	 */
+	@Test
+	public void shouldWriteWhereConditionForCompositeKeyUsingMapId() {
+
+		Delete delete = QueryBuilder.delete().from("table");
+
+		mappingCassandraConverter.write(id("firstname", "Walter").with("lastname", "White"), delete.where(),
+				mappingContext.getPersistentEntity(TypeWithCompositeKey.class));
+
+		assertThat(getWherePredicates(delete), hasEntry("firstname", (Object) "Walter"));
+		assertThat(getWherePredicates(delete), hasEntry("lastname", (Object) "White"));
+	}
+
+	/**
+	 * @see DATACASS-308
+	 */
+	@Test
+	public void shouldWriteWhereConditionForMapIdKeyUsingEntity() {
+
+		Delete delete = QueryBuilder.delete().from("table");
+
+		TypeWithMapId entity = new TypeWithMapId();
+		entity.setFirstname("Walter");
+		entity.setLastname("White");
+
+		mappingCassandraConverter.write(entity, delete.where(), mappingContext.getPersistentEntity(TypeWithMapId.class));
+
+		assertThat(getWherePredicates(delete), hasEntry("firstname", (Object) "Walter"));
+		assertThat(getWherePredicates(delete), hasEntry("lastname", (Object) "White"));
+	}
+
+	/**
+	 * @see DATACASS-308
+	 */
+	@Test
+	public void shouldWriteEnumWhereCondition() {
+
+		Delete delete = QueryBuilder.delete().from("table");
+
+		mappingCassandraConverter.write(Condition.MINT, delete.where(), mappingContext.getPersistentEntity(EnumPrimaryKey.class));
+
+		assertThat(getWherePredicates(delete), hasEntry("condition", (Object) "MINT"));
+	}
+
+	/**
+	 * @see DATACASS-308
+	 */
+	@Test
+	public void shouldWriteWhereConditionForMapIdKeyUsingMapId() {
+
+		Delete delete = QueryBuilder.delete().from("table");
+
+		mappingCassandraConverter.write(id("firstname", "Walter").with("lastname", "White"), delete.where(),
+				mappingContext.getPersistentEntity(TypeWithMapId.class));
+
+		assertThat(getWherePredicates(delete), hasEntry("firstname", (Object) "Walter"));
+		assertThat(getWherePredicates(delete), hasEntry("lastname", (Object) "White"));
+	}
+
+	/**
+	 * @see DATACASS-308
+	 */
+	@Test
+	public void shouldWriteWhereConditionForTypeWithPkClassKeyUsingEntity() {
+
+		Delete delete = QueryBuilder.delete().from("table");
+
+		CompositeKey key = new CompositeKey();
+		key.setFirstname("Walter");
+		key.setLastname("White");
+
+		TypeWithKeyClass entity = new TypeWithKeyClass();
+		entity.setKey(key);
+
+		mappingCassandraConverter.write(entity, delete.where(), mappingContext.getPersistentEntity(TypeWithKeyClass.class));
+
+		assertThat(getWherePredicates(delete), hasEntry("firstname", (Object) "Walter"));
+		assertThat(getWherePredicates(delete), hasEntry("lastname", (Object) "White"));
+	}
+
+	/**
+	 * @see DATACASS-308
+	 */
+	@Test(expected = IllegalArgumentException.class)
+	public void shouldFailWritingWhereConditionForTypeWithPkClassKeyUsingEntityWithNullId() {
+
+		Delete delete = QueryBuilder.delete().from("table");
+
+		mappingCassandraConverter.write(new TypeWithKeyClass(), delete.where(),
+				mappingContext.getPersistentEntity(TypeWithKeyClass.class));
+	}
+
+	/**
+	 * @see DATACASS-308
+	 */
+	@Test
+	public void shouldWriteWhereConditionForTypeWithPkClassKeyUsingKey() {
+
+		Delete delete = QueryBuilder.delete().from("table");
+
+		CompositeKey key = new CompositeKey();
+		key.setFirstname("Walter");
+		key.setLastname("White");
+
+		mappingCassandraConverter.write(key, delete.where(), mappingContext.getPersistentEntity(TypeWithKeyClass.class));
+
+		assertThat(getWherePredicates(delete), hasEntry("firstname", (Object) "Walter"));
+		assertThat(getWherePredicates(delete), hasEntry("lastname", (Object) "White"));
+	}
+
+	/**
+	 * @see DATACASS-308
+	 */
+	@Test
+	public void shouldWriteWhereConditionForTypeWithPkClassKeyUsingMapId() {
+
+		Delete delete = QueryBuilder.delete().from("table");
+
+		mappingCassandraConverter.write(id("firstname", "Walter").with("lastname", "White"), delete.where(),
+				mappingContext.getPersistentEntity(TypeWithKeyClass.class));
+
+		assertThat(getWherePredicates(delete), hasEntry("firstname", (Object) "Walter"));
+		assertThat(getWherePredicates(delete), hasEntry("lastname", (Object) "White"));
+	}
+
+	/**
+	 * @see DATACASS-308
+	 */
+	@Test(expected = IllegalArgumentException.class)
+	public void shouldFailWhereConditionForTypeWithPkClassKeyUsingMapIdHavingUnknownProperty() {
+
+		Delete delete = QueryBuilder.delete().from("table");
+
+		mappingCassandraConverter.write(id("unknown", "Walter"), delete.where(),
+				mappingContext.getPersistentEntity(TypeWithMapId.class));
+	}
+
 	@SuppressWarnings("unchecked")
 	private <T> List<T> getListValue(Insert statement) {
 		List<Object> values = getValues(statement);
@@ -857,7 +1070,8 @@ public class MappingCassandraConverterUnitTests {
 
 		List<Clause> clauses = (List<Clause>) ReflectionTestUtils.getField(where, "clauses");
 		for (Clause clause : clauses) {
-			result.put((String) ReflectionTestUtils.invokeMethod(clause, "name"), ReflectionTestUtils.getField(clause, "value"));
+			result.put((String) ReflectionTestUtils.invokeMethod(clause, "name"),
+					ReflectionTestUtils.getField(clause, "value"));
 		}
 
 		return result;

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/domain/CompositeKey.java

@@ -0,0 +1,35 @@
+/*
+ * Copyright 2016 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.cassandra.domain;
+
+import java.io.Serializable;
+
+import org.springframework.cassandra.core.PrimaryKeyType;
+import org.springframework.data.cassandra.mapping.PrimaryKeyClass;
+import org.springframework.data.cassandra.mapping.PrimaryKeyColumn;
+
+import lombok.Data;
+
+/**
+ * @author Mark Paluch
+ */
+@PrimaryKeyClass
+@Data
+public class CompositeKey implements Serializable {
+
+	@PrimaryKeyColumn(type = PrimaryKeyType.PARTITIONED, ordinal = 1) private String firstname;
+	@PrimaryKeyColumn(type = PrimaryKeyType.PARTITIONED, ordinal = 2) private String lastname;
+}

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/domain/TypeWithCompositeKey.java

@@ -0,0 +1,33 @@
+/*
+ * Copyright 2016 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.cassandra.domain;
+
+import org.springframework.cassandra.core.PrimaryKeyType;
+import org.springframework.data.cassandra.mapping.PrimaryKeyColumn;
+import org.springframework.data.cassandra.mapping.Table;
+
+import lombok.Data;
+
+/**
+ * @author Mark Paluch
+ */
+@Table
+@Data
+public class TypeWithCompositeKey {
+
+	@PrimaryKeyColumn(type = PrimaryKeyType.PARTITIONED, ordinal = 1) private String firstname;
+	@PrimaryKeyColumn(type = PrimaryKeyType.PARTITIONED, ordinal = 2) private String lastname;
+}

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/domain/TypeWithKeyClass.java

@@ -0,0 +1,31 @@
+/*
+ * Copyright 2016 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.cassandra.domain;
+
+import org.springframework.data.cassandra.mapping.PrimaryKey;
+import org.springframework.data.cassandra.mapping.Table;
+
+import lombok.Data;
+
+/**
+ * @author Mark Paluch
+ */
+@Table
+@Data
+public class TypeWithKeyClass {
+
+	@PrimaryKey CompositeKey key;
+}

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/domain/TypeWithMapId.java

@@ -0,0 +1,41 @@
+/*
+ * Copyright 2016 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.cassandra.domain;
+
+import org.springframework.cassandra.core.PrimaryKeyType;
+import org.springframework.data.cassandra.mapping.PrimaryKeyColumn;
+import org.springframework.data.cassandra.mapping.Table;
+import org.springframework.data.cassandra.repository.MapId;
+import org.springframework.data.cassandra.repository.MapIdentifiable;
+import org.springframework.data.cassandra.repository.support.BasicMapId;
+
+import lombok.Data;
+
+/**
+ * @author Mark Paluch
+ */
+@Table
+@Data
+public class TypeWithMapId implements MapIdentifiable {
+
+	@PrimaryKeyColumn(type = PrimaryKeyType.PARTITIONED, ordinal = 1) private String firstname;
+	@PrimaryKeyColumn(type = PrimaryKeyType.PARTITIONED, ordinal = 2) private String lastname;
+
+	@Override
+	public MapId getMapId() {
+		return BasicMapId.id("firstname", firstname).with("lastname", lastname);
+	}
+}