Add support for property @ValueConverter.

We now support property-specific value converters to apply value conversion for individual properties instead being limited to the type level.

Closes #1449

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/AbstractCassandraConfiguration.java

@@ -16,7 +16,6 @@
 package org.springframework.data.cassandra.config;
 
 import java.util.Arrays;
-import java.util.Collections;
 import java.util.Optional;
 import java.util.Set;
 
@@ -73,8 +72,6 @@ public abstract class AbstractCassandraConfiguration extends AbstractSessionConf
 
 		CqlSession cqlSession = getRequiredSession();
 
-
-
 		MappingCassandraConverter converter = new MappingCassandraConverter(
 				requireBeanOfType(CassandraMappingContext.class));
 
@@ -118,8 +115,6 @@ public abstract class AbstractCassandraConfiguration extends AbstractSessionConf
 
 		CqlSession cqlSession = getRequiredSession();
 
-
-
 		CassandraMappingContext mappingContext = new CassandraMappingContext(userTypeResolver(cqlSession),
 				SimpleTupleTypeFactory.DEFAULT);
 
@@ -181,7 +176,7 @@ public abstract class AbstractCassandraConfiguration extends AbstractSessionConf
 	 */
 	@Bean
 	public CassandraCustomConversions customConversions() {
-		return new CassandraCustomConversions(Collections.emptyList());
+		return CassandraCustomConversions.create(config -> {});
 	}
 
 	/**

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

@@ -0,0 +1,82 @@
+/*
+ * Copyright 2023 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
+ *
+ *      https://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.core.convert;
+
+import org.springframework.data.cassandra.core.mapping.CassandraPersistentProperty;
+import org.springframework.data.convert.ValueConversionContext;
+import org.springframework.data.mapping.model.PropertyValueProvider;
+import org.springframework.data.mapping.model.SpELContext;
+import org.springframework.data.util.TypeInformation;
+import org.springframework.lang.Nullable;
+
+import com.datastax.oss.driver.api.core.cql.Row;
+
+/**
+ * {@link ValueConversionContext} that allows to delegate read/write to an underlying {@link CassandraConverter}.
+ *
+ * @author Mark Paluch
+ * @since 4.2
+ */
+public class CassandraConversionContext implements ValueConversionContext<CassandraPersistentProperty> {
+
+	private final PropertyValueProvider<CassandraPersistentProperty> accessor;
+	private final CassandraPersistentProperty persistentProperty;
+	private final CassandraConverter cassandraConverter;
+
+	@Nullable private final SpELContext spELContext;
+
+	public CassandraConversionContext(PropertyValueProvider<CassandraPersistentProperty> accessor,
+			CassandraPersistentProperty persistentProperty, CassandraConverter CassandraConverter) {
+		this(accessor, persistentProperty, CassandraConverter, null);
+	}
+
+	public CassandraConversionContext(PropertyValueProvider<CassandraPersistentProperty> accessor,
+			CassandraPersistentProperty persistentProperty, CassandraConverter CassandraConverter,
+			@Nullable SpELContext spELContext) {
+
+		this.accessor = accessor;
+		this.persistentProperty = persistentProperty;
+		this.cassandraConverter = CassandraConverter;
+		this.spELContext = spELContext;
+	}
+
+	@Override
+	public CassandraPersistentProperty getProperty() {
+		return persistentProperty;
+	}
+
+	@Nullable
+	public Object getValue(String propertyPath) {
+		return accessor.getPropertyValue(persistentProperty.getOwner().getRequiredPersistentProperty(propertyPath));
+	}
+
+	@Override
+	@SuppressWarnings("unchecked")
+	public <T> T write(@Nullable Object value, TypeInformation<T> target) {
+		return (T) cassandraConverter.convertToColumnType(value, target);
+	}
+
+	@Override
+	public <T> T read(@Nullable Object value, TypeInformation<T> target) {
+		return value instanceof Row row ? cassandraConverter.read(target.getType(), row)
+				: ValueConversionContext.super.read(value, target);
+	}
+
+	@Nullable
+	public SpELContext getSpELContext() {
+		return spELContext;
+	}
+}

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

@@ -16,16 +16,29 @@
 package org.springframework.data.cassandra.core.convert;
 
 import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.Date;
 import java.util.List;
+import java.util.function.Consumer;
 import java.util.function.Predicate;
 
 import org.springframework.core.convert.converter.Converter;
+import org.springframework.core.convert.converter.ConverterFactory;
+import org.springframework.core.convert.converter.GenericConverter;
 import org.springframework.core.convert.converter.GenericConverter.ConvertiblePair;
+import org.springframework.data.cassandra.core.mapping.CassandraPersistentProperty;
 import org.springframework.data.cassandra.core.mapping.CassandraSimpleTypeHolder;
+import org.springframework.data.convert.ConverterBuilder;
 import org.springframework.data.convert.Jsr310Converters;
+import org.springframework.data.convert.PropertyValueConversions;
+import org.springframework.data.convert.PropertyValueConverter;
+import org.springframework.data.convert.PropertyValueConverterFactory;
+import org.springframework.data.convert.PropertyValueConverterRegistrar;
+import org.springframework.data.convert.SimplePropertyValueConversions;
 import org.springframework.data.mapping.model.SimpleTypeHolder;
