DATACASS-576 - Add support for Optimistic Locking.

We now support Optimistic Locking for insert, update and delete operations leveraging Cassandra's lightweight transaction support. Modifying statements are enhanced with IF conditions to conditionally insert and modify rows and to prevent concurrent modifications by throwing OptimisticLockingFailureException.

@Table
class Person {

  @Id String id;
  String firstname;
  String lastname;
  @Version Long version;
}

Person daenerys = template.insert(new Person("Daenerys"));

Person tmp = template.findOne(query(where("id").is(daenerys.getId())), Person.class);

daenerys.setLastname("Targaryen");
template.save(daenerys);

template.save(tmp); // throws OptimisticLockingFailureException

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

@@ -26,7 +26,9 @@ import org.springframework.context.ApplicationEvent;
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.context.ApplicationEventPublisherAware;
 import org.springframework.dao.DataAccessException;
+import org.springframework.dao.OptimisticLockingFailureException;
 import org.springframework.data.cassandra.SessionFactory;
+import org.springframework.data.cassandra.core.EntityOperations.AdaptibleEntity;
 import org.springframework.data.cassandra.core.convert.CassandraConverter;
 import org.springframework.data.cassandra.core.convert.MappingCassandraConverter;
 import org.springframework.data.cassandra.core.convert.QueryMapper;
@@ -104,6 +106,8 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations, Applica
 
 	private final SpelAwareProxyProjectionFactory projectionFactory;
 
+	private final EntityOperations operations;
+
 	private final StatementFactory statementFactory;
 
 	private @Nullable ApplicationEventPublisher eventPublisher;
@@ -168,6 +172,7 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations, Applica
 		this.cqlOperations = asyncCqlTemplate;
 		this.exceptionTranslator = asyncCqlTemplate.getExceptionTranslator();
 		this.projectionFactory = new SpelAwareProxyProjectionFactory();
+		this.operations = new EntityOperations(converter.getMappingContext());
 		this.statementFactory = new StatementFactory(new QueryMapper(converter), new UpdateMapper(converter));
 	}
 
