diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/mapping/BasicCassandraPersistentProperty.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/mapping/BasicCassandraPersistentProperty.java
index 928e6c379..c2e825df9 100644
--- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/mapping/BasicCassandraPersistentProperty.java
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/mapping/BasicCassandraPersistentProperty.java
@@ -32,6 +32,7 @@ import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationContextAware;
 import org.springframework.context.expression.BeanFactoryAccessor;
 import org.springframework.context.expression.BeanFactoryResolver;
+import org.springframework.core.annotation.AnnotatedElementUtils;
 import org.springframework.dao.InvalidDataAccessApiUsageException;
 import org.springframework.data.cassandra.util.SpelUtils;
 import org.springframework.data.mapping.Association;
@@ -106,7 +107,7 @@ public class BasicCassandraPersistentProperty extends AnnotationBasedPersistentP
 
 	@Override
 	public boolean isCompositePrimaryKey() {
-		return getField().getType().isAnnotationPresent(PrimaryKeyClass.class);
+		return AnnotatedElementUtils.findMergedAnnotation(getType(), PrimaryKeyClass.class) != null;
 	}
 
 	public Class<?> getCompositePrimaryKeyType() {
@@ -114,7 +115,7 @@ public class BasicCassandraPersistentProperty extends AnnotationBasedPersistentP
 			return null;
 		}
 
-		return getField().getType();
+		return getType();
 	}
 
 	@Override
@@ -290,7 +291,7 @@ public class BasicCassandraPersistentProperty extends AnnotationBasedPersistentP
 		}
 
 		// else we're dealing with a single-column field
-		String defaultName = getField().getName(); // TODO: replace with naming strategy class
+		String defaultName = getName(); // TODO: replace with naming strategy class
 		String overriddenName = null;
 		boolean forceQuote = false;
 
@@ -396,7 +397,7 @@ public class BasicCassandraPersistentProperty extends AnnotationBasedPersistentP
 
 		if (!isCompositePrimaryKey()) {
 			throw new IllegalStateException(String.format("[%s] does not represent a composite primary key property",
-					getField()));
+					getName()));
 		}
 
 		return getCompositePrimaryKeyEntity().getCompositePrimaryKeyProperties();
diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/mapping/CassandraPersistentPropertyComparator.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/mapping/CassandraPersistentPropertyComparator.java
index c871742e8..c294e96cb 100644
--- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/mapping/CassandraPersistentPropertyComparator.java
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/mapping/CassandraPersistentPropertyComparator.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2014 the original author or authors
+ * Copyright 2013-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.
@@ -15,7 +15,6 @@
  */
 package org.springframework.data.cassandra.mapping;
 
-import java.lang.reflect.Field;
 import java.util.Comparator;
 
 /**
@@ -25,6 +24,7 @@ import java.util.Comparator;
  * 
  * @author Alex Shvid
  * @author Matthew T. Adams
+ * @author Mark Paluch
  */
 public enum CassandraPersistentPropertyComparator implements Comparator<CassandraPersistentProperty> {
 
@@ -60,12 +60,10 @@ public enum CassandraPersistentPropertyComparator implements Comparator<Cassandr
 		boolean leftIsPrimaryKey = left.isPrimaryKeyColumn();
 		boolean rightIsPrimaryKey = right.isPrimaryKeyColumn();
 
-		Field leftField = left.getField();
-		Field rightField = right.getField();
-
+		
 		if (leftIsPrimaryKey && rightIsPrimaryKey) {
-			return CassandraPrimaryKeyColumnAnnotationComparator.IT.compare(leftField.getAnnotation(PrimaryKeyColumn.class),
-					rightField.getAnnotation(PrimaryKeyColumn.class));
+			return CassandraPrimaryKeyColumnAnnotationComparator.IT.compare(left.findAnnotation(PrimaryKeyColumn.class),
+					right.findAnnotation(PrimaryKeyColumn.class));
 		}
 
 		boolean leftIsKey = leftIsCompositePrimaryKey || leftIsPrimaryKey;
@@ -81,11 +79,11 @@ public enum CassandraPersistentPropertyComparator implements Comparator<Cassandr
 
 		// else, neither property is a composite primary key nor a primary key; compare @Column annotations
 
-		Column leftColumn = leftField.getAnnotation(Column.class);
-		Column rightColumn = rightField.getAnnotation(Column.class);
+		Column leftColumn = left.findAnnotation(Column.class);
+		Column rightColumn = right.findAnnotation(Column.class);
 
 		if (leftColumn == null && rightColumn == null) {
-			return leftField.getName().compareTo(rightField.getName());
+			return left.getName().compareTo(right.getName());
 		}
 
 		if (leftColumn != null && rightColumn != null) {
@@ -93,10 +91,10 @@ public enum CassandraPersistentPropertyComparator implements Comparator<Cassandr
 		}
 
 		if (leftColumn != null && rightColumn == null) {
-			return leftColumn.value().compareTo(rightField.getName());
+			return leftColumn.value().compareTo(left.getName());
 		}
 
 		// else leftColumn == null && rightColumn != null)
-		return leftField.getName().compareTo(rightColumn.value());
+		return left.getName().compareTo(rightColumn.value());
 	}
 }
diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/mapping/MappingContextUnitTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/mapping/MappingContextUnitTests.java
new file mode 100644
index 000000000..5415616ec
--- /dev/null
+++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/mapping/MappingContextUnitTests.java
@@ -0,0 +1,213 @@
+/*
+ * 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.mapping;
+
+import static org.hamcrest.Matchers.*;
+import static org.junit.Assert.*;
+
+import java.io.Serializable;
+import java.util.List;
+
+import org.junit.Test;
+import org.springframework.cassandra.core.PrimaryKeyType;
+import org.springframework.cassandra.core.cql.CqlIdentifier;
+import org.springframework.data.mapping.model.MappingException;
+
+/**
+ * Unit tests for {@link BasicCassandraMappingContext}.
+ *
+ * @author Matthew T. Adams
+ * @author Mark Paluch
+ */
+public class MappingContextUnitTests {
+
+	public static class Transient {}
+
+	@Table
+	public static class X {
+		@PrimaryKey String key;
+	}
+
+	@Table
+	public static class Y {
+		@PrimaryKey String key;
+	}
+
+	@Table
+	public static class PrimaryKeyOnProperty {
+
+		String key;
+
+		@PrimaryKey(value = "foo")
+		public String getKey() {
+			return key;
+		}
+
+		public void setKey(String key) {
+			this.key = key;
+		}
+	}
+
+	@Table
+	public static class PrimaryKeyColumnsOnProperty {
+
+		String firstname;
+		String lastname;
+
+		@PrimaryKeyColumn(ordinal = 1, type = PrimaryKeyType.PARTITIONED)
+		public String getFirstname() {
+			return firstname;
+		}
+
+		public void setFirstname(String firstname) {
+			this.firstname = firstname;
+		}
+
+		@PrimaryKeyColumn(name = "mylastname", ordinal = 2, type = PrimaryKeyType.CLUSTERED)
+		public String getLastname() {
+			return lastname;
+		}
+
+		public void setLastname(String lastname) {
+			this.lastname = lastname;
+		}
+	}
+	
+	@Table
+	public static class PrimaryKeyOnPropertyWithPrimaryKeyClass {
+
+		CompositePrimaryKeyClassWithProperties key;
+
+		@PrimaryKey
+		public CompositePrimaryKeyClassWithProperties getKey() {
+			return key;
+		}
+
+		public void setKey(CompositePrimaryKeyClassWithProperties key) {
+			this.key = key;
+		}
+	}
+
+	@PrimaryKeyClass
+	public static class CompositePrimaryKeyClassWithProperties implements Serializable{
+
+		String firstname;
+		String lastname;
+
+		@PrimaryKeyColumn(ordinal = 1, type = PrimaryKeyType.PARTITIONED)
+		public String getFirstname() {
+			return firstname;
+		}
+
+		public void setFirstname(String firstname) {
+			this.firstname = firstname;
+		}
+
+		@PrimaryKeyColumn(name = "mylastname", ordinal = 2, type = PrimaryKeyType.CLUSTERED)
+		public String getLastname() {
+			return lastname;
+		}
+
+		public void setLastname(String lastname) {
+			this.lastname = lastname;
+		}
+	}
+
+	BasicCassandraMappingContext ctx = new BasicCassandraMappingContext();
+
+	@Test(expected = MappingException.class)
+	public void testGetPersistentEntityOfTransientType() {
+
+		CassandraPersistentEntity<?> entity = ctx.getPersistentEntity(Transient.class);
+
+	}
+
+	@Test
+	public void testGetExistingPersistentEntityHappyPath() {
+
+		ctx.getPersistentEntity(X.class);
+
+		assertTrue(ctx.contains(X.class));
+		assertNotNull(ctx.getExistingPersistentEntity(X.class));
+		assertFalse(ctx.contains(Y.class));
+	}
+
+	/**
+	 * @see DATACASS-248
+	 */
+	@Test
+	public void primaryKeyOnPropertyShouldWork() {
+
+		CassandraPersistentEntity<?> persistentEntity = ctx.getPersistentEntity(PrimaryKeyOnProperty.class);
+
+		CassandraPersistentProperty idProperty = persistentEntity.getIdProperty();
+		assertThat(idProperty.getColumnName().toCql(), is(equalTo("foo")));
+
+		List<CqlIdentifier> columnNames = idProperty.getColumnNames();
+		assertThat(columnNames, hasSize(1));
+		assertThat(columnNames.get(0).toCql(), is(equalTo("foo")));
+	}
+
+	/**
+	 * @see DATACASS-248
+	 */
+	@Test
+	public void primaryKeyColumnsOnPropertyShouldWork() {
+
+		CassandraPersistentEntity<?> persistentEntity = ctx.getPersistentEntity(PrimaryKeyColumnsOnProperty.class);
+
+		assertThat(persistentEntity.isCompositePrimaryKey(), is(false));
+
+		CassandraPersistentProperty firstname = persistentEntity.getPersistentProperty("firstname");
+		assertThat(firstname.isCompositePrimaryKey(), is(false));
+		assertThat(firstname.isPrimaryKeyColumn(), is(true));
+		assertThat(firstname.isPartitionKeyColumn(), is(true));
+		assertThat(firstname.getColumnName().toCql(), is(equalTo("firstname")));
+
+		CassandraPersistentProperty lastname = persistentEntity.getPersistentProperty("lastname");
+		assertThat(lastname.isPrimaryKeyColumn(), is(true));
+		assertThat(lastname.isClusterKeyColumn(), is(true));
+		assertThat(lastname.getColumnName().toCql(), is(equalTo("mylastname")));
+	}
+	
+	/**
+	 * @see DATACASS-248
+	 */
+	@Test
+	public void primaryKeyClassWithprimaryKeyColumnsOnPropertyShouldWork() {
+
+		CassandraPersistentEntity<?> persistentEntity = ctx.getPersistentEntity(PrimaryKeyOnPropertyWithPrimaryKeyClass.class);
+		CassandraPersistentEntity<?> primaryKeyClass = ctx.getPersistentEntity(CompositePrimaryKeyClassWithProperties.class);
+
+		assertThat(persistentEntity.isCompositePrimaryKey(), is(false));
+		assertThat(persistentEntity.getPersistentProperty("key").isCompositePrimaryKey(), is(true));
+		
+		assertThat(primaryKeyClass.isCompositePrimaryKey(), is(true));
+		assertThat(primaryKeyClass.getCompositePrimaryKeyProperties(), hasSize(2));
+		
+		CassandraPersistentProperty firstname = primaryKeyClass.getPersistentProperty("firstname");
+		assertThat(firstname.isPrimaryKeyColumn(), is(true));
+		assertThat(firstname.isPartitionKeyColumn(), is(true));
+		assertThat(firstname.isClusterKeyColumn(), is(false));
+		assertThat(firstname.getColumnName().toCql(), is(equalTo("firstname")));
+		
+		CassandraPersistentProperty lastname = primaryKeyClass.getPersistentProperty("lastname");
+		assertThat(lastname.isPrimaryKeyColumn(), is(true));
+		assertThat(lastname.isPartitionKeyColumn(), is(false));
+		assertThat(lastname.isClusterKeyColumn(), is(true));
+		assertThat(lastname.getColumnName().toCql(), is(equalTo("mylastname")));
+	}
+}
diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/test/integration/mappingcontext/MappingContextIntegrationTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/test/integration/mappingcontext/MappingContextIntegrationTests.java
deleted file mode 100644
index 2f201544a..000000000
--- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/test/integration/mappingcontext/MappingContextIntegrationTests.java
+++ /dev/null
@@ -1,63 +0,0 @@
-/*
- * Copyright 2013-2014 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.test.integration.mappingcontext;
-
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertTrue;
-
-import org.junit.Test;
-import org.springframework.data.cassandra.mapping.CassandraPersistentEntity;
-import org.springframework.data.cassandra.mapping.BasicCassandraMappingContext;
-import org.springframework.data.cassandra.mapping.PrimaryKey;
-import org.springframework.data.cassandra.mapping.Table;
-import org.springframework.data.mapping.model.MappingException;
-
-public class MappingContextIntegrationTests {
-
-	public static class Transient {}
-
-	@Table
-	public static class X {
-		@PrimaryKey
-		String key;
-	}
-
-	@Table
-	public static class Y {
-		@PrimaryKey
-		String key;
-	}
-
-	BasicCassandraMappingContext ctx = new BasicCassandraMappingContext();
-
-	@Test(expected = MappingException.class)
-	public void testGetPersistentEntityOfTransientType() {
-
-		CassandraPersistentEntity<?> entity = ctx.getPersistentEntity(Transient.class);
-
-	}
-
-	@Test
-	public void testGetExistingPersistentEntityHappyPath() {
-
-		ctx.getPersistentEntity(X.class);
-
-		assertTrue(ctx.contains(X.class));
-		assertNotNull(ctx.getExistingPersistentEntity(X.class));
-		assertFalse(ctx.contains(Y.class));
-	}
-}