+import org.springframework.util.Assert;
 
 /**
  * Value object to capture custom conversion. {@link CassandraCustomConversions} also act as factory for
@@ -59,7 +72,33 @@ public class CassandraCustomConversions extends org.springframework.data.convert
 	 * @param converters must not be {@literal null}.
 	 */
 	public CassandraCustomConversions(List<?> converters) {
-		super(new CassandraConverterConfiguration(STORE_CONVERSIONS, converters));
+		super(new CassandraConverterConfiguration(converters));
+	}
+
+	/**
+	 * Create a new {@link CassandraCustomConversions} given {@link CassandraConverterConfigurationAdapter}.
+	 *
+	 * @param conversionConfiguration must not be {@literal null}.
+	 * @since 4.2
+	 */
+	protected CassandraCustomConversions(CassandraConverterConfigurationAdapter conversionConfiguration) {
+		super(conversionConfiguration.createConverterConfiguration());
+	}
+
+	/**
+	 * Functional style {@link org.springframework.data.convert.CustomConversions} creation giving users a convenient way
+	 * of configuring store specific capabilities by providing deferred hooks to what will be configured when creating the
+	 * {@link org.springframework.data.convert.CustomConversions#CustomConversions(ConverterConfiguration) instance}.
+	 *
+	 * @param configurer must not be {@literal null}.
+	 * @since 4.2
+	 */
+	public static CassandraCustomConversions create(Consumer<CassandraConverterConfigurationAdapter> configurer) {
+
+		CassandraConverterConfigurationAdapter adapter = new CassandraConverterConfigurationAdapter();
+		configurer.accept(adapter);
+
+		return new CassandraCustomConversions(adapter);
 	}
 
 	/**
@@ -68,14 +107,185 @@ public class CassandraCustomConversions extends org.springframework.data.convert
 	 */
 	static class CassandraConverterConfiguration extends ConverterConfiguration {
 
-		CassandraConverterConfiguration(StoreConversions storeConversions, List<?> userConverters) {
-			super(storeConversions, userConverters, getConverterFilter());
+		CassandraConverterConfiguration(List<?> converters) {
+			super(STORE_CONVERSIONS, converters, getConverterFilter());
+
+		}
+
+		CassandraConverterConfiguration(List<?> userConverters, PropertyValueConversions propertyValueConversions) {
+			super(STORE_CONVERSIONS, userConverters, getConverterFilter(), propertyValueConversions);
 		}
 
 		static Predicate<ConvertiblePair> getConverterFilter() {
 
 			return convertiblePair -> !(Jsr310Converters.supports(convertiblePair.getSourceType())
-				&& Date.class.isAssignableFrom(convertiblePair.getTargetType()));
+					&& Date.class.isAssignableFrom(convertiblePair.getTargetType()));
+		}
+	}
+
+	/**
+	 * {@link CassandraConverterConfigurationAdapter} encapsulates creation of
+	 * {@link org.springframework.data.convert.CustomConversions.ConverterConfiguration} with Cassandra specifics.
+	 *
+	 * @author Mark Paluch
+	 * @since 4.2
+	 */
+	public static class CassandraConverterConfigurationAdapter {
+
+		private final List<Object> customConverters = new ArrayList<>();
+
+		private final PropertyValueConversions internalValueConversion = PropertyValueConversions.simple(it -> {});
+		private PropertyValueConversions propertyValueConversions = internalValueConversion;
+
+		/**
+		 * Create a {@link CassandraConverterConfigurationAdapter} using the provided {@code converters} and our own codecs
+		 * for JSR-310 types.
+		 *
+		 * @param converters must not be {@literal null}.
+		 * @return
+		 */
+		public static CassandraConverterConfigurationAdapter from(List<?> converters) {
+
+			Assert.notNull(converters, "Converters must not be null");
+
+			CassandraConverterConfigurationAdapter adapter = new CassandraConverterConfigurationAdapter();
+			adapter.registerConverters(converters);
+
+			return adapter;
+		}
+
+		/**
+		 * Add a custom {@link Converter} implementation.
+		 *
+		 * @param converter must not be {@literal null}.
+		 * @return this.
+		 */
+		public CassandraConverterConfigurationAdapter registerConverter(Converter<?, ?> converter) {
+
+			Assert.notNull(converter, "Converter must not be null");
+
+			customConverters.add(converter);
+			return this;
+		}
+
+		/**
+		 * Add a custom {@link ConverterFactory} implementation.
+		 *
+		 * @param converterFactory must not be {@literal null}.
+		 * @return this.
+		 */
+		public CassandraConverterConfigurationAdapter registerConverterFactory(ConverterFactory<?, ?> converterFactory) {
+
+			Assert.notNull(converterFactory, "ConverterFactory must not be null");
+
+			customConverters.add(converterFactory);
+			return this;
+		}
+
+		/**
+		 * Add {@link Converter converters}, {@link ConverterFactory factories}, {@link ConverterBuilder.ConverterAware
+		 * converter-aware objects}, and {@link GenericConverter generic converters}.
+		 *
+		 * @param converters must not be {@literal null} nor contain {@literal null} values.
+		 * @return this.
+		 */
+		public CassandraConverterConfigurationAdapter registerConverters(Object... converters) {
+			return registerConverters(Arrays.asList(converters));
+		}
+
+		/**
+		 * Add {@link Converter converters}, {@link ConverterFactory factories}, {@link ConverterBuilder.ConverterAware
+		 * converter-aware objects}, and {@link GenericConverter generic converters}.
+		 *
+		 * @param converters must not be {@literal null} nor contain {@literal null} values.
+		 * @return this.
+		 */
+		public CassandraConverterConfigurationAdapter registerConverters(Collection<?> converters) {
+
+			Assert.notNull(converters, "Converters must not be null");
+			Assert.noNullElements(converters, "Converters must not be null nor contain null values");
+
+			customConverters.addAll(converters);
+			return this;
+		}
+
+		/**
+		 * Add a custom/default {@link PropertyValueConverterFactory} implementation used to serve
+		 * {@link PropertyValueConverter}.
+		 *
+		 * @param converterFactory must not be {@literal null}.
+		 * @return this.
+		 */
+		public CassandraConverterConfigurationAdapter registerPropertyValueConverterFactory(
+				PropertyValueConverterFactory converterFactory) {
+
+			Assert.state(valueConversions() instanceof SimplePropertyValueConversions,
+					"Configured PropertyValueConversions does not allow setting custom ConverterRegistry");
+
+			((SimplePropertyValueConversions) valueConversions()).setConverterFactory(converterFactory);
+			return this;
+		}
+
+		/**
+		 * Gateway to register property specific converters.
+		 *
+		 * @param configurationAdapter must not be {@literal null}.
+		 * @return this.
+		 */
+		@SuppressWarnings({ "rawtypes", "unchecked" })
+		public CassandraConverterConfigurationAdapter configurePropertyConversions(
+				Consumer<PropertyValueConverterRegistrar<CassandraPersistentProperty>> configurationAdapter) {
+
+			Assert.state(valueConversions() instanceof SimplePropertyValueConversions,
+					"Configured PropertyValueConversions does not allow setting custom ConverterRegistry");
+
+			PropertyValueConverterRegistrar propertyValueConverterRegistrar = new PropertyValueConverterRegistrar();
+			configurationAdapter.accept(propertyValueConverterRegistrar);
+
+			((SimplePropertyValueConversions) valueConversions())
+					.setValueConverterRegistry(propertyValueConverterRegistrar.buildRegistry());
+			return this;
+		}
+
+		/**
+		 * Optionally set the {@link PropertyValueConversions} to be applied during mapping.
+		 * <p>
+		 * Use this method if {@link #configurePropertyConversions(Consumer)} and
+		 * {@link #registerPropertyValueConverterFactory(PropertyValueConverterFactory)} are not sufficient.
+		 *
+		 * @param valueConversions must not be {@literal null}.
+		 * @return this.
+		 */
+		public CassandraConverterConfigurationAdapter withPropertyValueConversions(
+				PropertyValueConversions valueConversions) {
+
+			Assert.notNull(valueConversions, "PropertyValueConversions must not be null");
+
+			this.propertyValueConversions = valueConversions;
+			return this;
+		}
+
+		PropertyValueConversions valueConversions() {
+
+			if (this.propertyValueConversions == null) {
+				this.propertyValueConversions = internalValueConversion;
+			}
+
+			return this.propertyValueConversions;
+		}
+
+		CassandraConverterConfiguration createConverterConfiguration() {
+
+			if (hasDefaultPropertyValueConversions()
+					&& propertyValueConversions instanceof SimplePropertyValueConversions svc) {
+				svc.init();
+			}
+
+			return new CassandraConverterConfiguration(this.customConverters, this.propertyValueConversions);
+		}
+
+		private boolean hasDefaultPropertyValueConversions() {
+			return propertyValueConversions == internalValueConversion;
 		}
 	}
 }

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