@@ -488,7 +493,7 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations, Applica
 	 */
 	@Override
 	public <T> ListenableFuture<T> insert(T entity) {
-		return new MappingListenableFutureAdapter<>(insert(entity, InsertOptions.empty()), writeResult -> entity);
+		return new MappingListenableFutureAdapter<>(insert(entity, InsertOptions.empty()), EntityWriteResult::getEntity);
 	}
 
 	/* (non-Javadoc)
@@ -500,18 +505,37 @@ 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 = operations.forEntity(entity, converter.getConversionService());
 		CassandraPersistentEntity<?> persistentEntity = getRequiredPersistentEntity(entity.getClass());
 		CqlIdentifier tableName = persistentEntity.getTableName();
-		Insert insert = EntityQueryUtils.createInsertQuery(tableName.toCql(), entity, options, getConverter(),
+
+		T entityToUse = source.isVersionedEntity() ? source.initializeVersionProperty() : entity;
+
+		Insert insert = EntityQueryUtils.createInsertQuery(tableName.toCql(), entityToUse, options, getConverter(),
 				persistentEntity);
 
-		maybeEmitEvent(new BeforeSaveEvent<>(entity, tableName, insert));
+		if (source.isVersionedEntity()) {
+			return doInsertVersioned(insert.ifNotExists(), entityToUse, source, tableName);
+		}
 
-		return new MappingListenableFutureAdapter<>(getAsyncCqlOperations().execute(new AsyncStatementCallback(insert)),
-				resultSet -> {
-					maybeEmitEvent(new AfterSaveEvent<>(entity, tableName));
-					return EntityWriteResult.of(resultSet, entity);
-				});
+		return doInsert(insert, entityToUse, source, tableName);
+	}
+
+	private <T> ListenableFuture<EntityWriteResult<T>> doInsertVersioned(Insert insert, T entity,
+			AdaptibleEntity<T> source, CqlIdentifier tableName) {
+
+		return executeSave(entity, tableName, insert, result -> {
+			if (!result.wasApplied()) {
+				throw new OptimisticLockingFailureException(
+						String.format("Cannot insert entity %s with version, %s into table %s as it already exists", entity,
+								source.getVersion(), tableName));
+			}
+		});
+	}
+
+	private <T> ListenableFuture<EntityWriteResult<T>> doInsert(Insert insert, T entity, AdaptibleEntity<T> source,
+			CqlIdentifier tableName) {
+		return executeSave(entity, tableName, insert);
 	}
 
 	/* (non-Javadoc)
@@ -533,15 +557,38 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations, Applica
 
 		CassandraPersistentEntity<?> persistentEntity = getRequiredPersistentEntity(entity.getClass());
 		CqlIdentifier tableName = persistentEntity.getTableName();
+		AdaptibleEntity<T> source = operations.forEntity(entity, converter.getConversionService());
+
+		if (source.isVersionedEntity()) {
+			return doUpdateVersioned(source, options, tableName, persistentEntity);
+		}
+
+		return doUpdate(entity, options, tableName, persistentEntity);
+	}
+
+	private <T> ListenableFuture<EntityWriteResult<T>> doUpdate(T entity, UpdateOptions options, CqlIdentifier tableName,
+			CassandraPersistentEntity<?> persistentEntity) {
+
 		Update update = getStatementFactory().update(entity, options, getConverter(), persistentEntity, tableName);
 
-		maybeEmitEvent(new BeforeSaveEvent<>(entity, tableName, update));
+		return executeSave(entity, tableName, update);
+	}
 
-		return new MappingListenableFutureAdapter<>(getAsyncCqlOperations().execute(new AsyncStatementCallback(update)),
-				resultSet -> {
-					maybeEmitEvent(new AfterSaveEvent<>(entity, tableName));
-					return EntityWriteResult.of(resultSet, entity);
-				});
+	private <T> ListenableFuture<EntityWriteResult<T>> doUpdateVersioned(AdaptibleEntity<T> source, UpdateOptions options,
+			CqlIdentifier tableName, CassandraPersistentEntity<?> persistentEntity) {
+
+		Number previousVersion = source.getVersion();
+		T entity = source.incrementVersion();
+
+		Update update = getStatementFactory().update(entity, options, getConverter(), persistentEntity, tableName);
+
+		return executeSave(entity, 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,
+								source.getVersion(), tableName));
+			}
+		});
 	}
 
 	/* (non-Javadoc)
@@ -563,15 +610,31 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations, Applica
 
 		CassandraPersistentEntity<?> persistentEntity = getRequiredPersistentEntity(entity.getClass());
 		CqlIdentifier tableName = persistentEntity.getTableName();
+		AdaptibleEntity<Object> source = operations.forEntity(entity, converter.getConversionService());
+
 		Delete delete = getStatementFactory().delete(entity, options, getConverter(), persistentEntity, tableName);
 
-		maybeEmitEvent(new BeforeDeleteEvent<>(delete, entity.getClass(), tableName));
+		if (source.isVersionedEntity()) {
+			return doDeleteVersioned(delete, entity, source, tableName);
+		}
 
-		return new MappingListenableFutureAdapter<>(getAsyncCqlOperations().execute(new AsyncStatementCallback(delete)),
-				resultSet -> {
-					maybeEmitEvent(new AfterDeleteEvent<>(delete, entity.getClass(), tableName));
-					return WriteResult.of(resultSet);
-				});
+		return doDelete(delete, entity, tableName);
+	}
+
+	private ListenableFuture<WriteResult> doDeleteVersioned(Delete delete, Object entity, AdaptibleEntity<Object> source,
+			CqlIdentifier tableName) {
+
+		return executeDelete(entity, tableName, source.appendVersionCondition(delete), result -> {
+			if (!result.wasApplied()) {
+				throw new OptimisticLockingFailureException(
+						String.format("Cannot delete entity %s with version, %s in table %s. Has it been modified meanwhile?",
+								entity, source.getVersion(), tableName));
+			}
+		});
+	}
+
+	private ListenableFuture<WriteResult> doDelete(Delete delete, Object entity, CqlIdentifier tableName) {
+		return executeDelete(entity, tableName, delete, result -> {});
 	}
 
 	/* (non-Javadoc)
@@ -655,8 +718,52 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations, Applica
 		return this.statementFactory;
 	}
 
+	private <T> ListenableFuture<EntityWriteResult<T>> executeSave(T entity, CqlIdentifier tableName,
+			Statement statement) {
+		return executeSave(entity, tableName, statement, ignore -> {
+
+		});
+	}
+
+	private <T> ListenableFuture<EntityWriteResult<T>> executeSave(T entity, CqlIdentifier tableName, Statement statement,
+			Consumer<WriteResult> beforeAfterSaveEvent) {
+
+		maybeEmitEvent(new BeforeSaveEvent<>(entity, tableName, statement));
+
+		ListenableFuture<ResultSet> result = getAsyncCqlOperations().execute(new AsyncStatementCallback(statement));
+
+		return new MappingListenableFutureAdapter<>(result, resultSet -> {
+			EntityWriteResult<T> writeResult = EntityWriteResult.of(resultSet, entity);
+
+			beforeAfterSaveEvent.accept(writeResult);
+
+			maybeEmitEvent(new AfterSaveEvent<>(entity, tableName));
+
+			return writeResult;
+		});
+	}
+
+	private ListenableFuture<WriteResult> executeDelete(Object entity, CqlIdentifier tableName, Statement statement,
+			Consumer<WriteResult> resultConsumer) {
+
+		maybeEmitEvent(new BeforeDeleteEvent<>(statement, entity.getClass(), tableName));
+
+		ListenableFuture<ResultSet> result = getAsyncCqlOperations().execute(new AsyncStatementCallback(statement));
+
+		return new MappingListenableFutureAdapter<>(result, resultSet -> {
+
+			WriteResult writeResult = WriteResult.of(resultSet);
+
+			resultConsumer.accept(writeResult);
+
+			maybeEmitEvent(new AfterDeleteEvent<>(statement, entity.getClass(), tableName));
+
+			return writeResult;
+		});
+	}
+
 	private CqlIdentifier getTableName(Class<?> entityClass) {
-		return getRequiredPersistentEntity(entityClass).getTableName();
+		return operations.getTableName(entityClass);
 	}
 
 	private CqlIdentifier getTableName(Object entity) {

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

@@ -18,6 +18,7 @@ package org.springframework.data.cassandra.core;
 import lombok.Value;
 
 import java.util.List;
+import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.stream.Stream;
 import java.util.stream.StreamSupport;
@@ -26,7 +27,9 @@ import org.springframework.context.ApplicationEvent;
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.context.ApplicationEventPublisherAware;
 import org.springframework.dao.DataAccessException;
+import org.springframework.dao.OptimisticLockingFailureException;
 import org.springframework.data.cassandra.SessionFactory;
+import org.springframework.data.cassandra.core.EntityOperations.AdaptibleEntity;
 import org.springframework.data.cassandra.core.convert.CassandraConverter;
 import org.springframework.data.cassandra.core.convert.MappingCassandraConverter;
 import org.springframework.data.cassandra.core.convert.QueryMapper;
@@ -101,6 +104,8 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 
 	private final SpelAwareProxyProjectionFactory projectionFactory;
 
+	private final EntityOperations operations;
+
 	private final StatementFactory statementFactory;
 
 	private @Nullable ApplicationEventPublisher eventPublisher;
@@ -164,6 +169,7 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 		this.cqlOperations = cqlOperations;
 		this.mappingContext = converter.getMappingContext();
 		this.projectionFactory = new SpelAwareProxyProjectionFactory();
+		this.operations = new EntityOperations(converter.getMappingContext());
 		this.statementFactory = new StatementFactory(new QueryMapper(converter), new UpdateMapper(converter));
 	}
 
@@ -196,7 +202,7 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 	 */
 	@Override
 	public CqlIdentifier getTableName(Class<?> entityClass) {
-		return getRequiredPersistentEntity(entityClass).getTableName();
+		return operations.getTableName(entityClass);
 	}
 
 	/* (non-Javadoc)
@@ -557,19 +563,36 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 
 	<T> EntityWriteResult<T> doInsert(T entity, WriteOptions options, CqlIdentifier tableName) {
 
+		AdaptibleEntity<T> source = operations.forEntity(entity, converter.getConversionService());
 		CassandraPersistentEntity<?> persistentEntity = getRequiredPersistentEntity(entity.getClass());
 
-		Insert insert = EntityQueryUtils.createInsertQuery(tableName.toCql(), entity, options, getConverter(),
+		T entityToUse = source.isVersionedEntity() ? source.initializeVersionProperty() : entity;
+
+		Insert insert = EntityQueryUtils.createInsertQuery(tableName.toCql(), entityToUse, options, getConverter(),
 				persistentEntity);
 
-		maybeEmitEvent(new BeforeSaveEvent<>(entity, tableName, insert));
+		if (source.isVersionedEntity()) {
+			return doInsertVersioned(insert.ifNotExists(), entityToUse, source, tableName);
+		}
 
-		// noinspection ConstantConditions
-		WriteResult result = getCqlOperations().execute(new StatementCallback(insert));
+		return doInsert(insert, entityToUse, tableName);
+	}
 
-		maybeEmitEvent(new AfterSaveEvent<>(entity, tableName));
+	private <T> EntityWriteResult<T> doInsertVersioned(Insert insert, T entity, AdaptibleEntity<T> source,
+			CqlIdentifier tableName) {
 
-		return EntityWriteResult.of(result, entity);
+		return executeSave(entity, tableName, insert, result -> {
+
+			if (!result.wasApplied()) {
+				throw new OptimisticLockingFailureException(
+						String.format("Cannot insert entity %s with version, %s into table %s as it already exists", entity,
+								source.getVersion(), tableName));
+			}
+		});
+	}
+
+	private <T> EntityWriteResult<T> doInsert(Insert insert, T entity, CqlIdentifier tableName) {
+		return executeSave(entity, tableName, insert);
 	}
 
 	/* (non-Javadoc)
@@ -591,16 +614,39 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 
 		CassandraPersistentEntity<?> persistentEntity = getRequiredPersistentEntity(entity.getClass());
 		CqlIdentifier tableName = persistentEntity.getTableName();
+		AdaptibleEntity<T> source = operations.forEntity(entity, converter.getConversionService());
+
+		if (source.isVersionedEntity()) {
+			return doUpdateVersioned(source, options, tableName, persistentEntity);
+		}
+
+		return doUpdate(entity, options, tableName, persistentEntity);
+	}
+
+	private <T> EntityWriteResult<T> doUpdateVersioned(AdaptibleEntity<T> source, UpdateOptions options,
+			CqlIdentifier tableName, CassandraPersistentEntity<?> persistentEntity) {
+
+		Number previousVersion = source.getVersion();
+		T entity = source.incrementVersion();
+
 		Update update = getStatementFactory().update(entity, options, getConverter(), persistentEntity, tableName);
 
-		maybeEmitEvent(new BeforeSaveEvent<>(entity, tableName, update));
+		return executeSave(entity, tableName, source.appendVersionCondition(update, previousVersion), result -> {
 
-		// noinspection ConstantConditions
-		WriteResult result = getCqlOperations().execute(new StatementCallback(update));
+			if (!result.wasApplied()) {
+				throw new OptimisticLockingFailureException(
+						String.format("Cannot save entity %s with version %s to table %s. Has it been modified meanwhile?", entity,
+								source.getVersion(), tableName));
+			}
+		});
+	}
 
-		maybeEmitEvent(new AfterSaveEvent<>(entity, tableName));
+	private <T> EntityWriteResult<T> doUpdate(T entity, UpdateOptions options, CqlIdentifier tableName,
+			CassandraPersistentEntity<?> persistentEntity) {
 
-		return EntityWriteResult.of(result, entity);
+		Update update = getStatementFactory().update(entity, options, getConverter(), persistentEntity, tableName);
+
+		return executeSave(entity, tableName, update);
 	}
 
 	/* (non-Javadoc)
@@ -622,16 +668,32 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 
 		CassandraPersistentEntity<?> persistentEntity = getRequiredPersistentEntity(entity.getClass());
 		CqlIdentifier tableName = persistentEntity.getTableName();
+		AdaptibleEntity<Object> source = operations.forEntity(entity, converter.getConversionService());
+
 		Delete delete = getStatementFactory().delete(entity, options, getConverter(), persistentEntity, tableName);
 
-		maybeEmitEvent(new BeforeDeleteEvent<>(delete, entity.getClass(), tableName));
+		if (source.isVersionedEntity()) {
+			return doDeleteVersioned(delete, entity, source, tableName);
+		}
 
-		// noinspection ConstantConditions
-		WriteResult result = getCqlOperations().execute(new StatementCallback(delete));
+		return doDelete(delete, entity, tableName);
+	}
 
-		maybeEmitEvent(new AfterDeleteEvent<>(delete, entity.getClass(), tableName));
+	private WriteResult doDeleteVersioned(Delete delete, Object entity, AdaptibleEntity<Object> source,
+			CqlIdentifier tableName) {
 
-		return result;
+		return executeDelete(entity, tableName, source.appendVersionCondition(delete), result -> {
+
+			if (!result.wasApplied()) {
+				throw new OptimisticLockingFailureException(
+						String.format("Cannot delete entity %s with version, %s in table %s. Has it been modified meanwhile?",
+								entity, source.getVersion(), tableName));
+			}
+		});
+	}
+
+	private WriteResult doDelete(Delete delete, Object entity, CqlIdentifier tableName) {
+		return executeDelete(entity, tableName, delete, result -> {});
 	}
 
 	/* (non-Javadoc)
@@ -752,6 +814,36 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 		return this.statementFactory;
 	}
 
+	private <T> EntityWriteResult<T> executeSave(T entity, CqlIdentifier tableName, Statement statement) {
+		return executeSave(entity, tableName, statement, ignore -> {});
+	}
+
+	private <T> EntityWriteResult<T> executeSave(T entity, CqlIdentifier tableName, Statement statement,
+			Consumer<WriteResult> resultConsumer) {
+
+		maybeEmitEvent(new BeforeSaveEvent<>(entity, tableName, statement));
+
+		WriteResult result = getCqlOperations().execute(new StatementCallback(statement));
+		resultConsumer.accept(result);
+
+		maybeEmitEvent(new AfterSaveEvent<>(entity, tableName));
+
+		return EntityWriteResult.of(result, entity);
+	}
+
+	private WriteResult executeDelete(Object entity, CqlIdentifier tableName, Statement statement,
+			Consumer<WriteResult> resultConsumer) {
+
+		maybeEmitEvent(new BeforeDeleteEvent<>(statement, entity.getClass(), tableName));
+
+		WriteResult result = getCqlOperations().execute(new StatementCallback(statement));
+		resultConsumer.accept(result);
+
+		maybeEmitEvent(new AfterDeleteEvent<>(statement, entity.getClass(), tableName));
+
+		return result;
+	}
+
 	private CqlIdentifier getTableName(Object entity) {
 		return getRequiredPersistentEntity(entity.getClass()).getTableName();
 	}
@@ -790,15 +882,13 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 
 		return row -> {
 
-			maybeEmitEvent(new AfterLoadEvent(row, targetType, tableName));
+			maybeEmitEvent(new AfterLoadEvent<>(row, targetType, tableName));
 
 			Object source = getConverter().read(typeToRead, row);
 
 			T result = (T) (targetType.isInterface() ? getProjectionFactory().createProjection(targetType, source) : source);
 
-			if (result != null) {
-				maybeEmitEvent(new AfterConvertEvent<>(row, result, tableName));
-			}
+			maybeEmitEvent(new AfterConvertEvent<>(row, result, tableName));
 
 			return result;
 		};
@@ -834,7 +924,7 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 		 */
 		@Override
 		public WriteResult doInSession(Session session) throws DriverException, DataAccessException {
-			return WriteResult.of(session.execute(statement));
+			return WriteResult.of(session.execute(this.statement));
 		}
 
 		/* (non-Javadoc)
@@ -842,7 +932,7 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 		 */
 		@Override
 		public String getCql() {
-			return statement.toString();
+			return this.statement.toString();
 		}
 	}
 }

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

