DATACASS-618 - Integrate EntityCallbacks.

We now support EntityCallbacks via DefaultEntityCallbacks and DefaultReactiveEntityCallbacks.

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

@@ -21,6 +21,9 @@ import java.util.List;
 import java.util.function.Consumer;
 import java.util.function.Function;
 
+import org.springframework.beans.BeansException;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.ApplicationContextAware;
 import org.springframework.context.ApplicationEvent;
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.context.ApplicationEventPublisherAware;
@@ -41,16 +44,20 @@ import org.springframework.data.cassandra.core.cql.CqlIdentifier;
 import org.springframework.data.cassandra.core.cql.CqlProvider;
 import org.springframework.data.cassandra.core.cql.GuavaListenableFutureAdapter;
 import org.springframework.data.cassandra.core.cql.QueryOptions;
+import org.springframework.data.cassandra.core.cql.WriteOptions;
 import org.springframework.data.cassandra.core.cql.session.DefaultSessionFactory;
 import org.springframework.data.cassandra.core.mapping.CassandraPersistentEntity;
 import org.springframework.data.cassandra.core.mapping.event.AfterConvertEvent;
 import org.springframework.data.cassandra.core.mapping.event.AfterDeleteEvent;
 import org.springframework.data.cassandra.core.mapping.event.AfterLoadEvent;
 import org.springframework.data.cassandra.core.mapping.event.AfterSaveEvent;
+import org.springframework.data.cassandra.core.mapping.event.BeforeConvertCallback;
 import org.springframework.data.cassandra.core.mapping.event.BeforeDeleteEvent;
+import org.springframework.data.cassandra.core.mapping.event.BeforeSaveCallback;
 import org.springframework.data.cassandra.core.mapping.event.BeforeSaveEvent;
 import org.springframework.data.cassandra.core.query.Query;
 import org.springframework.data.domain.Slice;
+import org.springframework.data.mapping.callback.EntityCallbacks;
 import org.springframework.data.projection.ProjectionFactory;
 import org.springframework.data.projection.SpelAwareProxyProjectionFactory;
 import org.springframework.lang.Nullable;
@@ -89,7 +96,8 @@ import com.datastax.driver.core.querybuilder.Update;
  * @see org.springframework.data.cassandra.core.AsyncCassandraOperations
  * @since 2.0
  */
-public class AsyncCassandraTemplate implements AsyncCassandraOperations, ApplicationEventPublisherAware {
+public class AsyncCassandraTemplate
+		implements AsyncCassandraOperations, ApplicationEventPublisherAware, ApplicationContextAware {
 
 	private final AsyncCqlOperations cqlOperations;
 
@@ -105,6 +113,8 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations, Applica
 
 	private @Nullable ApplicationEventPublisher eventPublisher;
 
+	private @Nullable EntityCallbacks entityCallbacks;
+
 	/**
 	 * Creates an instance of {@link AsyncCassandraTemplate} initialized with the given {@link Session} and a default
 	 * {@link MappingCassandraConverter}.
@@ -176,6 +186,29 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations, Applica
 		this.eventPublisher = applicationEventPublisher;
 	}
 
+	/* (non-Javadoc)
+	 * @see org.springframework.context.ApplicationContextAware(org.springframework.context.ApplicationContext)
+	 */
+	@Override
+	public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
+
+		if (entityCallbacks == null) {
+			setEntityCallbacks(EntityCallbacks.create(applicationContext));
+		}
+
+		projectionFactory.setBeanFactory(applicationContext);
+		projectionFactory.setBeanClassLoader(applicationContext.getClassLoader());
+	}
+
+	/**
+	 * Configure {@link EntityCallbacks} to pre-/post-process entities during persistence operations.
+	 *
+	 * @param entityCallbacks
+	 */
+	public void setEntityCallbacks(@Nullable EntityCallbacks entityCallbacks) {
+		this.entityCallbacks = entityCallbacks;
+	}
+
 	/* (non-Javadoc)
 	 * @see org.springframework.data.cassandra.core.AsyncCassandraOperations#getAsyncCqlOperations()
 	 */
@@ -541,9 +574,14 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations, Applica
 		Assert.notNull(entity, "Entity must not be null");
 		Assert.notNull(options, "InsertOptions must not be null");
 
-		AdaptibleEntity<T> source = getEntityOperations().forEntity(entity, getConverter().getConversionService());
+		return doInsert(entity, options, getTableName(entity.getClass()));
+	}
+
+	private <T> ListenableFuture<EntityWriteResult<T>> doInsert(T entity, WriteOptions options, CqlIdentifier tableName) {
+
+		AdaptibleEntity<T> source = getEntityOperations().forEntity(maybeCallBeforeConvert(entity, tableName),
+				getConverter().getConversionService());
 		CassandraPersistentEntity<?> persistentEntity = getRequiredPersistentEntity(entity.getClass());
-		CqlIdentifier tableName = persistentEntity.getTableName();
 
 		T entityToUse = source.isVersionedEntity() ? source.initializeVersionProperty() : entity;
 
@@ -595,24 +633,26 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations, Applica
 		CassandraPersistentEntity<?> persistentEntity = getRequiredPersistentEntity(entity.getClass());
 		CqlIdentifier tableName = persistentEntity.getTableName();
 
-		return source.isVersionedEntity() ? doUpdateVersioned(source, options, tableName, persistentEntity)
-				: doUpdate(entity, options, tableName, persistentEntity);
+		T entityToUpdate = maybeCallBeforeConvert(entity, tableName);
+
+		return source.isVersionedEntity() ? doUpdateVersioned(entityToUpdate, options, tableName, persistentEntity)
+				: doUpdate(entityToUpdate, options, tableName, persistentEntity);
 	}
 
