diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/convert/DefaultColumnTypeResolver.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/convert/DefaultColumnTypeResolver.java
index 87670f0e8..1fb1b2f74 100644
--- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/convert/DefaultColumnTypeResolver.java
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/convert/DefaultColumnTypeResolver.java
@@ -22,6 +22,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.function.Supplier;
 import java.util.stream.StreamSupport;
 
@@ -76,6 +77,8 @@ class DefaultColumnTypeResolver implements ColumnTypeResolver {
 	private final UserTypeResolver userTypeResolver;
 	private final Supplier<CodecRegistry> codecRegistry;
 	private final Supplier<CustomConversions> customConversions;
+	private final Map<CassandraPersistentProperty, CassandraColumnType> columnTypeCache = new ConcurrentHashMap<>();
+	private final Map<TypeInformation<?>, CassandraColumnType> typeInformationColumnTypeCache = new ConcurrentHashMap<>();
 
 	/**
 	 * Creates a new {@link DefaultColumnTypeResolver}.
@@ -100,6 +103,18 @@ class DefaultColumnTypeResolver implements ColumnTypeResolver {
 
 		Assert.notNull(property, "Property must not be null");
 
+		CassandraColumnType cassandraColumnType = columnTypeCache.get(property);
+		if (cassandraColumnType == null) {
+			// avoid recursive update
+			cassandraColumnType = doResolve(property);
+			columnTypeCache.put(property, cassandraColumnType);
+		}
+
+		return cassandraColumnType;
+	}
+
+	private CassandraColumnType doResolve(CassandraPersistentProperty property) {
+
 		if (property.isAnnotationPresent(CassandraType.class)) {
 
 			CassandraType annotation = property.getRequiredAnnotation(CassandraType.class);
@@ -165,7 +180,17 @@ class DefaultColumnTypeResolver implements ColumnTypeResolver {
 
 	@Override
 	public CassandraColumnType resolve(TypeInformation<?> typeInformation) {
-		return resolve(typeInformation, FrozenIndicator.NOT_FROZEN);
+
+		Assert.notNull(typeInformation, "TypeInformation must not be null");
+
+		CassandraColumnType cassandraColumnType = typeInformationColumnTypeCache.get(typeInformation);
+		if (cassandraColumnType == null) {
+			// avoid recursive update
+			cassandraColumnType = resolve(typeInformation, FrozenIndicator.NOT_FROZEN);
+			typeInformationColumnTypeCache.put(typeInformation, cassandraColumnType);
+		}
+
+		return cassandraColumnType;
 	}
 
 	private CassandraColumnType resolve(TypeInformation<?> typeInformation, FrozenIndicator frozen) {
diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/convert/MappingCassandraConverter.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/convert/MappingCassandraConverter.java
index 349aca839..7157427cb 100644
--- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/convert/MappingCassandraConverter.java
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/convert/MappingCassandraConverter.java
@@ -1031,7 +1031,8 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 	@SuppressWarnings({ "rawtypes", "unchecked" })
 	private Object getPotentiallyConvertedSimpleRead(Object value, @Nullable Class<?> target) {
 
-		if (value == null || target == null || target.isAssignableFrom(value.getClass())) {
+		if (value == null || target == null
+				|| ClassUtils.resolvePrimitiveIfNecessary(target).isAssignableFrom(value.getClass())) {
 			return value;
 		}
 
diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/BasicCassandraPersistentProperty.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/BasicCassandraPersistentProperty.java
index 28d325ea3..b1d526d7d 100644
--- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/BasicCassandraPersistentProperty.java
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/BasicCassandraPersistentProperty.java
@@ -105,7 +105,9 @@ public class BasicCassandraPersistentProperty extends AnnotationBasedPersistentP
 			this.columnName = determineColumnName();
 		}
 
-		Assert.state(this.columnName != null, () -> String.format("Cannot determine column name for %s", this));
+		if (columnName == null) {
+			throw new IllegalStateException(String.format("Cannot determine column name for %s", this));
+		}
 
 		return this.columnName;
 	}
diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/CachingCassandraPersistentProperty.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/CachingCassandraPersistentProperty.java
index 384b51bd0..74d47f1f8 100644
--- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/CachingCassandraPersistentProperty.java
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/CachingCassandraPersistentProperty.java
@@ -15,6 +15,12 @@
  */
 package org.springframework.data.cassandra.core.mapping;
 
+import java.lang.annotation.Annotation;
+import java.lang.reflect.AnnotatedType;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.ConcurrentHashMap;
+
 import org.springframework.data.cassandra.core.cql.Ordering;
 import org.springframework.data.mapping.model.Property;
 import org.springframework.data.mapping.model.SimpleTypeHolder;
@@ -36,6 +42,7 @@ public class CachingCassandraPersistentProperty extends BasicCassandraPersistent
 	private final boolean isPrimaryKeyColumn;
 	private final boolean isEmbedded;
 	private final boolean isStaticColumn;
+	private final Map<Class<? extends Annotation>, Optional<AnnotatedType>> findAnnotatedTypeCache = new ConcurrentHashMap<>();
 
 	public CachingCassandraPersistentProperty(Property property, CassandraPersistentEntity<?> owner,
 			SimpleTypeHolder simpleTypeHolder) {
@@ -85,4 +92,10 @@ public class CachingCassandraPersistentProperty extends BasicCassandraPersistent
 	public boolean isEmbedded() {
 		return isEmbedded;
 	}
+
+	@Override
+	public AnnotatedType findAnnotatedType(Class<? extends Annotation> annotationType) {
+		return findAnnotatedTypeCache
+				.computeIfAbsent(annotationType, key -> Optional.ofNullable(super.findAnnotatedType(key))).orElse(null);
+	}
 }