@@ -0,0 +1,319 @@
+/*
+ * 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
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.cassandra.core;
+
+import lombok.AccessLevel;
+import lombok.NonNull;
+import lombok.RequiredArgsConstructor;
+
+import org.springframework.core.convert.ConversionService;
+import org.springframework.data.cassandra.core.cql.CqlIdentifier;
+import org.springframework.data.cassandra.core.mapping.CassandraPersistentEntity;
+import org.springframework.data.cassandra.core.mapping.CassandraPersistentProperty;
+import org.springframework.data.mapping.PersistentPropertyAccessor;
+import org.springframework.data.mapping.context.MappingContext;
+import org.springframework.data.mapping.model.ConvertingPropertyAccessor;
+import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
+import org.springframework.util.ClassUtils;
+
+import com.datastax.driver.core.Statement;
+import com.datastax.driver.core.querybuilder.Delete;
+import com.datastax.driver.core.querybuilder.QueryBuilder;
+import com.datastax.driver.core.querybuilder.Update;
+
+/**
+ * Common operations performed on an entity in the context of it's mapping metadata.
+ *
+ * @author Mark Paluch
+ * @since 2.2
+ * @see CassandraTemplate
+ * @see AsyncCassandraTemplate
+ * @see ReactiveCassandraTemplate
+ */
+@RequiredArgsConstructor
+class EntityOperations {
+
+	private final @NonNull MappingContext<? extends CassandraPersistentEntity<?>, CassandraPersistentProperty> context;
+
+	private CassandraPersistentEntity<?> getRequiredPersistentEntity(Class<?> entityType) {
+		return context.getRequiredPersistentEntity(ClassUtils.getUserClass(entityType));
+	}
+
+	/**
+	 * Creates a new {@link Entity} for the given bean.
+	 *
+	 * @param entity must not be {@literal null}.
+	 * @return
+	 */
+	public <T> Entity<T> forEntity(T entity) {
+
+		Assert.notNull(entity, "Bean must not be null!");
+
+		return MappedEntity.of(entity, context);
+	}
+
+	/**
+	 * Returns the table name to which the entity shall be persisted.
+	 *
+	 * @param entityClass entity class, must not be {@literal null}.
+	 * @return the table name to which the entity shall be persisted.
+	 */
+	public CqlIdentifier getTableName(Class<?> entityClass) {
+		return getRequiredPersistentEntity(entityClass).getTableName();
+	}
+
+	/**
+	 * Creates a new {@link AdaptibleEntity} for the given bean and {@link ConversionService}.
+	 *
+	 * @param entity must not be {@literal null}.
+	 * @param conversionService must not be {@literal null}.
+	 * @return
+	 */
+	public <T> AdaptibleEntity<T> forEntity(T entity, ConversionService conversionService) {
+
+		Assert.notNull(entity, "Bean must not be null!");
+		Assert.notNull(conversionService, "ConversionService must not be null!");
+
+		return AdaptibleMappedEntity.of(entity, context, conversionService);
+	}
+
+	/**
+	 * A representation of information about an entity.
+	 */
+	interface Entity<T> {
+
+		/**
+		 * Returns whether the entity is versioned, i.e. if it contains a version property.
+		 *
+		 * @return
+		 */
+		default boolean isVersionedEntity() {
+			return false;
+		}
+
+		/**
+		 * Returns the value of the version if the entity has a version property, {@literal null} otherwise.
+		 *
+		 * @return
+		 */
+		@Nullable
+		Object getVersion();
+
+		/**
+		 * Returns the underlying bean.
+		 *
+		 * @return
+		 */
+		T getBean();
+
+		/**
+		 * Returns whether the entity is considered to be new.
+		 *
+		 * @return
+		 */
+		boolean isNew();
+	}
+
+	/**
+	 * Information and commands on an entity.
+	 */
+	interface AdaptibleEntity<T> extends Entity<T> {
+
+		/**
+		 * Appends a {@code IF} condition to an {@link Update} statement for optimistic locking to perform the update only
+		 * if the version number matches. This method accepts {@code currentVersionNumber} as the {@link Update} typically
+		 * requires to increment the version number upon assembly time.
+		 *
+		 * @param update the {@link Update} statement to append the condition to.
+		 * @param currentVersionNumber previous version number.
+		 * @return the altered {@link Update} containing the {@code IF} condition for optimistic locking.
+		 */
+		Statement appendVersionCondition(Update update, Number currentVersionNumber);
+
+		/**
+		 * Appends a {@code IF} condition to an {@link Delete} statement for optimistic locking to perform the delete only
+		 * if the version number matches. The {@link #getVersion() version number} is derived from the actual state as
+		 * delete statements typically do not increment the version prior to statement creation.
+		 *
+		 * @param delete the {@link Delete} statement to append the condition to.
+		 * @return the altered {@link Delete} containing the {@code IF} condition for optimistic locking.
+		 * @see #getVersion()
+		 */
+		Statement appendVersionCondition(Delete delete);
+
+		/**
+		 * Initializes the version property of the of the current entity if available.
+		 *
+		 * @return the entity with the version property updated if available.
+		 */
+		T initializeVersionProperty();
+
+		/**
+		 * Increments the value of the version property if available.
+		 *
+		 * @return the entity with the version property incremented if available.
+		 */
+		T incrementVersion();
+
+		/**
+		 * Returns the current version value if the entity has a version property.
+		 *
+		 * @return the current version or {@literal null} in case it's uninitialized or the entity doesn't expose a version
+		 *         property.
+		 */
+		@Nullable
+		Number getVersion();
+	}
+
+	@RequiredArgsConstructor(access = AccessLevel.PROTECTED)
+	private static class MappedEntity<T> implements Entity<T> {
+
+		private final @NonNull CassandraPersistentEntity<?> entity;
+		private final @NonNull PersistentPropertyAccessor<T> propertyAccessor;
+
+		private static <T> MappedEntity<T> of(T bean,
+				MappingContext<? extends CassandraPersistentEntity<?>, CassandraPersistentProperty> context) {
+
+			CassandraPersistentEntity<?> entity = context.getRequiredPersistentEntity(bean.getClass());
+			PersistentPropertyAccessor<T> propertyAccessor = entity.getPropertyAccessor(bean);
+
+			return new MappedEntity<>(entity, propertyAccessor);
+		}
+
+		/* (non-Javadoc)
+		 * @see org.springframework.data.cassandra.core.EntityOperations.Entity#isVersionedEntity()
+		 */
+		@Override
+		public boolean isVersionedEntity() {
+			return entity.hasVersionProperty();
+		}
+
+		/* (non-Javadoc)
+		 * @see org.springframework.data.cassandra.core.EntityOperations.Entity#getVersion()
+		 */
+		@Override
+		@Nullable
+		public Object getVersion() {
+			return propertyAccessor.getProperty(entity.getRequiredVersionProperty());
+		}
+
+		/* (non-Javadoc)
+		 * @see org.springframework.data.cassandra.core.EntityOperations.Entity#getBean()
+		 */
+		@Override
+		public T getBean() {
+			return propertyAccessor.getBean();
+		}
+
+		/* (non-Javadoc)
+		 * @see org.springframework.data.cassandra.core.EntityOperations.Entity#isNew()
+		 */
+		@Override
+		public boolean isNew() {
+			return entity.isNew(propertyAccessor.getBean());
+		}
+	}
+
+	private static class AdaptibleMappedEntity<T> extends MappedEntity<T> implements AdaptibleEntity<T> {
+
+		private final CassandraPersistentEntity<?> entity;
+		private final ConvertingPropertyAccessor<T> propertyAccessor;
+
+		private AdaptibleMappedEntity(CassandraPersistentEntity<?> entity, ConvertingPropertyAccessor<T> propertyAccessor) {
+
+			super(entity, propertyAccessor);
+
+			this.entity = entity;
+			this.propertyAccessor = propertyAccessor;
+		}
+
+		private static <T> AdaptibleEntity<T> of(T bean,
+				MappingContext<? extends CassandraPersistentEntity<?>, CassandraPersistentProperty> context,
+				ConversionService conversionService) {
+
+			CassandraPersistentEntity<?> entity = context.getRequiredPersistentEntity(bean.getClass());
+			PersistentPropertyAccessor<T> propertyAccessor = entity.getPropertyAccessor(bean);
+
+			return new AdaptibleMappedEntity<>(entity, new ConvertingPropertyAccessor<>(propertyAccessor, conversionService));
+		}
+
+		/* (non-Javadoc)
+		 * @see org.springframework.data.cassandra.core.EntityOperations.AdaptibleEntity#appendVersionCondition(com.datastax.driver.core.querybuilder.Update, java.lang.Number)
+		 */
+		@Override
+		public Statement appendVersionCondition(com.datastax.driver.core.querybuilder.Update update,
+				Number currentVersionNumber) {
+			return update.onlyIf(QueryBuilder.eq(getVersionColumnName().toCql(), currentVersionNumber));
+		}
+
+		/* (non-Javadoc)
+		 * @see org.springframework.data.cassandra.core.EntityOperations.AdaptibleEntity#appendVersionCondition(com.datastax.driver.core.querybuilder.Delete)
+		 */
+		@Override
+		public Statement appendVersionCondition(Delete delete) {
+			return delete.onlyIf(QueryBuilder.eq(getVersionColumnName().toCql(), getVersion()));
+		}
+
+		/* (non-Javadoc)
+		 * @see org.springframework.data.cassandra.core.EntityOperations.AdaptibleEntity#initializeVersionProperty()
+		 */
+		@Override
+		public T initializeVersionProperty() {
+
+			if (!entity.hasVersionProperty()) {
+				return propertyAccessor.getBean();
+			}
+
+			CassandraPersistentProperty versionProperty = entity.getRequiredVersionProperty();
+
+			propertyAccessor.setProperty(versionProperty, versionProperty.getType().isPrimitive() ? 1 : 0);
+
+			return propertyAccessor.getBean();
+		}
+
+		/* (non-Javadoc)
+		 * @see org.springframework.data.cassandra.core.EntityOperations.AdaptibleEntity#incrementVersion()
+		 */
+		@Override
+		public T incrementVersion() {
+
+			CassandraPersistentProperty versionProperty = entity.getRequiredVersionProperty();
+			Number version = getVersion();
+			Number nextVersion = version == null ? 0 : version.longValue() + 1;
+
+			propertyAccessor.setProperty(versionProperty, nextVersion);
+
+			return propertyAccessor.getBean();
+		}
+
+		/* (non-Javadoc)
+		 * @see org.springframework.data.cassandra.core.EntityOperations.MappedEntity#getVersion()
+		 */
+		@Override
+		@Nullable
+		public Number getVersion() {
+
+			CassandraPersistentProperty versionProperty = entity.getRequiredVersionProperty();
+
+			return propertyAccessor.getProperty(versionProperty, Number.class);
+		}
+
+		private CqlIdentifier getVersionColumnName() {
+			return entity.getRequiredVersionProperty().getColumnName();
+		}
+	}
+}

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