@@ -0,0 +1,43 @@
+/*
+ * Copyright 2022-2023 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
+ *
+ *      https://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.core.convert;
+
+import org.springframework.data.convert.PropertyValueConverter;
+
+/**
+ * Cassandra-specific {@link PropertyValueConverter} extension. Converters can implement this interface for
+ * Cassandra-specific value conversions, for example:
+ *
+ * <pre class="code">
+ * static class MyJsonConverter implements CassandraValueConverter<Person, String> {
+ *
+ * 	&#64;Override
+ * 	public Person read(String value, CassandraConversionContext context) {
+ * 		return // decode JSON to Person object
+ * 	}
+ *
+ * 	&#64;Override
+ * 	public String write(Person value, CassandraConversionContext context) {
+ * 		return // marshal Person to JSON
+ * 	}
+ * }
+ * </pre>
+ *
+ * @author Mark Paluch
+ * @since 4.2
+ * @see org.springframework.data.convert.ValueConverter
+ */
+public interface CassandraValueConverter<S, T> extends PropertyValueConverter<S, T, CassandraConversionContext> {}

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

@@ -23,7 +23,7 @@ import org.springframework.data.mapping.model.SimpleTypeHolder;
  * Value object to capture custom conversion. That is essentially a {@link List} of converters and some additional logic
  * around them. The converters are pretty much builds up two sets of types which Cassandra basic types can be converted
  * into and from. These types will be considered simple ones (which means they neither need deeper inspection nor nested
- * conversion. Thus the {@link CustomConversions} also act as factory for {@link SimpleTypeHolder}
+ * conversion). Thus, the {@link CustomConversions} also act as factory for {@link SimpleTypeHolder}
  *
  * @author Mark Paluch
  * @since 1.5

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

@@ -28,6 +28,7 @@ import java.util.stream.StreamSupport;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.springframework.core.ResolvableType;
 import org.springframework.core.annotation.AnnotatedElementUtils;
 import org.springframework.dao.InvalidDataAccessApiUsageException;
 import org.springframework.data.cassandra.core.mapping.CassandraPersistentEntity;
@@ -38,6 +39,9 @@ import org.springframework.data.cassandra.core.mapping.CassandraType.Name;
 import org.springframework.data.cassandra.core.mapping.Frozen;
 import org.springframework.data.cassandra.core.mapping.UserTypeResolver;
 import org.springframework.data.convert.CustomConversions;
+import org.springframework.data.convert.PropertyValueConversions;
+import org.springframework.data.convert.PropertyValueConverter;
+import org.springframework.data.convert.ValueConversionContext;
 import org.springframework.data.mapping.MappingException;
 import org.springframework.data.mapping.context.MappingContext;
 import org.springframework.data.util.Lazy;
@@ -140,7 +144,29 @@ class DefaultColumnTypeResolver implements ColumnTypeResolver {
 			return resolve(annotation);
 		}
 
+		PropertyValueConversions pvc = customConversions.get().getPropertyValueConversions();
 		TypeInformation<?> typeInformation = property.getTypeInformation();
+
+		if (pvc != null && pvc.hasValueConverter(property)) {
+
+			PropertyValueConverter<Object, Object, ValueConversionContext<CassandraPersistentProperty>> converter = pvc
+					.getValueConverter(property);
+			ResolvableType resolvableType = ResolvableType.forClass(converter.getClass());
+			ResolvableType storeType = resolvableType.as(PropertyValueConverter.class).getGeneric(1);
+			Class<?> storeTypeClass = storeType.resolve();
+
+			if (storeTypeClass == Object.class || storeTypeClass == null) {
+
+				if (log.isDebugEnabled()) {
+					log.debug(String.format(
+							"PropertyValueConverter %s for Property %s.%s resolves to Object.class. Falling back to the property type %s.",
+							converter, property.getOwner().getName(), property.getName(), typeInformation));
+				}
+			} else {
+				typeInformation = TypeInformation.of(storeType);
+			}
+		}
+
 		return resolve(typeInformation, getFrozenInfo(property));
 	}
 