-	private <T> ListenableFuture<EntityWriteResult<T>> doUpdateVersioned(AdaptibleEntity<T> source, UpdateOptions options,
+	private <T> ListenableFuture<EntityWriteResult<T>> doUpdateVersioned(T entity, UpdateOptions options,
 			CqlIdentifier tableName, CassandraPersistentEntity<?> persistentEntity) {
 
+		AdaptibleEntity<T> source = getEntityOperations().forEntity(entity, getConverter().getConversionService());
 		Number previousVersion = source.getVersion();
+		T toSave = source.incrementVersion();
 
-		T entity = source.incrementVersion();
+		Update update = getStatementFactory().update(toSave, options, getConverter(), persistentEntity, tableName);
 
-		Update update = getStatementFactory().update(entity, options, getConverter(), persistentEntity, tableName);
-
-		return executeSave(entity, tableName, source.appendVersionCondition(update, previousVersion), result -> {
+		return executeSave(toSave, tableName, source.appendVersionCondition(update, previousVersion), result -> {
 
 			if (!result.wasApplied()) {
 				throw new OptimisticLockingFailureException(
-						String.format("Cannot save entity %s with version %s to table %s. Has it been modified meanwhile?", entity,
+						String.format("Cannot save entity %s with version %s to table %s. Has it been modified meanwhile?", toSave,
 								source.getVersion(), tableName));
 			}
 		});
@@ -726,16 +766,17 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations, Applica
 			Consumer<WriteResult> beforeAfterSaveEvent) {
 
 		maybeEmitEvent(new BeforeSaveEvent<>(entity, tableName, statement));
+		T entityToSave = maybeCallBeforeSave(entity, tableName, statement);
 
 		ListenableFuture<ResultSet> result = getAsyncCqlOperations().execute(new AsyncStatementCallback(statement));
 
 		return new MappingListenableFutureAdapter<>(result, resultSet -> {
 
-			EntityWriteResult<T> writeResult = EntityWriteResult.of(resultSet, entity);
+			EntityWriteResult<T> writeResult = EntityWriteResult.of(resultSet, entityToSave);
 
 			beforeAfterSaveEvent.accept(writeResult);
 
-			maybeEmitEvent(new AfterSaveEvent<>(entity, tableName));
+			maybeEmitEvent(new AfterSaveEvent<>(entityToSave, tableName));
 
 			return writeResult;
 		});
@@ -825,6 +866,24 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations, Applica
 		}
 	}
 
+	protected <T> T maybeCallBeforeConvert(T object, CqlIdentifier tableName) {
+
+		if (null != entityCallbacks) {
+			return (T) entityCallbacks.callback(BeforeConvertCallback.class, object, tableName);
+		}
+
+		return object;
+	}
+
+	protected <T> T maybeCallBeforeSave(T object, CqlIdentifier tableName, Statement statement) {
+
+		if (null != entityCallbacks) {
+			return (T) entityCallbacks.callback(BeforeSaveCallback.class, object, tableName, statement);
+		}
+
+		return object;
+	}
+
 	static class MappingListenableFutureAdapter<T, S>
 			extends org.springframework.util.concurrent.ListenableFutureAdapter<T, S> {
 

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

@@ -23,6 +23,9 @@ import java.util.function.Function;
 import java.util.stream.Stream;
 import java.util.stream.StreamSupport;
 
+import org.springframework.beans.BeansException;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.ApplicationContextAware;
 import org.springframework.context.ApplicationEvent;
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.context.ApplicationEventPublisherAware;
@@ -49,11 +52,14 @@ import org.springframework.data.cassandra.core.mapping.event.AfterConvertEvent;
 import org.springframework.data.cassandra.core.mapping.event.AfterDeleteEvent;
 import org.springframework.data.cassandra.core.mapping.event.AfterLoadEvent;
 import org.springframework.data.cassandra.core.mapping.event.AfterSaveEvent;
+import org.springframework.data.cassandra.core.mapping.event.BeforeConvertCallback;
 import org.springframework.data.cassandra.core.mapping.event.BeforeDeleteEvent;
+import org.springframework.data.cassandra.core.mapping.event.BeforeSaveCallback;
 import org.springframework.data.cassandra.core.mapping.event.BeforeSaveEvent;
 import org.springframework.data.cassandra.core.query.Columns;
 import org.springframework.data.cassandra.core.query.Query;
 import org.springframework.data.domain.Slice;
+import org.springframework.data.mapping.callback.EntityCallbacks;
 import org.springframework.data.mapping.context.MappingContext;
 import org.springframework.data.projection.ProjectionFactory;
 import org.springframework.data.projection.SpelAwareProxyProjectionFactory;
@@ -92,10 +98,12 @@ import com.datastax.driver.core.querybuilder.Update;
  * @see org.springframework.data.cassandra.core.CassandraOperations
  * @since 2.0
  */
-public class CassandraTemplate implements CassandraOperations, ApplicationEventPublisherAware {
+public class CassandraTemplate implements CassandraOperations, ApplicationEventPublisherAware, ApplicationContextAware {
 
 	private @Nullable ApplicationEventPublisher eventPublisher;
 
+	private @Nullable EntityCallbacks entityCallbacks;
+
 	private final CassandraConverter converter;
 
 	private final CqlOperations cqlOperations;
@@ -187,6 +195,29 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 		this.eventPublisher = applicationEventPublisher;
 	}
 
+	/* (non-Javadoc)
+	 * @see org.springframework.context.ApplicationContextAware(org.springframework.context.ApplicationContext)
+	 */
+	@Override
+	public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
+
+		if (entityCallbacks == null) {
+			setEntityCallbacks(EntityCallbacks.create(applicationContext));
+		}
+
+		projectionFactory.setBeanFactory(applicationContext);
+		projectionFactory.setBeanClassLoader(applicationContext.getClassLoader());
+	}
+
+	/**
+	 * Configure {@link EntityCallbacks} to pre-/post-process entities during persistence operations.
+	 *
+	 * @param entityCallbacks
+	 */
+	public void setEntityCallbacks(@Nullable EntityCallbacks entityCallbacks) {
+		this.entityCallbacks = entityCallbacks;
+	}
+
 	/* (non-Javadoc)
 	 * @see org.springframework.data.cassandra.core.CassandraOperations#getConverter()
 	 */
@@ -603,7 +634,8 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 
 	<T> EntityWriteResult<T> doInsert(T entity, WriteOptions options, CqlIdentifier tableName) {
 
-		AdaptibleEntity<T> source = getEntityOperations().forEntity(entity, getConverter().getConversionService());
+		AdaptibleEntity<T> source = getEntityOperations().forEntity(maybeCallBeforeConvert(entity, tableName),
+				getConverter().getConversionService());
 		CassandraPersistentEntity<?> persistentEntity = getRequiredPersistentEntity(entity.getClass());
 
 		T entityToUse = source.isVersionedEntity() ? source.initializeVersionProperty() : entity;
@@ -653,24 +685,27 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 		CassandraPersistentEntity<?> persistentEntity = getRequiredPersistentEntity(entity.getClass());
 		CqlIdentifier tableName = persistentEntity.getTableName();
 
-		return source.isVersionedEntity() ? doUpdateVersioned(source, options, tableName, persistentEntity)
-				: doUpdate(entity, options, tableName, persistentEntity);
+		T entityToUpdate = maybeCallBeforeConvert(entity, tableName);
+
+		return source.isVersionedEntity() ? doUpdateVersioned(entityToUpdate, options, tableName, persistentEntity)
+				: doUpdate(entityToUpdate, options, tableName, persistentEntity);
 	}
 
-	private <T> EntityWriteResult<T> doUpdateVersioned(AdaptibleEntity<T> source, UpdateOptions options,
-			CqlIdentifier tableName, CassandraPersistentEntity<?> persistentEntity) {
+	private <T> EntityWriteResult<T> doUpdateVersioned(T entity, UpdateOptions options, CqlIdentifier tableName,
+			CassandraPersistentEntity<?> persistentEntity) {
+
+		AdaptibleEntity<T> source = getEntityOperations().forEntity(entity, getConverter().getConversionService());
 
 		Number previousVersion = source.getVersion();
+		T toSave = source.incrementVersion();
 
-		T entity = source.incrementVersion();
+		Update update = getStatementFactory().update(toSave, options, getConverter(), persistentEntity, tableName);
 
-		Update update = getStatementFactory().update(entity, options, getConverter(), persistentEntity, tableName);
-
-		return executeSave(entity, tableName, source.appendVersionCondition(update, previousVersion), result -> {
+		return executeSave(toSave, tableName, source.appendVersionCondition(update, previousVersion), result -> {
 
 			if (!result.wasApplied()) {
 				throw new OptimisticLockingFailureException(
-						String.format("Cannot save entity %s with version %s to table %s. Has it been modified meanwhile?", entity,
+						String.format("Cannot save entity %s with version %s to table %s. Has it been modified meanwhile?", toSave,
 								source.getVersion(), tableName));
 			}
 		});
@@ -818,13 +853,14 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 			Consumer<WriteResult> resultConsumer) {
 
 		maybeEmitEvent(new BeforeSaveEvent<>(entity, tableName, statement));
+		T entityToSave = maybeCallBeforeSave(entity, tableName, statement);
 
 		WriteResult result = getCqlOperations().execute(new StatementCallback(statement));
 		resultConsumer.accept(result);
 
-		maybeEmitEvent(new AfterSaveEvent<>(entity, tableName));
+		maybeEmitEvent(new AfterSaveEvent<>(entityToSave, tableName));
 
-		return EntityWriteResult.of(result, entity);
+		return EntityWriteResult.of(result, entityToSave);
 	}
 
 	private WriteResult executeDelete(Object entity, CqlIdentifier tableName, Statement statement,
@@ -903,6 +939,24 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 		}
 	}
 
+	protected <T> T maybeCallBeforeConvert(T object, CqlIdentifier tableName) {
+
+		if (null != entityCallbacks) {
+			return (T) entityCallbacks.callback(BeforeConvertCallback.class, object, tableName);
+		}
+
+		return object;
+	}
+
+	protected <T> T maybeCallBeforeSave(T object, CqlIdentifier tableName, Statement statement) {
+
+		if (null != entityCallbacks) {
+			return (T) entityCallbacks.callback(BeforeSaveCallback.class, object, tableName, statement);
+		}
+
+		return object;
+	}
+
 	@Value
 	static class StatementCallback implements SessionCallback<WriteResult>, CqlProvider {
 

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

@@ -26,6 +26,9 @@ import java.util.function.Function;
 
 import org.reactivestreams.Publisher;
 
+import org.springframework.beans.BeansException;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.ApplicationContextAware;
 import org.springframework.context.ApplicationEvent;
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.context.ApplicationEventPublisherAware;
@@ -56,10 +59,14 @@ import org.springframework.data.cassandra.core.mapping.event.AfterLoadEvent;
 import org.springframework.data.cassandra.core.mapping.event.AfterSaveEvent;
 import org.springframework.data.cassandra.core.mapping.event.BeforeDeleteEvent;
 import org.springframework.data.cassandra.core.mapping.event.BeforeSaveEvent;
+import org.springframework.data.cassandra.core.mapping.event.ReactiveBeforeConvertCallback;
+import org.springframework.data.cassandra.core.mapping.event.ReactiveBeforeSaveCallback;
 import org.springframework.data.cassandra.core.query.Columns;
 import org.springframework.data.cassandra.core.query.Query;
 import org.springframework.data.domain.Slice;
 import org.springframework.data.domain.SliceImpl;
+import org.springframework.data.mapping.callback.EntityCallbacks;
+import org.springframework.data.mapping.callback.ReactiveEntityCallbacks;
 import org.springframework.data.projection.ProjectionFactory;
 import org.springframework.data.projection.SpelAwareProxyProjectionFactory;
 import org.springframework.lang.Nullable;
@@ -96,10 +103,13 @@ import com.datastax.driver.core.querybuilder.Update;
  * @author Hleb Albau
  * @since 2.0
  */
-public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, ApplicationEventPublisherAware {
+public class ReactiveCassandraTemplate
+		implements ReactiveCassandraOperations, ApplicationEventPublisherAware, ApplicationContextAware {
 
 	private @Nullable ApplicationEventPublisher eventPublisher;
 
+	private @Nullable ReactiveEntityCallbacks entityCallbacks;
+
 	private final CassandraConverter converter;
 
 	private final EntityOperations entityOperations;
@@ -189,6 +199,29 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A
 		this.eventPublisher = applicationEventPublisher;
 	}
 
+	/* (non-Javadoc)
+	* @see org.springframework.context.ApplicationContextAware(org.springframework.context.ApplicationContext)
+	*/
+	@Override
+	public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
+
+		if (entityCallbacks == null) {
+			setEntityCallbacks(ReactiveEntityCallbacks.create(applicationContext));
+		}
+
+		projectionFactory.setBeanFactory(applicationContext);
+		projectionFactory.setBeanClassLoader(applicationContext.getClassLoader());
+	}
+
+	/**
+	 * Configure {@link EntityCallbacks} to pre-/post-process entities during persistence operations.
+	 *
+	 * @param entityCallbacks
+	 */
+	public void setEntityCallbacks(@Nullable ReactiveEntityCallbacks entityCallbacks) {
+		this.entityCallbacks = entityCallbacks;
+	}
+
 	/* (non-Javadoc)
 	 * @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#getConverter()
 	 */
@@ -538,16 +571,20 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A
 
 	<T> Mono<EntityWriteResult<T>> doInsert(T entity, WriteOptions options, CqlIdentifier tableName) {
 
-		AdaptibleEntity<T> source = this.entityOperations.forEntity(entity, getConverter().getConversionService());
-		CassandraPersistentEntity<?> persistentEntity = getRequiredPersistentEntity(entity.getClass());
+		return maybeCallBeforeConvert(entity, tableName).flatMap(entityToInsert -> {
 
-		T entityToUse = source.isVersionedEntity() ? source.initializeVersionProperty() : entity;
+			AdaptibleEntity<T> source = this.entityOperations.forEntity(entityToInsert,
+					getConverter().getConversionService());
+			CassandraPersistentEntity<?> persistentEntity = getRequiredPersistentEntity(entityToInsert.getClass());
 
-		Insert insert = EntityQueryUtils.createInsertQuery(tableName.toCql(), entityToUse, options, getConverter(),
-				persistentEntity);
+			T entityToUse = source.isVersionedEntity() ? source.initializeVersionProperty() : entityToInsert;
 
-		return source.isVersionedEntity() ? doInsertVersioned(insert.ifNotExists(), entityToUse, source, tableName)
-				: doInsert(insert, entityToUse, tableName);
+			Insert insert = EntityQueryUtils.createInsertQuery(tableName.toCql(), entityToUse, options, getConverter(),
+					persistentEntity);
+
+			return source.isVersionedEntity() ? doInsertVersioned(insert.ifNotExists(), entityToUse, source, tableName)
+					: doInsert(insert, entityToUse, tableName);
+		});
 	}
 
 	private <T> Mono<EntityWriteResult<T>> doInsertVersioned(Insert insert, T entity, AdaptibleEntity<T> source,
@@ -593,25 +630,28 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A
 		CassandraPersistentEntity<?> persistentEntity = getRequiredPersistentEntity(entity.getClass());
 		CqlIdentifier tableName = persistentEntity.getTableName();
 
-		return source.isVersionedEntity() ? doUpdateVersioned(source, options, tableName, persistentEntity)
-				: doUpdate(entity, options, tableName, persistentEntity);
+		return maybeCallBeforeConvert(entity, tableName).flatMap(entityToUpdate -> {
+			return source.isVersionedEntity() ? doUpdateVersioned(entity, options, tableName, persistentEntity)
+					: doUpdate(entity, options, tableName, persistentEntity);
+		});
 	}
 
-	private <T> Mono<EntityWriteResult<T>> doUpdateVersioned(AdaptibleEntity<T> source, UpdateOptions options,
-			CqlIdentifier tableName, CassandraPersistentEntity<?> persistentEntity) {
+	private <T> Mono<EntityWriteResult<T>> doUpdateVersioned(T entity, UpdateOptions options, CqlIdentifier tableName,
+			CassandraPersistentEntity<?> persistentEntity) {
+
+		AdaptibleEntity<T> source = getEntityOperations().forEntity(entity, getConverter().getConversionService());
 
 		Number previousVersion = source.getVersion();
+		T toSave = source.incrementVersion();
 
-		T entity = source.incrementVersion();
+		Update update = getStatementFactory().update(toSave, options, getConverter(), persistentEntity, tableName);
 
-		Update update = getStatementFactory().update(entity, options, getConverter(), persistentEntity, tableName);
-
-		return executeSave(entity, tableName, source.appendVersionCondition(update, previousVersion), (result, sink) -> {
+		return executeSave(toSave, tableName, source.appendVersionCondition(update, previousVersion), (result, sink) -> {
 
 			if (!result.wasApplied()) {
 
 				sink.error(new OptimisticLockingFailureException(
-						String.format("Cannot save entity %s with version %s to table %s. Has it been modified meanwhile?", entity,
+						String.format("Cannot save entity %s with version %s to table %s. Has it been modified meanwhile?", toSave,
 								source.getVersion(), tableName)));
 
 				return;
@@ -763,14 +803,18 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A
 	private <T> Mono<EntityWriteResult<T>> executeSave(T entity, CqlIdentifier tableName, Statement statement,
 			BiConsumer<EntityWriteResult<T>, SynchronousSink<EntityWriteResult<T>>> handler) {
 
-		maybeEmitEvent(new BeforeSaveEvent<>(entity, tableName, statement));
+		return Mono.defer(() -> {
 
-		Flux<WriteResult> execute = getReactiveCqlOperations().execute(new StatementCallback(statement));
+			maybeEmitEvent(new BeforeSaveEvent<>(entity, tableName, statement));
+
+			return maybeCallBeforeSave(entity, tableName, statement).flatMapMany(entityToSave -> {
+				Flux<WriteResult> execute = getReactiveCqlOperations().execute(new StatementCallback(statement));
+
+				return execute.map(it -> EntityWriteResult.of(it, entityToSave)).handle(handler) //
+						.doOnNext(it -> maybeEmitEvent(new AfterSaveEvent<>(entityToSave, tableName)));
+			}).next();
+		});
 
-		return execute.map(it -> EntityWriteResult.of(it, entity)).handle(handler) //
-				.doOnSubscribe(it -> maybeEmitEvent(new BeforeSaveEvent<>(entity, tableName, statement))) //
-				.doOnNext(it -> maybeEmitEvent(new AfterSaveEvent<>(it, tableName))) //
-				.next();
 	}
 
 	private Mono<WriteResult> executeDelete(Object entity, CqlIdentifier tableName, Statement statement,
@@ -786,7 +830,6 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A
 				.next();
 	}
 
-	@SuppressWarnings("ConstantConditions")
 	private Mono<Integer> getEffectiveFetchSize(Statement statement) {
 
 		if (statement.getFetchSize() > 0) {
@@ -843,6 +886,24 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A
 		}
 	}
 
+	protected <T> Mono<T> maybeCallBeforeConvert(T object, CqlIdentifier tableName) {
+
+		if (null != entityCallbacks) {
+			return entityCallbacks.callback(ReactiveBeforeConvertCallback.class, object, tableName);
+		}
+
+		return Mono.just(object);
+	}
+
+	protected <T> Mono<T> maybeCallBeforeSave(T object, CqlIdentifier tableName, Statement statement) {
+
+		if (null != entityCallbacks) {
+			return entityCallbacks.callback(ReactiveBeforeSaveCallback.class, object, tableName, statement);
+		}
+
+		return Mono.just(object);
+	}
+
 	@Value
 	static class StatementCallback implements ReactiveSessionCallback<WriteResult>, CqlProvider {
 

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/BeforeConvertCallback.java

@@ -0,0 +1,40 @@
+/*
+ * Copyright 2019 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.mapping.event;
+
+import org.springframework.data.cassandra.core.cql.CqlIdentifier;
+import org.springframework.data.mapping.callback.EntityCallback;
+
+/**
+ * Callback being invoked before a domain object is converted to be persisted.
+ *
+ * @author Mark Paluch
+ * @since 2.2
+ * @see org.springframework.data.mapping.callback.EntityCallbacks
+ */
+@FunctionalInterface
+public interface BeforeConvertCallback<T> extends EntityCallback<T> {
+
+	/**
+	 * Entity callback method invoked before a domain object is converted to be persisted. Can return either the same or a
+	 * modified instance of the domain object.
+	 *
+	 * @param entity the domain object to save.
+	 * @param tableName name of the table.
+	 * @return the domain object to be persisted.
+	 */
+	T onBeforeConvert(T entity, CqlIdentifier tableName);
+}

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/BeforeSaveCallback.java

@@ -0,0 +1,44 @@
+/*
+ * Copyright 2019 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.mapping.event;
+
+import org.springframework.data.cassandra.core.cql.CqlIdentifier;
+import org.springframework.data.mapping.callback.EntityCallback;
+
+import com.datastax.driver.core.Statement;
+
+/**
+ * Entity callback triggered before save of a row.
+ *
+ * @author Mark Paluch
+ * @since 2.2
+ * @see org.springframework.data.mapping.callback.EntityCallbacks
+ */
+@FunctionalInterface
+public interface BeforeSaveCallback<T> extends EntityCallback<T> {
+
+	/**
+	 * Entity callback method invoked before a domain object is saved. Can return either the same of a modified instance
+	 * of the domain object and can modify {@link Statement} contents. This method is called after converting the
+	 * {@code entity} to {@link Statement} so effectively the row is used as outcome of invoking this callback.
+	 *
+	 * @param entity the domain object to save.
+	 * @param tableName name of the table.
+	 * @param statement {@link Statement} representing the {@code entity} operation.
+	 * @return the domain object to be persisted.
+	 */
+	T onBeforeSave(T entity, CqlIdentifier tableName, Statement statement);
+}

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/ReactiveBeforeConvertCallback.java

@@ -0,0 +1,42 @@
+/*
+ * Copyright 2019 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.mapping.event;
+
+import org.reactivestreams.Publisher;
+
+import org.springframework.data.cassandra.core.cql.CqlIdentifier;
+import org.springframework.data.mapping.callback.EntityCallback;
+
+/**
+ * Callback being invoked before a domain object is converted to be persisted.
+ *
+ * @author Mark Paluch
+ * @since 2.2
+ * @see org.springframework.data.mapping.callback.ReactiveEntityCallbacks
+ */
+@FunctionalInterface
+public interface ReactiveBeforeConvertCallback<T> extends EntityCallback<T> {
+
+	/**
+	 * Entity callback method invoked before a domain object is converted to be persisted. Can return either the same or a
+	 * modified instance of the domain object.
+	 *
+	 * @param entity the domain object to save.
+	 * @param tableName name of the table.
+	 * @return a {@link Publisher} emitting the domain object to be persisted.
+	 */
+	Publisher<T> onBeforeConvert(T entity, CqlIdentifier tableName);
+}

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/ReactiveBeforeSaveCallback.java

@@ -0,0 +1,46 @@
+/*
+ * Copyright 2019 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.mapping.event;
+
+import org.reactivestreams.Publisher;
+
+import org.springframework.data.cassandra.core.cql.CqlIdentifier;
+import org.springframework.data.mapping.callback.EntityCallback;
+
+import com.datastax.driver.core.Statement;
+
+/**
+ * Entity callback triggered before save of a row.
+ *
+ * @author Mark Paluch
+ * @since 2.2
+ * @see org.springframework.data.mapping.callback.ReactiveEntityCallbacks
+ */
+@FunctionalInterface
+public interface ReactiveBeforeSaveCallback<T> extends EntityCallback<T> {
+
+	/**
+	 * Entity callback method invoked before a domain object is saved. Can return either the same of a modified instance
+	 * of the domain object and can modify {@link Statement} contents. This method is called after converting the
+	 * {@code entity} to {@link Statement} so effectively the row is used as outcome of invoking this callback.
+	 *
+	 * @param entity the domain object to save.
+	 * @param tableName name of the table.
+	 * @param statement {@link Statement} representing the {@code entity} operation.
+	 * @return a {@link Publisher} emitting the domain object to be persisted.
+	 */
+	Publisher<T> onBeforeSave(T entity, CqlIdentifier tableName, Statement statement);
+}

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

@@ -15,15 +15,10 @@
  */
 package org.springframework.data.cassandra.core;
 
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.fail;
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.ArgumentMatchers.anyInt;
-import static org.mockito.ArgumentMatchers.anyString;
-import static org.mockito.Mockito.reset;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
-import static org.springframework.data.cassandra.core.query.Criteria.where;
+import static org.assertj.core.api.Assertions.*;
+import static org.mockito.ArgumentMatchers.*;
+import static org.mockito.Mockito.*;
+import static org.springframework.data.cassandra.core.query.Criteria.*;
 
 import java.util.ArrayList;
 import java.util.Collections;
@@ -42,10 +37,14 @@ import org.mockito.Mock;
 import org.mockito.junit.MockitoJUnitRunner;
 
 import org.springframework.data.cassandra.CassandraConnectionFailureException;
+import org.springframework.data.cassandra.core.mapping.event.BeforeConvertCallback;
+import org.springframework.data.cassandra.core.mapping.event.BeforeSaveCallback;
 import org.springframework.data.cassandra.core.query.Filter;
 import org.springframework.data.cassandra.core.query.Query;
 import org.springframework.data.cassandra.core.query.Update;
 import org.springframework.data.cassandra.domain.User;
+import org.springframework.data.cassandra.domain.VersionedUser;
+import org.springframework.data.mapping.callback.EntityCallbacks;
 import org.springframework.util.concurrent.ListenableFuture;
 
 import com.datastax.driver.core.ColumnDefinitions;
@@ -75,6 +74,10 @@ public class AsyncCassandraTemplateUnitTests {
 
 	AsyncCassandraTemplate template;
 
+	Object beforeSave;
+
+	Object beforeConvert;
+
 	@Before
 	public void setUp() {
 
@@ -82,6 +85,24 @@ public class AsyncCassandraTemplateUnitTests {
 
 		when(session.executeAsync(any(Statement.class))).thenReturn(new TestResultSetFuture(resultSet));
 		when(row.getColumnDefinitions()).thenReturn(columnDefinitions);
+
+		EntityCallbacks callbacks = EntityCallbacks.create();
+		callbacks.addEntityCallback((BeforeSaveCallback<Object>) (entity, tableName, statement) -> {
+
+			assertThat(tableName).isNotNull();
+			assertThat(statement).isNotNull();
+			beforeSave = entity;
+			return entity;
+		});
+
+		callbacks.addEntityCallback((BeforeConvertCallback<Object>) (entity, tableName) -> {
+
+			assertThat(tableName).isNotNull();
+			beforeConvert = entity;
+			return entity;
+		});
+
+		template.setEntityCallbacks(callbacks);
 	}
 
 	@Test // DATACASS-292
@@ -256,7 +277,7 @@ public class AsyncCassandraTemplateUnitTests {
 		assertThat(statementCaptor.getValue().toString()).isEqualTo("SELECT COUNT(1) FROM users;");
 	}
 
-	@Test // DATACASS-292
+	@Test // DATACASS-292, DATACASS-618
 	public void insertShouldInsertEntity() {
 
 		when(resultSet.wasApplied()).thenReturn(true);
@@ -269,6 +290,25 @@ public class AsyncCassandraTemplateUnitTests {
 		verify(session).executeAsync(statementCaptor.capture());
 		assertThat(statementCaptor.getValue().toString())
 				.isEqualTo("INSERT INTO users (firstname,id,lastname) VALUES ('Walter','heisenberg','White');");
+		assertThat(beforeConvert).isSameAs(user);
+		assertThat(beforeSave).isSameAs(user);
+	}
+
+	@Test // DATACASS-618
+	public void insertShouldInsertVersionedEntity() {
+
+		when(resultSet.wasApplied()).thenReturn(true);
+
+		VersionedUser user = new VersionedUser("heisenberg", "Walter", "White");
+
+		ListenableFuture<VersionedUser> future = template.insert(user);
+
+		assertThat(getUninterruptibly(future)).isEqualTo(user);
+		verify(session).executeAsync(statementCaptor.capture());
+		assertThat(statementCaptor.getValue().toString()).isEqualTo(
+				"INSERT INTO vusers (firstname,id,lastname,version) VALUES ('Walter','heisenberg','White',0) IF NOT EXISTS;");
+		assertThat(beforeConvert).isSameAs(user);
+		assertThat(beforeSave).isSameAs(user);
 	}
 
 	@Test // DATACASS-292
@@ -290,7 +330,7 @@ public class AsyncCassandraTemplateUnitTests {
 		}
 	}
 
-	@Test // DATACASS-292
+	@Test // DATACASS-292, DATACASS-618
 	public void updateShouldUpdateEntity() {
 
 		when(resultSet.wasApplied()).thenReturn(true);
@@ -303,6 +343,26 @@ public class AsyncCassandraTemplateUnitTests {
 		verify(session).executeAsync(statementCaptor.capture());
 		assertThat(statementCaptor.getValue().toString())
 				.isEqualTo("UPDATE users SET firstname='Walter',lastname='White' WHERE id='heisenberg';");
+		assertThat(beforeConvert).isSameAs(user);
+		assertThat(beforeSave).isSameAs(user);
+	}
+
+	@Test // DATACASS-618
+	public void updateShouldUpdateVersionedEntity() {
+
+		when(resultSet.wasApplied()).thenReturn(true);
+
+		VersionedUser user = new VersionedUser("heisenberg", "Walter", "White");
+		user.setVersion(0L);
+
+		ListenableFuture<VersionedUser> future = template.update(user);
+
+		assertThat(getUninterruptibly(future)).isEqualTo(user);
+		verify(session).executeAsync(statementCaptor.capture());
+		assertThat(statementCaptor.getValue().toString()).isEqualTo(
+				"UPDATE vusers SET firstname='Walter',lastname='White',version=1 WHERE id='heisenberg' IF version=0;");
+		assertThat(beforeConvert).isSameAs(user);
+		assertThat(beforeSave).isSameAs(user);
 	}
 
 	@Test // DATACASS-575
@@ -357,8 +417,8 @@ public class AsyncCassandraTemplateUnitTests {
 		template.update(query, update, User.class);
 
 		verify(session).executeAsync(statementCaptor.capture());
-		assertThat(statementCaptor.getValue().toString())
-				.isEqualTo("UPDATE users SET firstname='Walter' WHERE id='heisenberg' IF firstname='Walter' AND lastname='White';");
+		assertThat(statementCaptor.getValue().toString()).isEqualTo(
+				"UPDATE users SET firstname='Walter' WHERE id='heisenberg' IF firstname='Walter' AND lastname='White';");
 	}
 
 	@Test // DATACASS-292

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

@@ -15,15 +15,10 @@
  */
 package org.springframework.data.cassandra.core;
 
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.fail;
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.ArgumentMatchers.anyInt;
-import static org.mockito.ArgumentMatchers.anyString;
-import static org.mockito.Mockito.reset;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
-import static org.springframework.data.cassandra.core.query.Criteria.where;
+import static org.assertj.core.api.Assertions.*;
+import static org.mockito.ArgumentMatchers.*;
+import static org.mockito.Mockito.*;
+import static org.springframework.data.cassandra.core.query.Criteria.*;
 
 import java.util.Collections;
 import java.util.List;
@@ -39,10 +34,14 @@ import org.mockito.Mockito;
 import org.mockito.junit.MockitoJUnitRunner;
 
 import org.springframework.data.cassandra.CassandraConnectionFailureException;
+import org.springframework.data.cassandra.core.mapping.event.BeforeConvertCallback;
+import org.springframework.data.cassandra.core.mapping.event.BeforeSaveCallback;
 import org.springframework.data.cassandra.core.query.Filter;
 import org.springframework.data.cassandra.core.query.Query;
 import org.springframework.data.cassandra.core.query.Update;
 import org.springframework.data.cassandra.domain.User;
+import org.springframework.data.cassandra.domain.VersionedUser;
+import org.springframework.data.mapping.callback.EntityCallbacks;
 
 import com.datastax.driver.core.ColumnDefinitions;
 import com.datastax.driver.core.DataType;
@@ -70,6 +69,10 @@ public class CassandraTemplateUnitTests {
 
 	CassandraTemplate template;
 
+	Object beforeSave;
+
+	Object beforeConvert;
+
 	@Before
 	public void setUp() {
 
@@ -77,6 +80,24 @@ public class CassandraTemplateUnitTests {
 
 		when(session.execute(any(Statement.class))).thenReturn(resultSet);
 		when(row.getColumnDefinitions()).thenReturn(columnDefinitions);
+
+		EntityCallbacks callbacks = EntityCallbacks.create();
+		callbacks.addEntityCallback((BeforeSaveCallback<Object>) (entity, tableName, statement) -> {
+
+			assertThat(tableName).isNotNull();
+			assertThat(statement).isNotNull();
+			beforeSave = entity;
+			return entity;
+		});
+
+		callbacks.addEntityCallback((BeforeConvertCallback<Object>) (entity, tableName) -> {
+
+			assertThat(tableName).isNotNull();
+			beforeConvert = entity;
+			return entity;
+		});
+
+		template.setEntityCallbacks(callbacks);
 	}
 
 	@Test // DATACASS-292
@@ -241,7 +262,7 @@ public class CassandraTemplateUnitTests {
 		assertThat(statementCaptor.getValue().toString()).isEqualTo("SELECT COUNT(1) FROM users;");
 	}
 
-	@Test // DATACASS-292
+	@Test // DATACASS-292, DATACASS-618
 	public void insertShouldInsertEntity() {
 
 		when(resultSet.wasApplied()).thenReturn(true);
@@ -253,6 +274,24 @@ public class CassandraTemplateUnitTests {
 		verify(session).execute(statementCaptor.capture());
 		assertThat(statementCaptor.getValue().toString())
 				.isEqualTo("INSERT INTO users (firstname,id,lastname) VALUES ('Walter','heisenberg','White');");
+		assertThat(beforeConvert).isSameAs(user);
+		assertThat(beforeSave).isSameAs(user);
+	}
+
+	@Test // DATACASS-618
+	public void insertShouldInsertVersionedEntity() {
+
+		when(resultSet.wasApplied()).thenReturn(true);
+
+		VersionedUser user = new VersionedUser("heisenberg", "Walter", "White");
+
+		template.insert(user);
+
+		verify(session).execute(statementCaptor.capture());
+		assertThat(statementCaptor.getValue().toString()).isEqualTo(
+				"INSERT INTO vusers (firstname,id,lastname,version) VALUES ('Walter','heisenberg','White',0) IF NOT EXISTS;");
+		assertThat(beforeConvert).isSameAs(user);
+		assertThat(beforeSave).isSameAs(user);
 	}
 
 	@Test // DATACASS-250
@@ -314,7 +353,7 @@ public class CassandraTemplateUnitTests {
 		assertThat(writeResult.wasApplied()).isFalse();
 	}
 
-	@Test // DATACASS-292
+	@Test // DATACASS-292, DATACASS-618
 	public void updateShouldUpdateEntity() {
 
 		when(resultSet.wasApplied()).thenReturn(true);
@@ -326,6 +365,25 @@ public class CassandraTemplateUnitTests {
 		verify(session).execute(statementCaptor.capture());
 		assertThat(statementCaptor.getValue().toString())
 				.isEqualTo("UPDATE users SET firstname='Walter',lastname='White' WHERE id='heisenberg';");
+		assertThat(beforeConvert).isSameAs(user);
+		assertThat(beforeSave).isSameAs(user);
+	}
+
+	@Test // DATACASS-618
+	public void updateShouldUpdateVersionedEntity() {
+
+		when(resultSet.wasApplied()).thenReturn(true);
+
+		VersionedUser user = new VersionedUser("heisenberg", "Walter", "White");
+		user.setVersion(0L);
+
+		template.update(user);
+
+		verify(session).execute(statementCaptor.capture());
+		assertThat(statementCaptor.getValue().toString()).isEqualTo(
+				"UPDATE vusers SET firstname='Walter',lastname='White',version=1 WHERE id='heisenberg' IF version=0;");
+		assertThat(beforeConvert).isSameAs(user);
+		assertThat(beforeSave).isSameAs(user);
 	}
 
 	@Test // DATACASS-250
@@ -384,8 +442,8 @@ public class CassandraTemplateUnitTests {
 		template.update(query, update, User.class);
 
 		verify(session).execute(statementCaptor.capture());
-		assertThat(statementCaptor.getValue().toString())
-			.isEqualTo("UPDATE users SET firstname='Walter' WHERE id='heisenberg' IF firstname='Walter' AND lastname='White';");
+		assertThat(statementCaptor.getValue().toString()).isEqualTo(
+				"UPDATE users SET firstname='Walter' WHERE id='heisenberg' IF firstname='Walter' AND lastname='White';");
 	}
 
 	@Test // DATACASS-292

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

@@ -15,21 +15,17 @@
  */
 package org.springframework.data.cassandra.core;
 
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.ArgumentMatchers.anyInt;
-import static org.mockito.ArgumentMatchers.anyString;
-import static org.mockito.Mockito.reset;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
-import static org.springframework.data.cassandra.core.query.Criteria.where;
-
-import java.util.Collections;
+import static org.assertj.core.api.Assertions.*;
+import static org.mockito.ArgumentMatchers.*;
+import static org.mockito.Mockito.*;
+import static org.springframework.data.cassandra.core.query.Criteria.*;
 
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
 
+import java.util.Collections;
+
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -40,10 +36,14 @@ import org.mockito.junit.MockitoJUnitRunner;
 
 import org.springframework.data.cassandra.ReactiveResultSet;
 import org.springframework.data.cassandra.ReactiveSession;
+import org.springframework.data.cassandra.core.mapping.event.ReactiveBeforeConvertCallback;
+import org.springframework.data.cassandra.core.mapping.event.ReactiveBeforeSaveCallback;
 import org.springframework.data.cassandra.core.query.Filter;
 import org.springframework.data.cassandra.core.query.Query;
 import org.springframework.data.cassandra.core.query.Update;
 import org.springframework.data.cassandra.domain.User;
+import org.springframework.data.cassandra.domain.VersionedUser;
+import org.springframework.data.mapping.callback.ReactiveEntityCallbacks;
 
 import com.datastax.driver.core.ColumnDefinitions;
 import com.datastax.driver.core.DataType;
@@ -68,6 +68,10 @@ public class ReactiveCassandraTemplateUnitTests {
 
 	ReactiveCassandraTemplate template;
 
+	Object beforeSave;
+
+	Object beforeConvert;
+
 	@Before
 	public void setUp() {
 
@@ -75,6 +79,24 @@ public class ReactiveCassandraTemplateUnitTests {
 
 		when(session.execute(any(Statement.class))).thenReturn(Mono.just(reactiveResultSet));
 		when(row.getColumnDefinitions()).thenReturn(columnDefinitions);
+
+		ReactiveEntityCallbacks callbacks = ReactiveEntityCallbacks.create();
+		callbacks.addEntityCallback((ReactiveBeforeSaveCallback<Object>) (entity, tableName, statement) -> {
+
+			assertThat(tableName).isNotNull();
+			assertThat(statement).isNotNull();
+			beforeSave = entity;
+			return Mono.just(entity);
+		});
+
+		callbacks.addEntityCallback((ReactiveBeforeConvertCallback<Object>) (entity, tableName) -> {
+
+			assertThat(tableName).isNotNull();
+			beforeConvert = entity;
+			return Mono.just(entity);
+		});
+
+		template.setEntityCallbacks(callbacks);
 	}
 
 	@Test // DATACASS-335
@@ -226,7 +248,7 @@ public class ReactiveCassandraTemplateUnitTests {
 		assertThat(statementCaptor.getValue().toString()).isEqualTo("SELECT COUNT(1) FROM users;");
 	}
 
-	@Test // DATACASS-335
+	@Test // DATACASS-335, DATACASS-618
 	public void insertShouldInsertEntity() {
 
 		when(reactiveResultSet.wasApplied()).thenReturn(true);
@@ -238,6 +260,24 @@ public class ReactiveCassandraTemplateUnitTests {
 		verify(session).execute(statementCaptor.capture());
 		assertThat(statementCaptor.getValue().toString())
 				.isEqualTo("INSERT INTO users (firstname,id,lastname) VALUES ('Walter','heisenberg','White');");
+		assertThat(beforeConvert).isSameAs(user);
+		assertThat(beforeSave).isSameAs(user);
+	}
+
+	@Test // DATACASS-618
+	public void insertShouldInsertVersionedEntity() {
+
+		when(reactiveResultSet.wasApplied()).thenReturn(true);
+		when(reactiveResultSet.rows()).thenReturn(Flux.just(row));
+
+		VersionedUser user = new VersionedUser("heisenberg", "Walter", "White");
+		StepVerifier.create(template.insert(user)).expectNext(user).verifyComplete();
+
+		verify(session).execute(statementCaptor.capture());
+		assertThat(statementCaptor.getValue().toString()).isEqualTo(
+				"INSERT INTO vusers (firstname,id,lastname,version) VALUES ('Walter','heisenberg','White',0) IF NOT EXISTS;");
+		assertThat(beforeConvert).isSameAs(user);
+		assertThat(beforeSave).isSameAs(user);
 	}
 
 	@Test // DATACASS-335
@@ -254,7 +294,7 @@ public class ReactiveCassandraTemplateUnitTests {
 				}).verify();
 	}
 
-	@Test // DATACASS-335
+	@Test // DATACASS-335, DATACASS-618
 	public void updateShouldUpdateEntity() {
 
 		when(reactiveResultSet.wasApplied()).thenReturn(true);
@@ -267,6 +307,26 @@ public class ReactiveCassandraTemplateUnitTests {
 		verify(session).execute(statementCaptor.capture());
 		assertThat(statementCaptor.getValue().toString())
 				.isEqualTo("UPDATE users SET firstname='Walter',lastname='White' WHERE id='heisenberg';");
+		assertThat(beforeConvert).isSameAs(user);
+		assertThat(beforeSave).isSameAs(user);
+	}
+
+	@Test // DATACASS-618
+	public void updateShouldUpdateVersionedEntity() {
+
+		when(reactiveResultSet.wasApplied()).thenReturn(true);
+		when(reactiveResultSet.rows()).thenReturn(Flux.just(row));
+
+		VersionedUser user = new VersionedUser("heisenberg", "Walter", "White");
+		user.setVersion(0L);
+
+		StepVerifier.create(template.update(user)).expectNext(user).verifyComplete();
+
+		verify(session).execute(statementCaptor.capture());
+		assertThat(statementCaptor.getValue().toString()).isEqualTo(
+				"UPDATE vusers SET firstname='Walter',lastname='White',version=1 WHERE id='heisenberg' IF version=0;");
+		assertThat(beforeConvert).isSameAs(user);
+		assertThat(beforeSave).isSameAs(user);
 	}
 
 	@Test // DATACASS-575
@@ -341,8 +401,8 @@ public class ReactiveCassandraTemplateUnitTests {
 				.verifyComplete();
 
 		verify(session).execute(statementCaptor.capture());
-		assertThat(statementCaptor.getValue().toString())
-				.isEqualTo("UPDATE users SET firstname='Walter' WHERE id='heisenberg' IF firstname='Walter' AND lastname='White';");
+		assertThat(statementCaptor.getValue().toString()).isEqualTo(
+				"UPDATE users SET firstname='Walter' WHERE id='heisenberg' IF firstname='Walter' AND lastname='White';");
 	}
 
 	@Test // DATACASS-335

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

@@ -0,0 +1,56 @@
+/*
+ * Copyright 2019 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.domain;
+
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.NoArgsConstructor;
+
+import org.springframework.data.annotation.Id;
+import org.springframework.data.annotation.PersistenceConstructor;
+import org.springframework.data.annotation.Version;
+import org.springframework.data.cassandra.core.mapping.Table;
+
+/**
+ * @author Mark Paluch
+ */
+@Table("vusers")
+@Data
+@NoArgsConstructor
+@EqualsAndHashCode(of = "id")
+public class VersionedUser {
+
+	/*
+	 * Primary Row ID
+	 */
+	@Id private String id;
+
+	@Version private Long version;
+
+	/*
+	 * Public information
+	 */
+	private String firstname;
+	private String lastname;
+
+	@PersistenceConstructor
+	public VersionedUser(String id, String firstname, String lastname) {
+
+		this.id = id;
+		this.firstname = firstname;
+		this.lastname = lastname;
+	}
+}