@@ -16,21 +16,25 @@
 package org.springframework.data.cassandra.core;
 
 import java.util.Collections;
+import java.util.function.BiConsumer;
 import java.util.function.Function;
 
 import lombok.Value;
 
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.core.publisher.SynchronousSink;
 
 import org.reactivestreams.Publisher;
 import org.springframework.context.ApplicationEvent;
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.context.ApplicationEventPublisherAware;
 import org.springframework.dao.DataAccessException;
+import org.springframework.dao.OptimisticLockingFailureException;
 import org.springframework.data.cassandra.ReactiveResultSet;
 import org.springframework.data.cassandra.ReactiveSession;
 import org.springframework.data.cassandra.ReactiveSessionFactory;
+import org.springframework.data.cassandra.core.EntityOperations.AdaptibleEntity;
 import org.springframework.data.cassandra.core.convert.CassandraConverter;
 import org.springframework.data.cassandra.core.convert.MappingCassandraConverter;
 import org.springframework.data.cassandra.core.convert.QueryMapper;
@@ -100,14 +104,16 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A
 
 	private final CassandraConverter converter;
 
-	private final MappingContext<? extends CassandraPersistentEntity<?>, CassandraPersistentProperty> mappingContext;
-
 	private final ReactiveCqlOperations cqlOperations;
 
-	private final StatementFactory statementFactory;
+	private final MappingContext<? extends CassandraPersistentEntity<?>, CassandraPersistentProperty> mappingContext;
 
 	private final SpelAwareProxyProjectionFactory projectionFactory;
 
+	private final EntityOperations operations;
+
+	private final StatementFactory statementFactory;
+
 	private @Nullable ApplicationEventPublisher eventPublisher;
 
 	/**
@@ -170,6 +176,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A
 		this.cqlOperations = reactiveCqlOperations;
 		this.mappingContext = this.converter.getMappingContext();
 		this.projectionFactory = new SpelAwareProxyProjectionFactory();
+		this.operations = new EntityOperations(converter.getMappingContext());
 		this.statementFactory = new StatementFactory(new QueryMapper(converter), new UpdateMapper(converter));
 	}
 
@@ -497,19 +504,39 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A
 
 	<T> Mono<EntityWriteResult<T>> doInsert(T entity, WriteOptions options, CqlIdentifier tableName) {
 
+		AdaptibleEntity<T> source = operations.forEntity(entity, converter.getConversionService());
 		CassandraPersistentEntity<?> persistentEntity = getRequiredPersistentEntity(entity.getClass());
 
-		Insert insert = EntityQueryUtils.createInsertQuery(tableName.toCql(), entity, options, getConverter(),
+		T entityToUse = source.isVersionedEntity() ? source.initializeVersionProperty() : entity;
+
+		Insert insert = EntityQueryUtils.createInsertQuery(tableName.toCql(), entityToUse, options, getConverter(),
 				persistentEntity);
 
-		// noinspection ConstantConditions
-		Mono<EntityWriteResult<T>> result = getReactiveCqlOperations() //
-				.execute(new StatementCallback(insert)) //
-				.doOnSubscribe(it -> maybeEmitEvent(new BeforeSaveEvent<>(entity, tableName, insert))) //
-				.map(it -> EntityWriteResult.of(it, entity)) //
-				.next();
+		if (source.isVersionedEntity()) {
+			return doInsertVersioned(insert.ifNotExists(), entityToUse, source, tableName);
+		}
 
-		return result.doOnNext(it -> maybeEmitEvent(new AfterSaveEvent<>(entity, tableName)));
+		return doInsert(insert, entityToUse, tableName);
+	}
+
+	private <T> Mono<EntityWriteResult<T>> doInsertVersioned(Insert insert, T entity, AdaptibleEntity<T> source,
+			CqlIdentifier tableName) {
+
+		return executeSave(entity, tableName, insert, (result, sink) -> {
+
+			if (!result.wasApplied()) {
+				sink.error(new OptimisticLockingFailureException(
+						String.format("Cannot insert entity %s with version, %s into table %s as it already exists", entity,
+								source.getVersion(), tableName)));
+				return;
+			}
+
+			sink.next(result);
+		});
+	}
+
+	private <T> Mono<EntityWriteResult<T>> doInsert(Insert insert, T entity, CqlIdentifier tableName) {
+		return executeSave(entity, tableName, insert);
 	}
 
 	/* (non-Javadoc)
@@ -517,7 +544,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A
 	 */
 	@Override
 	public <T> Mono<T> update(T entity) {
-		return update(entity, UpdateOptions.empty()).map(writeResult -> entity);
+		return update(entity, UpdateOptions.empty()).map(EntityWriteResult::getEntity);
 	}
 
 	/* (non-Javadoc)
@@ -531,15 +558,42 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A
 
 		CassandraPersistentEntity<?> persistentEntity = getRequiredPersistentEntity(entity.getClass());
 		CqlIdentifier tableName = persistentEntity.getTableName();
+		AdaptibleEntity<T> source = operations.forEntity(entity, converter.getConversionService());
+
+		if (source.isVersionedEntity()) {
+			return doUpdateVersioned(source, options, tableName, persistentEntity);
+		}
+
+		return doUpdate(entity, options, tableName, persistentEntity);
+	}
+
+	private <T> Mono<EntityWriteResult<T>> doUpdateVersioned(AdaptibleEntity<T> source, UpdateOptions options,
+			CqlIdentifier tableName, CassandraPersistentEntity<?> persistentEntity) {
+
+		Number previousVersion = source.getVersion();
+		T entity = source.incrementVersion();
+
 		Update update = getStatementFactory().update(entity, options, getConverter(), persistentEntity, tableName);
 
-		Mono<EntityWriteResult<T>> result = getReactiveCqlOperations() //
-				.execute(new StatementCallback(update)) //
-				.doOnSubscribe(it -> maybeEmitEvent(new BeforeSaveEvent<>(entity, tableName, update))) //
-				.map(it -> EntityWriteResult.of(it, entity)) //
-				.next();
+		return executeSave(entity, tableName, source.appendVersionCondition(update, previousVersion), (result, sink) -> {
 
-		return result.doOnNext(it -> maybeEmitEvent(new AfterSaveEvent<>(entity, tableName)));
+			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,
+								source.getVersion(), tableName)));
+				return;
+			}
+
+			sink.next(result);
+		});
+	}
+
+	private <T> Mono<EntityWriteResult<T>> doUpdate(T entity, UpdateOptions options, CqlIdentifier tableName,
+			CassandraPersistentEntity<?> persistentEntity) {
+
+		Update update = getStatementFactory().update(entity, options, getConverter(), persistentEntity, tableName);
+
+		return executeSave(entity, tableName, update);
 	}
 
 	/* (non-Javadoc)
@@ -561,14 +615,35 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A
 
 		CassandraPersistentEntity<?> persistentEntity = getRequiredPersistentEntity(entity.getClass());
 		CqlIdentifier tableName = persistentEntity.getTableName();
+		AdaptibleEntity<Object> source = operations.forEntity(entity, converter.getConversionService());
+
 		Delete delete = getStatementFactory().delete(entity, options, getConverter(), persistentEntity, tableName);
 
-		Mono<WriteResult> result = getReactiveCqlOperations() //
-				.execute(new StatementCallback(delete)) //
-				.doOnSubscribe(it -> maybeEmitEvent(new BeforeDeleteEvent<>(delete, entity.getClass(), tableName))) //
-				.next();
+		if (source.isVersionedEntity()) {
+			return doDeleteVersioned(delete, entity, source, tableName);
+		}
 
-		return result.doOnNext(it -> maybeEmitEvent(new AfterDeleteEvent<>(delete, entity.getClass(), tableName)));
+		return doDelete(delete, entity, tableName);
+	}
+
+	private Mono<WriteResult> doDeleteVersioned(Delete delete, Object entity, AdaptibleEntity<Object> source,
+			CqlIdentifier tableName) {
+
+		return executeDelete(entity, tableName, source.appendVersionCondition(delete), (result, sink) -> {
+
+			if (!result.wasApplied()) {
+				sink.error(new OptimisticLockingFailureException(
+						String.format("Cannot delete entity %s with version, %s in table %s. Has it been modified meanwhile?",
+								entity, source.getVersion(), tableName)));
+				return;
+			}
+
+			sink.next(result);
+		});
+	}
+
+	private Mono<WriteResult> doDelete(Delete delete, Object entity, CqlIdentifier tableName) {
+		return executeDelete(entity, tableName, delete, (result, sink) -> sink.next(result));
 	}
 
 	/* (non-Javadoc)
@@ -685,7 +760,37 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A
 	}
 
 	CqlIdentifier getTableName(Class<?> entityClass) {
-		return getRequiredPersistentEntity(entityClass).getTableName();
+		return operations.getTableName(entityClass);
+	}
+
+	private <T> Mono<EntityWriteResult<T>> executeSave(T entity, CqlIdentifier tableName, Statement statement) {
+		return executeSave(entity, tableName, statement, (writeResult, sink) -> sink.next(writeResult));
+	}
+
+	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));
+
+		Flux<WriteResult> execute = getReactiveCqlOperations().execute(new StatementCallback(statement));
+
+		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,
+			BiConsumer<WriteResult, SynchronousSink<WriteResult>> handler) {
+
+		maybeEmitEvent(new BeforeDeleteEvent<>(statement, entity.getClass(), tableName));
+
+		Flux<WriteResult> execute = getReactiveCqlOperations().execute(new StatementCallback(statement));
+
+		return execute.map(it -> EntityWriteResult.of(it, entity)).handle(handler) //
+				.doOnSubscribe(it -> maybeEmitEvent(new BeforeSaveEvent<>(entity, tableName, statement))) //
+				.doOnNext(it -> maybeEmitEvent(new AfterDeleteEvent<>(statement, entity.getClass(), tableName))) //
+				.next();
 	}
 
 	private CqlIdentifier getTableName(Object entity) {

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/MappingCassandraEntityInformation.java

@@ -21,7 +21,7 @@ import org.springframework.data.cassandra.core.mapping.CassandraPersistentEntity
 import org.springframework.data.cassandra.core.mapping.CassandraPersistentProperty;
 import org.springframework.data.cassandra.core.mapping.MapId;
 import org.springframework.data.cassandra.repository.query.CassandraEntityInformation;
-import org.springframework.data.repository.core.support.AbstractEntityInformation;
+import org.springframework.data.repository.core.support.PersistentEntityInformation;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
@@ -33,7 +33,7 @@ import org.springframework.util.Assert;
  * @author Matthew T. Adams
  * @author Mark Paluch
  */