@@ -195,9 +221,8 @@ class DefaultColumnTypeResolver implements ColumnTypeResolver {
 
 		return getCustomWriteTarget(typeInformation)
 				.map(it -> createCassandraTypeDescriptor(tryResolve(it), TypeInformation.of(it)))
-			.orElseGet(() -> typeInformation.getType().isEnum()
-				? ColumnType.create(String.class, DataTypes.TEXT)
-				: createCassandraTypeDescriptor(typeInformation, frozen));
+				.orElseGet(() -> typeInformation.getType().isEnum() ? ColumnType.create(String.class, DataTypes.TEXT)
+						: createCassandraTypeDescriptor(typeInformation, frozen));
 	}
 
 	private Optional<Class<?>> getCustomWriteTarget(TypeInformation<?> typeInformation) {
@@ -234,10 +259,8 @@ class DefaultColumnTypeResolver implements ColumnTypeResolver {
 			case MAP:
 				assertTypeArguments(annotation.typeArguments().length, 2);
 
-				CassandraColumnType keyType = createCassandraTypeDescriptor(
-						getRequiredDataType(annotation, 0));
-				CassandraColumnType valueType = createCassandraTypeDescriptor(
-						getRequiredDataType(annotation, 1));
+				CassandraColumnType keyType = createCassandraTypeDescriptor(getRequiredDataType(annotation, 0));
+				CassandraColumnType valueType = createCassandraTypeDescriptor(getRequiredDataType(annotation, 1));
 
 				return ColumnType.mapOf(keyType, valueType);
 
@@ -442,9 +465,8 @@ class DefaultColumnTypeResolver implements ColumnTypeResolver {
 
 		DataType dataType = tryResolve(typeInformation.getType());
 
-		return dataType == null
-			? new UnresolvableCassandraType(typeInformation)
-			: new DefaultCassandraColumnType(typeInformation, dataType);
+		return dataType == null ? new UnresolvableCassandraType(typeInformation)
+				: new DefaultCassandraColumnType(typeInformation, dataType);
 	}
 
 	private DataType getRequiredDataType(CassandraType annotation, int typeIndex) {

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

@@ -50,6 +50,7 @@ import org.springframework.data.mapping.model.ConvertingPropertyAccessor;
 import org.springframework.data.mapping.model.DefaultSpELExpressionEvaluator;
 import org.springframework.data.mapping.model.EntityInstantiator;
 import org.springframework.data.mapping.model.ParameterValueProvider;
+import org.springframework.data.mapping.model.PropertyValueProvider;
 import org.springframework.data.mapping.model.SpELContext;
 import org.springframework.data.mapping.model.SpELExpressionEvaluator;
 import org.springframework.data.mapping.model.SpELExpressionParameterValueProvider;
@@ -257,7 +258,8 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 	 * @see org.springframework.data.cassandra.core.mapping.CassandraPersistentEntity
 	 */
 	@SuppressWarnings("unchecked")
-	private <S> ConvertingPropertyAccessor<S> newConvertingPropertyAccessor(S source, CassandraPersistentEntity<?> entity) {
+	private <S> ConvertingPropertyAccessor<S> newConvertingPropertyAccessor(S source,
+			CassandraPersistentEntity<?> entity) {
 		return new ConvertingPropertyAccessor<>(entity.getPropertyAccessor(source), getConversionService());
 	}
 
@@ -850,9 +852,20 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 	private <T> T getWriteValue(CassandraPersistentProperty property, ConvertingPropertyAccessor<?> propertyAccessor) {
 
 		ColumnType cassandraTypeDescriptor = cassandraTypeResolver.resolve(property);
+		Object value = propertyAccessor.getProperty(property, cassandraTypeDescriptor.getType());
 
-		return (T) getWriteValue(propertyAccessor.getProperty(property, cassandraTypeDescriptor.getType()),
-				cassandraTypeDescriptor);
+		if (getCustomConversions().hasValueConverter(property)) {
+			return (T) getCustomConversions().getPropertyValueConversions().getValueConverter(property).write(value,
+					new CassandraConversionContext(new PropertyValueProvider<>() {
+						@Nullable
+						@Override
+						public <T> T getPropertyValue(CassandraPersistentProperty property) {
+							return (T) propertyAccessor.getProperty(property);
+						}
+					}, property, this, spELContext));
+		}
+
+		return (T) getWriteValue(value, cassandraTypeDescriptor);
 	}
 
 	/**
@@ -1081,6 +1094,11 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 
 		Object value = valueProvider.getPropertyValue(property);
 
+		if (getCustomConversions().hasValueConverter(property)) {
+			return getCustomConversions().getPropertyValueConversions().getValueConverter(property).read(value,
+					new CassandraConversionContext(valueProvider, property, this, spELContext));
+		}
+
 		return value == null ? null : context.convert(value, property.getTypeInformation());
 	}
 

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

@@ -21,6 +21,7 @@ import java.util.HashSet;
 import java.util.List;
 import java.util.Optional;
 import java.util.Set;
+import java.util.stream.Collectors;
 
 import org.springframework.data.cassandra.core.mapping.BasicCassandraPersistentEntity;
 import org.springframework.data.cassandra.core.mapping.CassandraMappingContext;
@@ -36,6 +37,8 @@ import org.springframework.data.cassandra.core.query.Criteria;
 import org.springframework.data.cassandra.core.query.CriteriaDefinition;
 import org.springframework.data.cassandra.core.query.CriteriaDefinition.Predicate;
 import org.springframework.data.cassandra.core.query.Filter;
+import org.springframework.data.convert.PropertyValueConverter;
+import org.springframework.data.convert.ValueConversionContext;
 import org.springframework.data.domain.Sort;
 import org.springframework.data.domain.Sort.Order;
 import org.springframework.data.mapping.PersistentProperty;
@@ -44,6 +47,7 @@ import org.springframework.data.mapping.PropertyHandler;
 import org.springframework.data.mapping.PropertyPath;
 import org.springframework.data.mapping.PropertyReferenceException;
 import org.springframework.data.mapping.context.MappingContext;
+import org.springframework.data.mapping.model.PropertyValueProvider;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
@@ -130,13 +134,8 @@ public class QueryMapper {
 			});
 
 			Predicate predicate = criteriaDefinition.getPredicate();
-
 			Object value = predicate.getValue();
-
-			ColumnType typeDescriptor = getColumnType(field, value, ColumnTypeTransformer.of(field, predicate.getOperator()));
-
-			Object mappedValue = value != null ? getConverter().convertToColumnType(value, typeDescriptor) : null;
-
+			Object mappedValue = value != null ? getMappedValue(field, predicate, value) : null;
 			Predicate mappedPredicate = new Predicate(predicate.getOperator(), mappedValue);
 
 			result.add(Criteria.of(field.getMappedKey(), mappedPredicate));
@@ -145,6 +144,39 @@ public class QueryMapper {
 		return Filter.from(result);
 	}
 
+	@Nullable
+	private Object getMappedValue(Field field, Predicate predicate, Object value) {
+
+		if (field.getProperty().isPresent()
+				&& field.getProperty().filter(it -> converter.getCustomConversions().hasValueConverter(it)).isPresent()) {
+
+			CassandraPersistentProperty property = field.getProperty().get();
+			CassandraConversionContext conversionContext = new CassandraConversionContext(new PropertyValueProvider<>() {
+				@Override
+				public <T> T getPropertyValue(CassandraPersistentProperty property) {
+					throw new IllegalStateException("No enclosing property available");
+				}
+			}, property, converter);
+
+			PropertyValueConverter<Object, Object, ValueConversionContext<CassandraPersistentProperty>> valueConverter = converter
+					.getCustomConversions().getPropertyValueConversions().getValueConverter(property);
+
+			/* might be an $in clause with multiple entries */
+			if (!property.isCollectionLike() && value instanceof List<?> collection) {
+				return collection.stream().map(it -> valueConverter.write(it, conversionContext)).toList();
+			}
+
+			if (!property.isCollectionLike() && value instanceof Set<?> collection) {
+				return collection.stream().map(it -> valueConverter.write(it, conversionContext)).collect(Collectors.toSet());
+			}
+
+			return valueConverter.write(value, conversionContext);
+		}
+
+		ColumnType typeDescriptor = getColumnType(field, value, ColumnTypeTransformer.of(field, predicate.getOperator()));
+		return getConverter().convertToColumnType(value, typeDescriptor);
+	}
+
 	/**
 	 * Map {@link Columns} with a {@link CassandraPersistentEntity type hint} to {@link ColumnSelector}s.
 	 *

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

@@ -27,12 +27,16 @@ import org.assertj.core.api.SoftAssertions;
 import org.junit.jupiter.api.Test;
 import org.springframework.data.cassandra.core.mapping.BasicCassandraPersistentEntity;
 import org.springframework.data.cassandra.core.mapping.CassandraMappingContext;
+import org.springframework.data.cassandra.core.mapping.CassandraPersistentEntity;
 import org.springframework.data.cassandra.core.mapping.CassandraPersistentProperty;
 import org.springframework.data.cassandra.core.mapping.CassandraType;
 import org.springframework.data.cassandra.core.mapping.Frozen;
 import org.springframework.data.cassandra.core.mapping.UserDefinedType;
+import org.springframework.data.convert.PropertyValueConverter;
+import org.springframework.data.convert.ValueConverter;
 import org.springframework.data.mapping.MappingException;
 import org.springframework.data.util.TypeInformation;
+import org.springframework.lang.Nullable;
 
 import com.datastax.oss.driver.api.core.data.TupleValue;
 import com.datastax.oss.driver.api.core.type.DataType;
@@ -322,6 +326,59 @@ public class ColumnTypeResolverUnitTests {
 		});
 	}
 
+	@Test // GH-1449
+	void shouldConsiderDeclarativePropertyValueConverter() {
+
+		CassandraPersistentEntity<?> entity = mappingContext
+				.getRequiredPersistentEntity(TypeWithPropertyValueConverters.class);
+
+		DataType dataType = resolver.resolve(entity.getRequiredPersistentProperty("declarative")).getDataType();
+		assertThat(dataType).isEqualTo(DataTypes.INT);
+	}
+
+	@Test // GH-1449
+	void shouldConsiderLambdaPropertyValueConverterFallbackToPropertyType() {
+
+		CassandraPersistentEntity<?> entity = mappingContext
+				.getRequiredPersistentEntity(TypeWithPropertyValueConverters.class);
+
+		CassandraCustomConversions conversions = CassandraCustomConversions.create(adapter -> {
+
+			adapter.configurePropertyConversions(registrar -> {
+
+				registrar.registerConverter(TypeWithPropertyValueConverters.class, "programmatic", String.class)
+						.writing((from, ctx) -> from.length()).reading((from, ctx) -> from.toString());
+			});
+		});
+
+		resolver = new DefaultColumnTypeResolver(mappingContext, SchemaFactory.ShallowUserTypeResolver.INSTANCE,
+				() -> CodecRegistry.DEFAULT, () -> conversions);
+
+		DataType dataType = resolver.resolve(entity.getRequiredPersistentProperty("programmatic")).getDataType();
+		assertThat(dataType).isEqualTo(DataTypes.TEXT);
+	}
+
+	@Test // GH-1449
+	void shouldConsiderRegisteredPropertyValueConverter() {
+
+		CassandraPersistentEntity<?> entity = mappingContext
+				.getRequiredPersistentEntity(TypeWithPropertyValueConverters.class);
+
+		CassandraCustomConversions conversions = CassandraCustomConversions.create(adapter -> {
+
+			adapter.configurePropertyConversions(registrar -> {
+				registrar.registerConverter(TypeWithPropertyValueConverters.class, "programmatic",
+						new CharacterCountingConverter());
+			});
+		});
+
+		resolver = new DefaultColumnTypeResolver(mappingContext, SchemaFactory.ShallowUserTypeResolver.INSTANCE,
+				() -> CodecRegistry.DEFAULT, () -> conversions);
+
+		DataType dataType = resolver.resolve(entity.getRequiredPersistentProperty("programmatic")).getDataType();
+		assertThat(dataType).isEqualTo(DataTypes.INT);
+	}
+
 	static class Person {
 
 		String name;
@@ -377,6 +434,30 @@ public class ColumnTypeResolverUnitTests {
 		@CassandraType(type = CassandraType.Name.TIMEUUID) UUID timeUUID;
 	}
 
+	private static class TypeWithPropertyValueConverters {
+
+		@ValueConverter(CharacterCountingConverter.class) String declarative;
+
+		String programmatic;
+
+	}
+
+	static class CharacterCountingConverter
+			implements PropertyValueConverter<String, Integer, CassandraConversionContext> {
+
+		@Nullable
+		@Override
+		public String read(Integer value, CassandraConversionContext context) {
+			return value.toString();
+		}
+
+		@Nullable
+		@Override
+		public Integer write(String value, CassandraConversionContext context) {
+			return value.length();
+		}
+	}
+
 	enum MyEnum {
 		INSTANCE;
 	}

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

@@ -53,15 +53,18 @@ import org.springframework.data.cassandra.domain.User;
 import org.springframework.data.cassandra.domain.UserToken;
 import org.springframework.data.cassandra.support.UserDefinedTypeBuilder;
 import org.springframework.data.cassandra.test.util.RowMockUtil;
+import org.springframework.data.convert.ValueConverter;
 import org.springframework.data.projection.EntityProjection;
 import org.springframework.data.projection.EntityProjectionIntrospector;
 import org.springframework.data.projection.SpelAwareProxyProjectionFactory;
+import org.springframework.lang.Nullable;
 import org.springframework.util.ObjectUtils;
 
 import com.datastax.oss.driver.api.core.CqlIdentifier;
 import com.datastax.oss.driver.api.core.cql.Row;
 import com.datastax.oss.driver.api.core.data.UdtValue;
 import com.datastax.oss.driver.api.core.type.DataTypes;
+import com.datastax.oss.driver.api.core.type.TupleType;
 import com.datastax.oss.driver.internal.core.data.DefaultTupleValue;
 import com.datastax.oss.driver.internal.core.type.DefaultTupleType;
 
@@ -816,6 +819,96 @@ public class MappingCassandraConverterUnitTests {
 		assertThat(map.get("Europe/Paris")).hasOnlyElementsOfType(LocalDate.class);
 	}
 
+	@Test // GH-1449
+	void shouldConsiderPropertyValueConverterOnRowWrite() {
+
+		TypeWithPropertyValueConverter toInsert = new TypeWithPropertyValueConverter();
+		toInsert.name = "Walter";
+		toInsert.other = "Some other value";
+		toInsert.tuple = new TupleWithConverter();
+		toInsert.tuple.name = "Heisenberg";
+		toInsert.tuple.other = "Mike";
+
+		Map<CqlIdentifier, Object> object = new LinkedHashMap<>();
+		mappingCassandraConverter.write(toInsert, object);
+
+		assertThat(object).containsEntry(CqlIdentifier.fromCql("name"), "Other: Some other value, reversed: retlaW");
+		assertThat(object).containsEntry(CqlIdentifier.fromCql("other"), "Some other value");
+		assertThat(object).containsKey(CqlIdentifier.fromCql("tuple"));
+
+		DefaultTupleValue tupleValue = (DefaultTupleValue) object.get(CqlIdentifier.fromCql("tuple"));
+		assertThat(tupleValue.getString(0)).isEqualTo("Other: Mike, reversed: grebnesieH");
+		assertThat(tupleValue.getString(1)).isEqualTo("Mike");
+	}
+
+	@Test // GH-1449
+	void shouldConsiderPropertyValueConverterOnRowRead() {
+
+		TupleType tupleType = DataTypes.tupleOf(DataTypes.TEXT, DataTypes.TEXT);
+		DefaultTupleValue tupleValue = new DefaultTupleValue(tupleType);
+		tupleValue.setString(0, "grebnesieH");
+		tupleValue.setString(1, "Mike");
+
+		rowMock = RowMockUtil.newRowMock(RowMockUtil.column("name", "retlaW", DataTypes.TEXT),
+				RowMockUtil.column("other", "Some other value", DataTypes.TEXT),
+				RowMockUtil.column("tuple", tupleValue, tupleType));
+
+		TypeWithPropertyValueConverter result = mappingCassandraConverter.read(TypeWithPropertyValueConverter.class,
+				rowMock);
+
+		assertThat(result.name).isEqualTo("Other: Some other value, reversed: Walter");
+		assertThat(result.other).isEqualTo("Some other value");
+		assertThat(result.tuple).isNotNull();
+		assertThat(result.tuple.name).isEqualTo("Other: Mike, reversed: Heisenberg");
+		assertThat(result.tuple.other).isEqualTo("Mike");
+	}
+
+	@Test // GH-1449
+	void shouldConsiderProgrammaticConverterRead() {
+
+		CassandraCustomConversions conversions = CassandraCustomConversions.create(adapter -> {
+
+			adapter.configurePropertyConversions(registrar -> {
+
+				registrar.registerConverter(AllPossibleTypes.class, "id", String.class)
+						.writing((from, ctx) -> from.toUpperCase()).reading((from, ctx) -> from.toLowerCase());
+			});
+		});
+
+		MappingCassandraConverter converter = new MappingCassandraConverter(mappingContext);
+		converter.setCustomConversions(conversions);
+
+		rowMock = RowMockUtil.newRowMock(RowMockUtil.column("id", "WALTER", DataTypes.TEXT));
+
+		AllPossibleTypes result = converter.read(AllPossibleTypes.class, rowMock);
+
+		assertThat(result.getId()).isEqualTo("walter");
+	}
+
+	@Test // GH-1449
+	void shouldConsiderProgrammaticConverterWrite() {
+
+		CassandraCustomConversions conversions = CassandraCustomConversions.create(adapter -> {
+
+			adapter.configurePropertyConversions(registrar -> {
+
+				registrar.registerConverter(AllPossibleTypes.class, "id", String.class)
+						.writing((from, ctx) -> from.toUpperCase()).reading((from, ctx) -> from.toLowerCase());
+			});
+		});
+
+		MappingCassandraConverter converter = new MappingCassandraConverter(mappingContext);
+		converter.setCustomConversions(conversions);
+
+		AllPossibleTypes apt = new AllPossibleTypes();
+		apt.setId("walter");
+
+		Map<CqlIdentifier, Object> object = new LinkedHashMap<>();
+		converter.write(apt, object);
+
+		assertThat(object).containsEntry(CqlIdentifier.fromCql("id"), "WALTER");
+	}
+
 	@Test // DATACASS-189
 	void writeShouldSkipTransientProperties() {
 
@@ -1180,6 +1273,49 @@ public class MappingCassandraConverterUnitTests {
 		private Map<ZoneId, List<java.time.LocalDate>> times;
 	}
 
+	private static class TypeWithPropertyValueConverter {
+
+		@ValueConverter(ReversingValueConverter.class) private String name;
+
+		private String other;
+
+		private TupleWithConverter tuple;
+	}
+
+	@Tuple
+	private static class TupleWithConverter {
+
+		@Element(0)
+		@ValueConverter(ReversingValueConverter.class) private String name;
+
+		@Element(1) private String other;
+
+	}
+
+	static class ReversingValueConverter implements CassandraValueConverter<String, String> {
+
+		@Nullable
+		@Override
+		public String read(@Nullable String value, CassandraConversionContext context) {
+			return String.format("Other: %s, reversed: %s", context.getValue("other"), reverse(value));
+		}
+
+		@Nullable
+		@Override
+		public String write(@Nullable String value, CassandraConversionContext context) {
+			return String.format("Other: %s, reversed: %s", context.getValue("other"), reverse(value));
+		}
+
+		private String reverse(String source) {
+
+			if (source == null) {
+				return null;
+			}
+
+			return new StringBuilder(source).reverse().toString();
+		}
+	}
+
 	private static class WithTransient {
 
 		@Id String id;

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

@@ -55,9 +55,12 @@ import org.springframework.data.cassandra.core.query.Filter;
 import org.springframework.data.cassandra.core.query.Query;
 import org.springframework.data.cassandra.domain.TypeWithKeyClass;
 import org.springframework.data.cassandra.support.UserDefinedTypeBuilder;
+import org.springframework.data.convert.PropertyValueConverter;
+import org.springframework.data.convert.ValueConverter;
 import org.springframework.data.domain.Sort;
 import org.springframework.data.domain.Sort.Direction;
 import org.springframework.data.domain.Sort.Order;
+import org.springframework.lang.Nullable;
 
 import com.datastax.oss.driver.api.core.CqlIdentifier;
 import com.datastax.oss.driver.api.core.data.TupleValue;
@@ -274,6 +277,17 @@ public class QueryMapperUnitTests {
 		assertThat(mappedCriteriaDefinition.getColumnName().toString()).isEqualTo("first_name");
 	}
 
+	@Test // GH-1449
+	void shouldConsiderPropertyValueConverter() {
+
+		Query query = Query.query(Criteria.where("reverseName").is("Heisenberg"));
+
+		Filter mappedObject = queryMapper.getMappedObject(query, personPersistentEntity);
+
+		CriteriaDefinition mappedCriteriaDefinition = mappedObject.iterator().next();
+		assertThat(mappedCriteriaDefinition.getPredicate().getValue()).isEqualTo("grebnesieH");
+	}
+
 	@Test // DATACASS-343
 	void shouldCreateSelectExpression() {
 
@@ -437,6 +451,32 @@ public class QueryMapperUnitTests {
 
 		@Column("first_name") String firstName;
 
+		@ValueConverter(ReversingValueConverter.class) String reverseName;
+
+	}
+
+	static class ReversingValueConverter implements PropertyValueConverter<String, String, CassandraConversionContext> {
+
+		@Nullable
+		@Override
+		public String read(@Nullable String value, CassandraConversionContext context) {
+			return reverse(value);
+		}
+
+		@Nullable
+		@Override
+		public String write(@Nullable String value, CassandraConversionContext context) {
+			return reverse(value);
+		}
+
+		private String reverse(String source) {
+
+			if (source == null) {
+				return null;
+			}
+
+			return new StringBuilder(source).reverse().toString();
+		}
 	}
 
 	static class WithPrimaryKeyClass {

src/main/antora/modules/ROOT/examples/ConverterConfiguration.java

@@ -30,12 +30,10 @@ public class ConverterConfiguration extends AbstractCassandraConfiguration {
 	@Override
 	public CassandraCustomConversions customConversions() {
 
-		List<Converter<?, ?>> converters = new ArrayList<>();
-
-		converters.add(new PersonReadConverter());
-		converters.add(new PersonWriteConverter());
-
-		return new CassandraCustomConversions(converters);
+		return CassandraCustomConversions.create(config -> {
+			config.registerConverter(new PersonReadConverter()));
+			config.registerConverter(new PersonWriteConverter()));
+		});
 	}
 
 	// other methods omitted...

src/main/antora/modules/ROOT/examples/ReversingValueConverter.java

@@ -0,0 +1,43 @@
+/*
+ * Copyright 2020-2023 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
+ *
+ *      https://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.example;
+
+// tag::class[]
+class ReversingValueConverter implements PropertyValueConverter<String, String, ValueConversionContext> {
+
+	@Override
+	public String read(String value, ValueConversionContext context) {
+		return reverse(value);
+	}
+
+	@Override
+	public String write(String value, ValueConversionContext context) {
+		return reverse(value);
+	}
+
+	// end::class[]
+	private String reverse(String source) {
+
+		if (source == null) {
+			return null;
+		}
+
+		return new StringBuilder(source).reverse().toString();
+	}
+	// tag::class[]
+}
+// end::class[]

src/main/antora/modules/ROOT/examples/SchemaConfiguration.java

@@ -38,12 +38,10 @@ public class SchemaConfiguration extends AbstractCassandraConfiguration {
 	@Override
 	public CassandraCustomConversions customConversions() {
 
-		List<Converter<?, ?>> converters = new ArrayList<>();
-
-		converters.add(new PersonReadConverter());
-		converters.add(new PersonWriteConverter());
-
-		return new CassandraCustomConversions(converters);
+		return CassandraCustomConversions.create(config -> {
+			config.registerConverter(new PersonReadConverter()));
+			config.registerConverter(new PersonWriteConverter()));
+		});
 	}
 
 	@Override

src/main/antora/modules/ROOT/nav.adoc

@@ -14,7 +14,8 @@
 ** xref:cassandra/template.adoc[]
 ** xref:cassandra/prepared-statements.adoc[]
 ** xref:object-mapping.adoc[]
-** xref:cassandra/converters.adoc[]
+** xref:cassandra/converters.adoc[Type-based Converter]
+** xref:cassandra/property-converters.adoc[]
 ** xref:cassandra/events.adoc[]
 ** xref:cassandra/auditing.adoc[]
 

src/main/antora/modules/ROOT/pages/cassandra/property-converters.adoc

@@ -0,0 +1,96 @@
+[[cassandra.property-converters]]
+= Property-based Converters
+
+While xref:cassandra/converters.adoc[type-based conversion] already offers ways to influence the conversion and representation of certain types within the target store, it has limitations when only certain values or properties of a particular type should be considered for conversion.
+Property-based converters allow configuring conversion rules on a per-property basis, either declaratively (via `@ValueConverter`) or programmatically (by registering a `PropertyValueConverter` for a specific property).
+
+A `PropertyValueConverter` can transform a given value into its store representation (write) and back (read) as the following listing shows.
+The additional `ValueConversionContext` provides additional information, such as mapping metadata and direct `read` and `write` methods.
+
+.A simple `PropertyValueConverter`
+====
+[source,java]
+----
+include::example$ReversingValueConverter.java[tags=class]
+----
+====
+
+You can obtain `PropertyValueConverter` instances from `CustomConversions#getPropertyValueConverter(…)` by delegating to `PropertyValueConversions`, typically by using a `PropertyValueConverterFactory` to provide the actual converter.
+Depending on your application's needs, you can chain or decorate multiple instances of `PropertyValueConverterFactory` -- for example, to apply caching.
+By default, Spring Data Cassandra uses a caching implementation that can serve types with a default constructor or enum values.
+A set of predefined factories is available through the factory methods in `PropertyValueConverterFactory`.
+You can use `PropertyValueConverterFactory.beanFactoryAware(…)` to obtain a `PropertyValueConverter` instance from an `ApplicationContext`.
+
+You can change the default behavior through `ConverterConfiguration`.
+
+[[cassandra.property-converters.declarative]]
+== Declarative Value Converter
+
+The most straight forward usage of a `PropertyValueConverter` is by annotating properties with the `@ValueConverter` annotation that defines the converter type:
+
+.Declarative PropertyValueConverter
+====
+[source,java]
+----
+class Person {
+
+  @ValueConverter(ReversingValueConverter.class)
+  String ssn;
+}
+----
+====
+
+[[cassandra.property-converters.programmatic]]
+== Programmatic Value Converter Registration
+
+Programmatic registration registers `PropertyValueConverter` instances for properties within an entity model by using a `PropertyValueConverterRegistrar`, as the following example shows.
+The difference between declarative registration and programmatic registration is that programmatic registration happens entirely outside the entity model.
+Such an approach is useful if you cannot or do not want to annotate the entity model.
+
+.Programmatic PropertyValueConverter registration
+====
+[source,java]
+----
+PropertyValueConverterRegistrar registrar = new PropertyValueConverterRegistrar();
+
+registrar.registerConverter(Address.class, "street", new PropertyValueConverter() { … }); <1>
+
+// type safe registration
+registrar.registerConverter(Person.class, Person::getSsn())                               <2>
+  .writing(value -> encrypt(value))
+  .reading(value -> decrypt(value));
+----
+
+<1> Register a converter for the field identified by its name.
+<2> Type safe variant that allows to register a converter and its conversion functions.
+This method uses class proxies to determine the property.
+Make sure that neither the class nor the accessors are `final` as otherwise this approach doesn't work.
+====
+
+WARNING: Dot notation (such as `registerConverter(Person.class, "address.street", …)`) for nagivating across properties into nested objects is *not* supported when registering converters.
+
+WARNING: Schema derivation can only derive the column type from a registered converter if the converter is a `PropertyValueConverter` class.
+Generics cannot be determined from lambdas and using a lambda will fall back to the property type.
+
+TIP: `CassandraValueConverter` offers a pre-typed `PropertyValueConverter` interface that uses `CassandraConversionContext`.
+
+[[cassandra.conversions.-configuration]]
+== CassandraCustomConversions configuration
+
+By default, `CassandraCustomConversions` can handle declarative value converters, depending on the configured `PropertyValueConverterFactory`.
+`CassandraConverterConfigurationAdapter` helps you to set up programmatic value conversions or define the `PropertyValueConverterFactory` to be used or to register converters.
+
+.Configuration Sample
+====
+[source,java]
+----
+CassandraCustomConversions conversions = CassandraCustomConversions.create(adapter -> {
+  adapter.registerConverter(…);
+  adapter.configurePropertyConversions(registrar -> {
+    registrar.registerConverter(Person.class, "name", String.class)
+        .writing((from, ctx) -> …)
+        .reading((from, ctx) -> …);
+  });
+});
+----
+====