-public class MappingCassandraEntityInformation<T, ID> extends AbstractEntityInformation<T, ID>
+public class MappingCassandraEntityInformation<T, ID> extends PersistentEntityInformation<T, ID>
 		implements CassandraEntityInformation<T, ID> {
 
 	private final CassandraPersistentEntity<T> entityMetadata;
@@ -47,7 +47,7 @@ public class MappingCassandraEntityInformation<T, ID> extends AbstractEntityInfo
 	 */
 	public MappingCassandraEntityInformation(CassandraPersistentEntity<T> entity, CassandraConverter converter) {
 
-		super(entity.getType());
+		super(entity);
 
 		this.entityMetadata = entity;
 		this.converter = converter;

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/SimpleCassandraRepository.java

@@ -24,11 +24,14 @@ import java.util.Optional;
 import org.springframework.data.cassandra.core.CassandraOperations;
 import org.springframework.data.cassandra.core.CassandraTemplate;
 import org.springframework.data.cassandra.core.InsertOptions;
+import org.springframework.data.cassandra.core.mapping.BasicCassandraPersistentEntity;
+import org.springframework.data.cassandra.core.mapping.CassandraPersistentProperty;
 import org.springframework.data.cassandra.core.query.Query;
 import org.springframework.data.cassandra.repository.CassandraRepository;
 import org.springframework.data.cassandra.repository.query.CassandraEntityInformation;
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.Slice;
+import org.springframework.data.mapping.context.AbstractMappingContext;
 import org.springframework.data.util.StreamUtils;
 import org.springframework.data.util.Streamable;
 import org.springframework.util.Assert;
@@ -53,6 +56,8 @@ public class SimpleCassandraRepository<T, ID> implements CassandraRepository<T,
 
 	private final CassandraOperations operations;
 
+	private final AbstractMappingContext<BasicCassandraPersistentEntity<?>, CassandraPersistentProperty> mappingContext;
+
 	/**
 	 * Create a new {@link SimpleCassandraRepository} for the given {@link CassandraEntityInformation} and
 	 * {@link CassandraTemplate}.
@@ -67,6 +72,7 @@ public class SimpleCassandraRepository<T, ID> implements CassandraRepository<T,
 
 		this.entityInformation = metadata;
 		this.operations = operations;
+		this.mappingContext = operations.getConverter().getMappingContext();
 	}
 
 	/* (non-Javadoc)
@@ -77,6 +83,14 @@ public class SimpleCassandraRepository<T, ID> implements CassandraRepository<T,
 
 		Assert.notNull(entity, "Entity must not be null");
 
+		BasicCassandraPersistentEntity<?> persistentEntity = mappingContext.getPersistentEntity(entity.getClass());
+		if (persistentEntity != null && persistentEntity.hasVersionProperty()) {
+
+			if (!entityInformation.isNew(entity)) {
+				return operations.update(entity);
+			}
+		}
+
 		return operations.insert(entity, INSERT_NULLS).getEntity();
 	}
 
@@ -91,7 +105,7 @@ public class SimpleCassandraRepository<T, ID> implements CassandraRepository<T,
 		List<S> result = new ArrayList<>();
 
 		for (S entity : entities) {
-			result.add(operations.insert(entity, INSERT_NULLS).getEntity());
+			result.add(save(entity));
 		}
 
 		return result;

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/SimpleReactiveCassandraRepository.java

@@ -19,10 +19,14 @@ import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 import org.reactivestreams.Publisher;
+import org.springframework.data.cassandra.core.EntityWriteResult;
 import org.springframework.data.cassandra.core.InsertOptions;
 import org.springframework.data.cassandra.core.ReactiveCassandraOperations;
+import org.springframework.data.cassandra.core.mapping.BasicCassandraPersistentEntity;
+import org.springframework.data.cassandra.core.mapping.CassandraPersistentProperty;
 import org.springframework.data.cassandra.repository.ReactiveCassandraRepository;
 import org.springframework.data.cassandra.repository.query.CassandraEntityInformation;
+import org.springframework.data.mapping.context.AbstractMappingContext;
 import org.springframework.util.Assert;
 
 import com.datastax.driver.core.querybuilder.Insert;
@@ -44,6 +48,8 @@ public class SimpleReactiveCassandraRepository<T, ID> implements ReactiveCassand
 
 	private final ReactiveCassandraOperations operations;
 
+	private final AbstractMappingContext<BasicCassandraPersistentEntity<?>, CassandraPersistentProperty> mappingContext;
+
 	/**
 	 * Create a new {@link SimpleReactiveCassandraRepository} for the given {@link CassandraEntityInformation} and
 	 * {@link ReactiveCassandraOperations}.
@@ -59,6 +65,7 @@ public class SimpleReactiveCassandraRepository<T, ID> implements ReactiveCassand
 
 		this.entityInformation = metadata;
 		this.operations = operations;
+		this.mappingContext = operations.getConverter().getMappingContext();
 	}
 
 	/* (non-Javadoc)
@@ -69,7 +76,15 @@ public class SimpleReactiveCassandraRepository<T, ID> implements ReactiveCassand
 
 		Assert.notNull(entity, "Entity must not be null");
 
-		return operations.insert(entity, INSERT_NULLS).thenReturn(entity);
+		BasicCassandraPersistentEntity<?> persistentEntity = mappingContext.getPersistentEntity(entity.getClass());
+		if (persistentEntity != null && persistentEntity.hasVersionProperty()) {
+
+			if (!entityInformation.isNew(entity)) {
+				return operations.update(entity);
+			}
+		}
+
+		return operations.insert(entity, INSERT_NULLS).map(EntityWriteResult::getEntity);
 	}
 
 	/**
@@ -103,8 +118,7 @@ public class SimpleReactiveCassandraRepository<T, ID> implements ReactiveCassand
 
 		Assert.notNull(entityStream, "The given Publisher of entities must not be null");
 
-		return Flux.from(entityStream)
-				.flatMap(entity -> operations.insert(entity, INSERT_NULLS).thenReturn(entity));
+		return Flux.from(entityStream).flatMap(this::save);
 	}
 
 	/* (non-Javadoc)

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/config/CassandraSessionFactoryBeanUnitTests.java

@@ -16,8 +16,7 @@
 package org.springframework.data.cassandra.config;
 
 import static org.assertj.core.api.Assertions.*;
-import static org.mockito.ArgumentMatchers.anyBoolean;
-import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.ArgumentMatchers.*;
 import static org.mockito.Mockito.*;
 import static org.springframework.data.cassandra.config.CassandraSessionFactoryBean.*;
 
@@ -29,6 +28,7 @@ import org.junit.runner.RunWith;
 import org.mockito.Mock;
 import org.mockito.junit.MockitoJUnitRunner;
 import org.springframework.data.cassandra.core.convert.CassandraConverter;
+import org.springframework.data.cassandra.core.mapping.CassandraMappingContext;
 
 import com.datastax.driver.core.Cluster;
 import com.datastax.driver.core.Session;
@@ -55,6 +55,7 @@ public class CassandraSessionFactoryBeanUnitTests {
 	public void setup() {
 
 		when(mockCluster.connect()).thenReturn(mockSession);
+		when(mockConverter.getMappingContext()).thenReturn(new CassandraMappingContext());
 
 		factoryBean = spy(new CassandraSessionFactoryBean());
 		factoryBean.setCluster(mockCluster);

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

@@ -0,0 +1,161 @@
+/*
+ * 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
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.cassandra.core;
+
+import static org.assertj.core.api.Assertions.*;
+
+import lombok.Data;
+import lombok.experimental.Wither;
+
+import java.util.concurrent.Future;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.springframework.dao.OptimisticLockingFailureException;
+import org.springframework.data.annotation.Id;
+import org.springframework.data.annotation.PersistenceConstructor;
+import org.springframework.data.annotation.Version;
+import org.springframework.data.cassandra.core.convert.MappingCassandraConverter;
+import org.springframework.data.cassandra.core.cql.CqlTemplate;
+import org.springframework.data.cassandra.core.query.Query;
+import org.springframework.data.cassandra.repository.support.SchemaTestUtils;
+import org.springframework.data.cassandra.test.util.AbstractKeyspaceCreatingIntegrationTest;
+
+/**
+ * Integration tests for optimistic locking through {@link AsyncCassandraTemplate}.
+ *
+ * @author Mark Paluch
+ */
+public class AsyncOptimisticLockingIntegrationTests extends AbstractKeyspaceCreatingIntegrationTest {
+
+	AsyncCassandraTemplate template;
+
+	@Before
+	public void setUp() {
+
+		MappingCassandraConverter converter = new MappingCassandraConverter();
+		converter.afterPropertiesSet();
+
+		template = new AsyncCassandraTemplate(session, converter);
+
+		CassandraTemplate syncTemplate = new CassandraTemplate(new CqlTemplate(session), converter);
+
+		SchemaTestUtils.potentiallyCreateTableFor(VersionedEntity.class, syncTemplate);
+		SchemaTestUtils.truncate(VersionedEntity.class, syncTemplate);
+	}
+
+	@Test // DATACASS-576
+	public void shouldInsertVersioned() {
+
+		VersionedEntity versionedEntity = new VersionedEntity(42);
+
+		VersionedEntity saved = getUninterruptibly(template.insert(versionedEntity));
+		VersionedEntity loaded = getUninterruptibly(template.selectOne(Query.empty(), VersionedEntity.class));
+
+		assertThat(saved.version).isEqualTo(1);
+		assertThat(loaded).isNotNull();
+		assertThat(loaded.version).isEqualTo(1);
+	}
+
+	@Test // DATACASS-576
+	public void duplicateInsertShouldFail() {
+
+		getUninterruptibly(template.insert(new VersionedEntity(42)));
+
+		assertThatThrownBy(() -> getUninterruptibly(template.insert(new VersionedEntity(42))))
+				.hasRootCauseInstanceOf(OptimisticLockingFailureException.class);
+	}
+
+	@Test // DATACASS-576
+	public void shouldUpdateVersioned() {
+
+		VersionedEntity versionedEntity = new VersionedEntity(42);
+
+		VersionedEntity saved = getUninterruptibly(template.insert(versionedEntity));
+		VersionedEntity updated = getUninterruptibly(template.update(saved));
+		VersionedEntity loaded = getUninterruptibly(template.selectOne(Query.empty(), VersionedEntity.class));
+
+		assertThat(saved.version).isEqualTo(1);
+		assertThat(updated.version).isEqualTo(2);
+		assertThat(loaded).isNotNull();
+		assertThat(loaded.version).isEqualTo(2);
+	}
+
+	@Test // DATACASS-576
+	public void updateForOutdatedEntityShouldFail() {
+
+		VersionedEntity versionedEntity = new VersionedEntity(42);
+
+		getUninterruptibly(template.insert(versionedEntity));
+		assertThatThrownBy(() -> getUninterruptibly(template.update(new VersionedEntity(42, 5, "f"))))
+				.hasRootCauseInstanceOf(OptimisticLockingFailureException.class);
+	}
+
+	@Test // DATACASS-576
+	public void shouldDeleteVersionedEntity() {
+
+		VersionedEntity versionedEntity = new VersionedEntity(42);
+
+		VersionedEntity saved = getUninterruptibly(template.insert(versionedEntity));
+		getUninterruptibly(template.delete(saved));
+
+		VersionedEntity loaded = getUninterruptibly(template.selectOne(Query.empty(), VersionedEntity.class));
+		assertThat(loaded).isNull();
+	}
+
+	@Test // DATACASS-576
+	public void deleteForOutdatedEntityShouldFail() {
+
+		getUninterruptibly(template.insert(new VersionedEntity(42)));
+
+		assertThatThrownBy(() -> getUninterruptibly(template.delete(new VersionedEntity(42))))
+				.hasRootCauseInstanceOf(OptimisticLockingFailureException.class);
+
+		VersionedEntity loaded = getUninterruptibly(template.selectOne(Query.empty(), VersionedEntity.class));
+		assertThat(loaded).isNotNull();
+	}
+
+	private static <T> T getUninterruptibly(Future<T> future) {
+
+		try {
+			return future.get();
+		} catch (Exception cause) {
+			throw new IllegalStateException(cause);
+		}
+	}
+
+	@Data
+	@Wither
+	static class VersionedEntity {
+
+		@Id final long id;
+
+		@Version final long version;
+
+		final String name;
+
+		public VersionedEntity(long id) {
+			this(id, 0, null);
+		}
+
+		@PersistenceConstructor
+		public VersionedEntity(long id, long version, String name) {
+			this.id = id;
+			this.version = version;
+			this.name = name;
+		}
+	}
+}

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

@@ -0,0 +1,146 @@
+/*
+ * 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
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.cassandra.core;
+
+import static org.assertj.core.api.Assertions.*;
+
+import lombok.Data;
+import lombok.experimental.Wither;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.springframework.dao.OptimisticLockingFailureException;
+import org.springframework.data.annotation.Id;
+import org.springframework.data.annotation.PersistenceConstructor;
+import org.springframework.data.annotation.Version;
+import org.springframework.data.cassandra.core.convert.MappingCassandraConverter;
+import org.springframework.data.cassandra.repository.support.SchemaTestUtils;
+import org.springframework.data.cassandra.test.util.AbstractKeyspaceCreatingIntegrationTest;
+
+/**
+ * Integration tests for optimistic locking through {@link CassandraTemplate}.
+ *
+ * @author Mark Paluch
+ */
+public class OptimisticLockingIntegrationTests extends AbstractKeyspaceCreatingIntegrationTest {
+
+	CassandraTemplate template;
+
+	@Before
+	public void setUp() {
+
+		MappingCassandraConverter converter = new MappingCassandraConverter();
+		converter.afterPropertiesSet();
+
+		template = new CassandraTemplate(session, converter);
+
+		SchemaTestUtils.potentiallyCreateTableFor(VersionedEntity.class, template);
+		SchemaTestUtils.truncate(VersionedEntity.class, template);
+	}
+
+	@Test // DATACASS-576
+	public void shouldInsertVersioned() {
+
+		VersionedEntity versionedEntity = new VersionedEntity(42);
+
+		VersionedEntity saved = template.insert(versionedEntity);
+		VersionedEntity loaded = template.query(VersionedEntity.class).firstValue();
+
+		assertThat(saved.version).isEqualTo(1);
+		assertThat(loaded).isNotNull();
+		assertThat(loaded.version).isEqualTo(1);
+	}
+
+	@Test // DATACASS-576
+	public void duplicateInsertShouldFail() {
+
+		template.insert(new VersionedEntity(42));
+
+		assertThatThrownBy(() -> template.insert(new VersionedEntity(42)))
+				.isInstanceOf(OptimisticLockingFailureException.class);
+	}
+
+	@Test // DATACASS-576
+	public void shouldUpdateVersioned() {
+
+		VersionedEntity versionedEntity = new VersionedEntity(42);
+
+		VersionedEntity saved = template.insert(versionedEntity);
+		VersionedEntity updated = template.update(saved);
+		VersionedEntity loaded = template.query(VersionedEntity.class).firstValue();
+
+		assertThat(saved.version).isEqualTo(1);
+		assertThat(updated.version).isEqualTo(2);
+		assertThat(loaded).isNotNull();
+		assertThat(loaded.version).isEqualTo(2);
+	}
+
+	@Test // DATACASS-576
+	public void updateForOutdatedEntityShouldFail() {
+
+		VersionedEntity versionedEntity = new VersionedEntity(42);
+
+		template.insert(versionedEntity);
+		assertThatThrownBy(() -> template.update(new VersionedEntity(42, 5, "f")))
+				.isInstanceOf(OptimisticLockingFailureException.class);
+	}
+
+	@Test // DATACASS-576
+	public void shouldDeleteVersionedEntity() {
+
+		VersionedEntity versionedEntity = new VersionedEntity(42);
+
+		VersionedEntity saved = template.insert(versionedEntity);
+		template.delete(saved);
+
+		VersionedEntity loaded = template.query(VersionedEntity.class).firstValue();
+		assertThat(loaded).isNull();
+	}
+
+	@Test // DATACASS-576
+	public void deleteForOutdatedEntityShouldFail() {
+
+		template.insert(new VersionedEntity(42));
+
+		assertThatThrownBy(() -> template.delete(new VersionedEntity(42)))
+				.isInstanceOf(OptimisticLockingFailureException.class);
+
+		VersionedEntity loaded = template.query(VersionedEntity.class).firstValue();
+		assertThat(loaded).isNotNull();
+	}
+
+	@Data
+	@Wither
+	static class VersionedEntity {
+
+		@Id final long id;
+
+		@Version final long version;
+
+		final String name;
+
+		public VersionedEntity(long id) {
+			this(id, 0, null);
+		}
+
+		@PersistenceConstructor
+		public VersionedEntity(long id, long version, String name) {
+			this.id = id;
+			this.version = version;
+			this.name = name;
+		}
+	}
+}

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

@@ -0,0 +1,180 @@
+/*
+ * 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
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.cassandra.core;
+
+import static org.assertj.core.api.Assertions.*;
+
+import lombok.Data;
+import lombok.experimental.Wither;
+import reactor.test.StepVerifier;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.springframework.dao.OptimisticLockingFailureException;
+import org.springframework.data.annotation.Id;
+import org.springframework.data.annotation.PersistenceConstructor;
+import org.springframework.data.annotation.Version;
+import org.springframework.data.cassandra.core.convert.MappingCassandraConverter;
+import org.springframework.data.cassandra.core.cql.CqlTemplate;
+import org.springframework.data.cassandra.core.cql.session.DefaultBridgedReactiveSession;
+import org.springframework.data.cassandra.core.query.Query;
+import org.springframework.data.cassandra.repository.support.SchemaTestUtils;
+import org.springframework.data.cassandra.test.util.AbstractKeyspaceCreatingIntegrationTest;
+
+/**
+ * Integration tests for optimistic locking through {@link ReactiveCassandraTemplate}.
+ *
+ * @author Mark Paluch
+ */
+public class ReactiveOptimisticLockingIntegrationTests extends AbstractKeyspaceCreatingIntegrationTest {
+
+	ReactiveCassandraTemplate template;
+
+	@Before
+	public void setUp() {
+
+		MappingCassandraConverter converter = new MappingCassandraConverter();
+		converter.afterPropertiesSet();
+
+		template = new ReactiveCassandraTemplate(new DefaultBridgedReactiveSession(session), converter);
+
+		CassandraTemplate syncTemplate = new CassandraTemplate(new CqlTemplate(session), converter);
+
+		SchemaTestUtils.potentiallyCreateTableFor(VersionedEntity.class, syncTemplate);
+		SchemaTestUtils.truncate(VersionedEntity.class, syncTemplate);
+	}
+
+	@Test // DATACASS-576
+	public void shouldInsertVersioned() {
+
+		VersionedEntity versionedEntity = new VersionedEntity(42);
+
+		template.insert(versionedEntity) //
+				.as(StepVerifier::create) //
+				.consumeNextWith(actual -> {
+
+					assertThat(actual.version).isEqualTo(1);
+				}).verifyComplete();
+
+		template.selectOne(Query.empty(), VersionedEntity.class) //
+				.as(StepVerifier::create) //
+				.consumeNextWith(actual -> {
+
+					assertThat(actual.version).isEqualTo(1);
+				}).verifyComplete();
+	}
+
+	@Test // DATACASS-576
+	public void duplicateInsertShouldFail() {
+
+		template.insert(new VersionedEntity(42)) //
+				.as(StepVerifier::create) //
+				.expectNextCount(1) //
+				.verifyComplete();
+
+		template.insert(new VersionedEntity(42)) //
+				.as(StepVerifier::create) //
+				.verifyError(OptimisticLockingFailureException.class);
+	}
+
+	@Test // DATACASS-576
+	public void shouldUpdateVersioned() {
+
+		VersionedEntity versionedEntity = new VersionedEntity(42);
+
+		template.insert(versionedEntity).flatMap(template::update) //
+				.as(StepVerifier::create) //
+				.consumeNextWith(actual -> {
+
+					assertThat(actual.version).isEqualTo(2);
+				}).verifyComplete();
+
+		template.selectOne(Query.empty(), VersionedEntity.class) //
+				.as(StepVerifier::create) //
+				.consumeNextWith(actual -> {
+
+					assertThat(actual.version).isEqualTo(2);
+				}).verifyComplete();
+	}
+
+	@Test // DATACASS-576
+	public void updateForOutdatedEntityShouldFail() {
+
+		template.insert(new VersionedEntity(42)) //
+				.as(StepVerifier::create) //
+				.expectNextCount(1) //
+				.verifyComplete();
+
+		template.update(new VersionedEntity(42, 5, "f")) //
+				.as(StepVerifier::create) //
+				.verifyError(OptimisticLockingFailureException.class);
+	}
+
+	@Test // DATACASS-576
+	public void shouldDeleteVersionedEntity() {
+
+		VersionedEntity versionedEntity = new VersionedEntity(42);
+
+		template.insert(versionedEntity).flatMap(template::delete) //
+				.as(StepVerifier::create) //
+				.expectNextCount(1) //
+				.verifyComplete();
+
+		template.query(VersionedEntity.class).first() //
+				.as(StepVerifier::create) //
+				.verifyComplete();
+	}
+
+	@Test // DATACASS-576
+	public void deleteForOutdatedEntityShouldFail() {
+
+		template.insert(new VersionedEntity(42))//
+				.as(StepVerifier::create) //
+				.expectNextCount(1) //
+				.verifyComplete();
+
+		template.delete(new VersionedEntity(42)) //
+				.as(StepVerifier::create) //
+				.verifyError(OptimisticLockingFailureException.class);
+
+		template.query(VersionedEntity.class).first() //
+				.as(StepVerifier::create) //
+				.expectNextCount(1) //
+				.verifyComplete();
+	}
+
+	@Data
+	@Wither
+	static class VersionedEntity {
+
+		@Id final long id;
+
+		@Version final long version;
+
+		final String name;
+
+		public VersionedEntity(long id) {
+			this(id, 0, null);
+		}
+
+		@PersistenceConstructor
+		public VersionedEntity(long id, long version, String name) {
+			this.id = id;
+			this.version = version;
+			this.name = name;
+		}
+	}
+}

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/support/CassandraRepositoryFactoryUnitTests.java

@@ -61,7 +61,6 @@ public class CassandraRepositoryFactoryUnitTests {
 	public void usesMappingCassandraEntityInformationIfMappingContextSet() {
 
 		when(mappingContext.getRequiredPersistentEntity(Person.class)).thenReturn(entity);
-		when(entity.getType()).thenReturn(Person.class);
 
 		CassandraRepositoryFactory repositoryFactory = new CassandraRepositoryFactory(template);
 
@@ -75,7 +74,6 @@ public class CassandraRepositoryFactoryUnitTests {
 	public void createsRepositoryWithIdTypeLong() {
 
 		when(mappingContext.getRequiredPersistentEntity(Person.class)).thenReturn(entity);
-		when(entity.getType()).thenReturn(Person.class);
 
 		CassandraRepositoryFactory repositoryFactory = new CassandraRepositoryFactory(template);
 		MyPersonRepository repository = repositoryFactory.getRepository(MyPersonRepository.class);

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/support/ReactiveCassandraRepositoryFactoryUnitTests.java

@@ -60,7 +60,6 @@ public class ReactiveCassandraRepositoryFactoryUnitTests {
 	public void usesMappingCassandraEntityInformationIfMappingContextSet() {
 
 		when(mappingContext.getRequiredPersistentEntity(Person.class)).thenReturn(entity);
-		when(entity.getType()).thenReturn(Person.class);
 
 		ReactiveCassandraRepositoryFactory repositoryFactory = new ReactiveCassandraRepositoryFactory(template);
 
@@ -74,7 +73,6 @@ public class ReactiveCassandraRepositoryFactoryUnitTests {
 	public void createsRepositoryWithIdTypeLong() {
 
 		when(mappingContext.getRequiredPersistentEntity(Person.class)).thenReturn(entity);
-		when(entity.getType()).thenReturn(Person.class);
 
 		ReactiveCassandraRepositoryFactory repositoryFactory = new ReactiveCassandraRepositoryFactory(template);
 		MyPersonRepository repository = repositoryFactory.getRepository(MyPersonRepository.class);

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/support/SimpleCassandraRepositoryUnitTests.java

@@ -29,6 +29,7 @@ import org.mockito.Captor;
 import org.mockito.Mock;
 import org.mockito.junit.MockitoJUnitRunner;
 import org.springframework.data.annotation.Id;
+import org.springframework.data.annotation.Version;
 import org.springframework.data.cassandra.core.CassandraOperations;
 import org.springframework.data.cassandra.core.EntityWriteResult;
 import org.springframework.data.cassandra.core.InsertOptions;
@@ -71,6 +72,7 @@ public class SimpleCassandraRepositoryUnitTests {
 	@Before
 	public void before() {
 		mappingContext.setUserTypeResolver(userTypeResolver);
+		when(cassandraOperations.getConverter()).thenReturn(converter);
 	}
 
 	@Test // DATACASS-428, DATACASS-560, DATACASS-573
@@ -91,6 +93,41 @@ public class SimpleCassandraRepositoryUnitTests {
 		verify(cassandraOperations).insert(person, InsertOptions.builder().withInsertNulls().build());
 	}
 
+	@Test // DATACASS-576
+	public void shouldInsertNewVersionedEntity() {
+
+		when(cassandraOperations.insert(any(), any(InsertOptions.class))).thenReturn(writeResult);
+
+		CassandraPersistentEntity<?> entity = converter.getMappingContext()
+				.getRequiredPersistentEntity(VersionedPerson.class);
+
+		repository = new SimpleCassandraRepository<Object, String>(new MappingCassandraEntityInformation(entity, converter),
+				cassandraOperations);
+
+		VersionedPerson versionedPerson = new VersionedPerson();
+
+		repository.save(versionedPerson);
+
+		verify(cassandraOperations).insert(versionedPerson, InsertOptions.builder().withInsertNulls().build());
+	}
+
+	@Test // DATACASS-576
+	public void shouldUpdateExistingVersionedEntity() {
+
+		CassandraPersistentEntity<?> entity = converter.getMappingContext()
+				.getRequiredPersistentEntity(VersionedPerson.class);
+
+		repository = new SimpleCassandraRepository<Object, String>(new MappingCassandraEntityInformation(entity, converter),
+				cassandraOperations);
+
+		VersionedPerson versionedPerson = new VersionedPerson();
+		versionedPerson.setVersion(2);
+
+		repository.save(versionedPerson);
+
+		verify(cassandraOperations).update(versionedPerson);
+	}
+
 	@Test // DATACASS-428, DATACASS-560, DATACASS-573
 	public void saveShouldUpdateNewEntity() {
 
@@ -166,4 +203,12 @@ public class SimpleCassandraRepositoryUnitTests {
 
 		@Id String id;
 	}
+
+	@Data
+	static class VersionedPerson {
+
+		@Id String id;
+		@Version long version;
+	}
+
 }

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/support/SimpleReactiveCassandraRepositoryUnitTests.java

@@ -0,0 +1,114 @@
+/*
+ * 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
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.cassandra.repository.support;
+
+import static org.mockito.Mockito.*;
+
+import lombok.Data;
+import reactor.core.publisher.Mono;
+
+import java.io.Serializable;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Captor;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+import org.springframework.data.annotation.Id;
+import org.springframework.data.annotation.Version;
+import org.springframework.data.cassandra.core.EntityWriteResult;
+import org.springframework.data.cassandra.core.InsertOptions;
+import org.springframework.data.cassandra.core.ReactiveCassandraOperations;
+import org.springframework.data.cassandra.core.convert.MappingCassandraConverter;
+import org.springframework.data.cassandra.core.mapping.CassandraMappingContext;
+import org.springframework.data.cassandra.core.mapping.CassandraPersistentEntity;
+import org.springframework.data.cassandra.core.mapping.UserTypeResolver;
+
+import com.datastax.driver.core.UserType;
+import com.datastax.driver.core.querybuilder.Insert;
+
+/**
+ * Unit tests for {@link SimpleReactiveCassandraRepository}.
+ *
+ * @author Mark Paluch
+ */
+@RunWith(MockitoJUnitRunner.class)
+@SuppressWarnings("unchecked")
+public class SimpleReactiveCassandraRepositoryUnitTests {
+
+	CassandraMappingContext mappingContext = new CassandraMappingContext();
+	MappingCassandraConverter converter = new MappingCassandraConverter(mappingContext);
+
+	SimpleReactiveCassandraRepository<Object, ? extends Serializable> repository;
+
+	@Mock ReactiveCassandraOperations cassandraOperations;
+	@Mock UserTypeResolver userTypeResolver;
+	@Mock UserType userType;
+	@Mock EntityWriteResult writeResult;
+
+	@Captor ArgumentCaptor<Insert> insertCaptor;
+
+	@Before
+	public void before() {
+		mappingContext.setUserTypeResolver(userTypeResolver);
+		when(cassandraOperations.getConverter()).thenReturn(converter);
+	}
+
+	@Test // DATACASS-576
+	public void shouldInsertNewVersionedEntity() {
+
+		when(cassandraOperations.insert(any(), any(InsertOptions.class))).thenReturn(Mono.just(writeResult));
+
+		CassandraPersistentEntity<?> entity = converter.getMappingContext()
+				.getRequiredPersistentEntity(VersionedPerson.class);
+
+		repository = new SimpleReactiveCassandraRepository<Object, String>(
+				new MappingCassandraEntityInformation(entity, converter), cassandraOperations);
+
+		VersionedPerson versionedPerson = new VersionedPerson();
+
+		repository.save(versionedPerson);
+
+		verify(cassandraOperations).insert(versionedPerson, InsertOptions.builder().withInsertNulls().build());
+	}
+
+	@Test // DATACASS-576
+	public void shouldUpdateExistingVersionedEntity() {
+
+		CassandraPersistentEntity<?> entity = converter.getMappingContext()
+				.getRequiredPersistentEntity(VersionedPerson.class);
+
+		repository = new SimpleReactiveCassandraRepository<Object, String>(
+				new MappingCassandraEntityInformation(entity, converter), cassandraOperations);
+
+		VersionedPerson versionedPerson = new VersionedPerson();
+		versionedPerson.setVersion(2);
+
+		repository.save(versionedPerson);
+
+		verify(cassandraOperations).update(versionedPerson);
+	}
+
+	@Data
+	static class VersionedPerson {
+
+		@Id String id;
+		@Version long version;
+	}
+
+}

src/main/asciidoc/new-features.adoc

@@ -10,6 +10,7 @@ This chapter summarizes changes and new features for each release.
 * Kotlin Coroutine extensions for `ReactiveFluentCassandraOperations`.
 * Lightweight transaction support via `DeleteOptions` using the Template API.
 * Filter conditions for lightweight transaction update and delete (`UPDATE … IF <condition>`, `DELETE … IF <condition>`).
+* Optimistic Locking support.
 
 [[new-features.2-1-0]]
 == What's new in Spring Data for Apache Cassandra 2.1

src/main/asciidoc/reference/cassandra.adoc

@@ -1167,6 +1167,45 @@ You can use the following overloaded methods to remove an object from the databa
 * `T` *delete* `(T entity, QueryOptions queryOptions)`: Deletes the given object applying `QueryOptions`.
 * `boolean` *deleteById* `(Object id, Class<?> entityClass)`: Deletes the object using the given Id.
 
+[[cassandra.template.optimistic-locking]]
+=== Optimistic Locking
+
+The `@Version` annotation provides syntax similar to that of JPA in the context of Cassandra and makes sure updates are only applied to rows with a matching version.
+Optimistic Locking leverages Cassandra's lightweight transactions to conditionally insert, update and delete rows.
+Therefore, `INSERT` statements are executed with the `IF NOT EXISTS` condition.
+For updates and deletes, the actual value of the version property is added to the `UPDATE` condition in such a way that the modification does not have any effect if another operation altered the row in the meantime.
+In that case, an `OptimisticLockingFailureException` is thrown.
+The following example shows these features:
+
+====
+[source,java]
+----
+@Table
+class Person {
+
+  @Id String id;
+  String firstname;
+  String lastname;
+  @Version Long version;
+}
+
+Person daenerys = template.insert(new Person("Daenerys"));                            <1>
+
+Person tmp = template.findOne(query(where("id").is(daenerys.getId())), Person.class); <2>
+
+daenerys.setLastname("Targaryen");
+template.save(daenerys);                                                              <3>
+
+template.save(tmp); // throws OptimisticLockingFailureException                       <4>
+----
+<1> Intially insert document. `version` is set to `0`.
+<2> Load the just inserted document. `version` is still `0`.
+<3> Update the document with `version = 0`. Set the `lastname` and bump `version` to `1`.
+<4> Try to update the previously loaded document that still has `version = 0`. The operation fails with an `OptimisticLockingFailureException`, as the current `version` is `1`.
+====
+
+NOTE: Optimistic Locking is only supported with single-entity operations and not for batch operations.
+
 [[cassandra.template.query]]
 == Querying Rows
 

src/main/asciidoc/reference/mapping.adoc

@@ -427,6 +427,7 @@ Types are derived from the declaration by default.
 * `@Tuple`: Applied at the type level to use a type as a mapped tuple.
 * `@Element`: Applied at the field level to specify element or field ordinals within a mapped tuple.
 Types are derived from the property declaration by default.
+* `@Version`: Applied at field level is used for optimistic locking and checked for modification on save operations. The initial value is `zero` which is bumped automatically on every update.
 
 The mapping metadata infrastructure is defined in the separate, spring-data-commons project that is both
 technology- and data store-agnostic.