DATACASS-288 - Add support for Cassandra Batch operations.

We now support Cassandra batching via CassandraBatchOperations. Batch operations allow to insert/update/delete entities in an atomic way.

  Group walter = new Group(new GroupKey("users", "0x1", "walter"));
  Group mike = new Group(new GroupKey("users", "0x1", "mike"));
  Group tuco = new Group(new GroupKey("users", "0x1", "tuco"));

  CassandraBatchOperations batchOperations = cassandraOperations.batchOps();
  batchOperations.insert(walter).update(mike).delete(tuco).execute();

Original pull request: #78.

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

@@ -0,0 +1,106 @@
+/*
+ * Copyright 2016 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.cassandra.core;
+
+/**
+ * Batch operations for insert/update/delete actions on a table. {@link CassandraBatchOperations} use logged Cassandra
+ * {@code BATCH}es for single entities and collections of entities. A {@link CassandraBatchOperations} instance cannot
+ * be modified/used once it was executed.
+ * <p>
+ * Batches are atomic by default. In the context of a Cassandra batch operation, atomic means that if any of the batch
+ * succeeds, all of it will. Statement order does not matter within a batch. {@link CassandraBatchOperations} applies
+ * all rows using the same {@link #withTimestamp(long) timestamp} if supplied, otherwise Cassandra will generate a
+ * timestamp.
+ * <p>
+ * Multi partition batches should only be used to achieve atomicity for a few writes on different tables. Apart from
+ * this they should be avoided because they’re too expensive. Single partition batches can be used to get atomicity and
+ * isolation, they're not much more expensive than normal writes.
+ *
+ * @author Mark Paluch
+ * @since 1.5
+ */
+public interface CassandraBatchOperations {
+
+	/**
+	 * Execute the batch. The batch can be executed only once.
+	 * 
+	 * @throws IllegalStateException if the batch is executed after it was executed once
+	 */
+	void execute();
+
+	/**
+	 * Apply a given {@code timestamp} to the whole batch.
+	 *
+	 * @param timestamp the timestamp to apply.
+	 * @return {@code this} {@link CassandraBatchOperations}
+	 * @throws IllegalStateException if the batch was already executed
+	 */
+	CassandraBatchOperations withTimestamp(long timestamp);
+
+	/**
+	 * Add a single insert to the batch.
+	 *
+	 * @param entity the entity to insert, must not be {@literal null}.
+	 * @return {@code this} {@link CassandraBatchOperations}
+	 * @throws IllegalStateException if the batch was already executed
+	 */
+	CassandraBatchOperations insert(Object entity);
+
+	/**
+	 * Add a collection of inserts to the batch.
+	 * 
+	 * @param entities the entities to insert, must not be {@literal null}.
+	 * @return {@code this} {@link CassandraBatchOperations}
+	 * @throws IllegalStateException if the batch was already executed
+	 */
+	CassandraBatchOperations insert(Iterable<? extends Object> entities);
+
+	/**
+	 * Add a single update to the batch.
+	 * 
+	 * @param entity the entity to update, must not be {@literal null}.
+	 * @return {@code this} {@link CassandraBatchOperations}
+	 * @throws IllegalStateException if the batch was already executed
+	 */
+	CassandraBatchOperations update(Object entity);
+
+	/**
+	 * Add a collection of updates to the batch.
+	 * 
+	 * @param entities the entities to insert, must not be {@literal null}.
+	 * @return {@code this} {@link CassandraBatchOperations}
+	 * @throws IllegalStateException if the batch was already executed
+	 */
+	CassandraBatchOperations update(Iterable<? extends Object> entities);
+
+	/**
+	 * Add a single delete to the batch.
+	 * 
+	 * @param entity the entity to delete, must not be {@literal null}.
+	 * @return {@code this} {@link CassandraBatchOperations}
+	 * @throws IllegalStateException if the batch was already executed
+	 */
+	CassandraBatchOperations delete(Object entity);
+
+	/**
+	 * Add a collection of deletes to the batch.
+	 * 
+	 * @param entities the entities to delete, must not be {@literal null}.
+	 * @return {@code this} {@link CassandraBatchOperations}
+	 * @throws IllegalStateException if the batch was already executed
+	 */
+	CassandraBatchOperations delete(Iterable<? extends Object> entities);
+}

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

@@ -0,0 +1,177 @@
+/*
+ * Copyright 2016 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.cassandra.core;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.springframework.util.Assert;
+
+import com.datastax.driver.core.querybuilder.Batch;
+import com.datastax.driver.core.querybuilder.QueryBuilder;
+
+/**
+ * Default implementation for {@link CassandraBatchOperations}.
+ *
+ * @author Mark Paluch
+ * @since 1.5
+ */
+class CassandraBatchTemplate implements CassandraBatchOperations {
+
+	private final CassandraTemplate cassandraTemplate;
+	private final Batch batch;
+	private AtomicBoolean executed = new AtomicBoolean();
+
+	public CassandraBatchTemplate(CassandraTemplate cassandraTemplate) {
+
+		Assert.notNull(cassandraTemplate, "CassandraTemplate must not be null");
+
+		this.cassandraTemplate = cassandraTemplate;
+		this.batch = QueryBuilder.batch();
+	}
+
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.data.cassandra.core.CassandraBatchOperations#execute()
+	 */
+	@Override
+	public void execute() {
+
+		if (executed.compareAndSet(false, true)) {
+			cassandraTemplate.execute(batch);
+			return;
+		}
+
+		ensureNotExecuted();
+	}
+
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.data.cassandra.core.CassandraBatchOperations#withTimestamp(long)
+	 */
+	@Override
+	public CassandraBatchOperations withTimestamp(long timestamp) {
+
+		ensureNotExecuted();
+
+		batch.using(QueryBuilder.timestamp(timestamp));
+		return this;
+	}
+
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.data.cassandra.core.CassandraBatchOperations#insert(java.lang.Object)
+	 */
+	@Override
+	public CassandraBatchOperations insert(Object entity) {
+
+		ensureNotExecuted();
+		Assert.notNull(entity, "Entity must not be null");
+
+		batch.add(cassandraTemplate.createInsertQuery(entity, null));
+		return this;
+	}
+
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.data.cassandra.core.CassandraBatchOperations#insert(java.lang.Iterable)
+	 */
+	@Override
+	public CassandraBatchOperations insert(Iterable<? extends Object> entities) {
+
+		ensureNotExecuted();
+		Assert.notNull(entities, "Entities must not be null");
+
+		for (Object entity : entities) {
+
+			Assert.notNull(entity, "Entity must not be null");
+			batch.add(cassandraTemplate.createInsertQuery(entity, null));
+		}
+
+		return this;
+	}
+
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.data.cassandra.core.CassandraBatchOperations#update(java.lang.Object)
+	 */
+	@Override
+	public CassandraBatchOperations update(Object entity) {
+
+		ensureNotExecuted();
+		Assert.notNull(entity, "Entity must not be null");
+
+		batch.add(cassandraTemplate.createUpdateQuery(entity, null));
+
+		return this;
+	}
+
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.data.cassandra.core.CassandraBatchOperations#update(java.lang.Iterable)
+	 */
+	@Override
+	public CassandraBatchOperations update(Iterable<? extends Object> entities) {
+
+		ensureNotExecuted();
+		Assert.notNull(entities, "Entities must not be null");
+
+		for (Object entity : entities) {
+
+			Assert.notNull(entity, "Entity must not be null");
+			batch.add(cassandraTemplate.createUpdateQuery(entity, null));
+		}
+
+		return this;
+	}
+
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.data.cassandra.core.CassandraBatchOperations#delete(java.lang.Object)
+	 */
+	@Override
+	public CassandraBatchOperations delete(Object entity) {
+
+		ensureNotExecuted();
+		Assert.notNull(entity, "Entity must not be null");
+
+		batch.add(cassandraTemplate.createDeleteQuery(entity, null));
+
+		return this;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.cassandra.core.CassandraBatchOperations#delete(java.lang.Iterable)
+	 */
+	@Override
+	public CassandraBatchOperations delete(Iterable<? extends Object> entities) {
+
+		ensureNotExecuted();
+		Assert.notNull(entities, "Entities must not be null");
+
+		for (Object entity : entities) {
+
+			Assert.notNull(entity, "Entity must not be null");
+			batch.add(cassandraTemplate.createDeleteQuery(entity, null));
+		}
+
+		return this;
+	}
+
+	private void ensureNotExecuted() {
+		Assert.state(!executed.get(), "This Cassandra Batch was already executed");
+	}
+}

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

@@ -59,7 +59,7 @@ public interface CassandraOperations extends CqlOperations {
 	 * Returns a {@link java.util.Iterator} that wraps the Cassandra {@link com.datastax.driver.core.ResultSet}.
 	 *
 	 * @param <T> element return type.
-	 * @param query query to execute.  Must not be empty or {@literal null}.
+	 * @param query query to execute. Must not be empty or {@literal null}.
 	 * @param entityClass Class type of the elements in the {@link Iterator} stream. Must not be {@literal null}.
 	 * @return an {@link Iterator} (stream) over the elements in the query result set.
 	 * @since 1.5
@@ -465,6 +465,7 @@ public interface CassandraOperations extends CqlOperations {
 
 	/**
 	 * Deletes all entities of a given class.
+	 * 
 	 * @param entityClass The entity type must not be {@literal null}.
 	 */
 	<T> void deleteAll(Class<T> entityClass);
@@ -533,6 +534,14 @@ public interface CassandraOperations extends CqlOperations {
 	 */
 	<T> Cancellable deleteAsynchronously(List<T> entities, DeletionListener<T> listener, QueryOptions options);
 
+	/**
+	 * Returns a new {@link CassandraBatchOperations}. Each {@link CassandraBatchOperations} instance can be executed only
+	 * once so you might want to obtain new {@link CassandraBatchOperations} instances for each batch.
+	 *
+	 * @return a new {@link CassandraBatchOperations} associated with the given entity class.
+	 */
+	CassandraBatchOperations batchOps();
+
 	/**
 	 * Returns the underlying {@link CassandraConverter}.
 	 *

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

@@ -704,16 +704,23 @@ public class CassandraTemplate extends CqlTemplate implements CassandraOperation
 	protected <T> T doInsert(T entity, WriteOptions options) {
 
 		Assert.notNull(entity, "Entity must not be null");
-		Insert insert = createInsertQuery(getTableName(entity.getClass()).toCql(), entity, options, cassandraConverter);
+		Insert insert = createInsertQuery(entity, options);
 		execute(insert);
 		return entity;
 	}
 
+	<T> Insert createInsertQuery(T entity, WriteOptions options) {
+
+		Assert.notNull(entity, "Entity must not be null");
+
+		return createInsertQuery(getTableName(entity.getClass()).toCql(), entity, options, cassandraConverter);
+	}
+
 	protected <T> Cancellable doInsertAsync(final T entity, final WriteListener<T> listener, WriteOptions options) {
 
 		Assert.notNull(entity, "Entity must not be null");
 
-		Insert insert = createInsertQuery(getTableName(entity.getClass()).toCql(), entity, options, cassandraConverter);
+		Insert insert = createInsertQuery(entity, options);
 
 		AsynchronousQueryListener queryListener = (listener == null ? null : new AsynchronousQueryListener() {
 
@@ -840,17 +847,24 @@ public class CassandraTemplate extends CqlTemplate implements CassandraOperation
 		return executeAsynchronously(batch, queryListener);
 	}
 
+	<T> Delete createDeleteQuery(T entity, QueryOptions options) {
+
+		Assert.notNull(entity, "Entity must not be null");
+
+		return createDeleteQuery(getTableName(entity.getClass()).toCql(), entity, options, cassandraConverter);
+	}
+
 	protected <T> void doDelete(T entity, QueryOptions options) {
 		Assert.notNull(entity, "Entity must not be null");
 
-		execute(createDeleteQuery(getTableName(entity.getClass()).toCql(), entity, options, cassandraConverter));
+		execute(createDeleteQuery(entity, options));
 	}
 
 	protected <T> Cancellable doDeleteAsync(final T entity, final DeletionListener<T> listener, QueryOptions options) {
 
 		Assert.notNull(entity, "Entity must not be null");
 
-		Delete delete = createDeleteQuery(getTableName(entity.getClass()).toCql(), entity, options, cassandraConverter);
+		Delete delete = createDeleteQuery(entity, options);
 
 		AsynchronousQueryListener queryListener = (listener == null ? null : new AsynchronousQueryListener() {
 			@Override
@@ -867,10 +881,17 @@ public class CassandraTemplate extends CqlTemplate implements CassandraOperation
 		return executeAsynchronously(delete, queryListener);
 	}
 
+	<T> Update createUpdateQuery(T entity, WriteOptions options) {
+
+		Assert.notNull(entity, "Entity must not be null");
+
+		return createUpdateQuery(getTableName(entity.getClass()).toCql(), entity, options, cassandraConverter);
+	}
+
 	protected <T> T doUpdate(T entity, WriteOptions options) {
 		Assert.notNull(entity, "Entity must not be null");
 
-		execute(createUpdateQuery(getTableName(entity.getClass()).toCql(), entity, options, cassandraConverter));
+		execute(createUpdateQuery(entity, options));
 
 		return entity;
 	}
@@ -893,8 +914,16 @@ public class CassandraTemplate extends CqlTemplate implements CassandraOperation
 			}
 		});
 
-		return executeAsynchronously(createUpdateQuery(getTableName(entity.getClass()).toCql(), entity, options,
-			cassandraConverter), queryListener);
+		return executeAsynchronously(createUpdateQuery(entity, options), queryListener);
+	}
+
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.data.cassandra.core.CassandraOperations#batchOps(java.lang.Class)
+	 */
+	@Override
+	public CassandraBatchOperations batchOps() {
+		return new CassandraBatchTemplate(this);
 	}
 
 	/**

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

@@ -0,0 +1,238 @@
+/*
+ * Copyright 2016 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.cassandra.core;
+
+import static org.hamcrest.Matchers.*;
+import static org.junit.Assert.*;
+
+import java.util.Arrays;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.data.cassandra.domain.FlatGroup;
+import org.springframework.data.cassandra.domain.Group;
+import org.springframework.data.cassandra.domain.GroupKey;
+import org.springframework.data.cassandra.test.integration.support.AbstractSpringDataEmbeddedCassandraIntegrationTest;
+import org.springframework.data.cassandra.test.integration.support.IntegrationTestConfig;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+import com.datastax.driver.core.ResultSet;
+import com.datastax.driver.core.Row;
+
+/**
+ * Integration tests for {@link CassandraBatchTemplate}.
+ * 
+ * @author Mark Paluch
+ */
+@RunWith(SpringJUnit4ClassRunner.class)
+@ContextConfiguration
+public class CassandraBatchTemplateIntegrationTests extends AbstractSpringDataEmbeddedCassandraIntegrationTest {
+
+	@Configuration
+	public static class Config extends IntegrationTestConfig {
+
+		@Override
+		public String[] getEntityBasePackages() {
+			return new String[] { Group.class.getPackage().getName() };
+		}
+	}
+
+	@Autowired CassandraTemplate cassandraTemplate;
+
+	@Before
+	public void setUp() throws Exception {
+		cassandraTemplate.deleteAll(Group.class);
+	}
+
+	/**
+	 * @see DATACASS-288
+	 */
+	@Test
+	public void shouldInsertEntities() {
+
+		Group walter = new Group(new GroupKey("users", "0x1", "walter"));
+		Group mike = new Group(new GroupKey("users", "0x1", "mike"));
+
+		CassandraBatchOperations batchOperations = new CassandraBatchTemplate(cassandraTemplate);
+		batchOperations.insert(walter).insert(mike).execute();
+
+		Group loaded = cassandraTemplate.selectOneById(Group.class, walter.getId());
+		assertThat(loaded.getId().getUsername(), is(equalTo(walter.getId().getUsername())));
+	}
+
+	/**
+	 * @see DATACASS-288
+	 */
+	@Test
+	public void shouldInsertCollectionOfEntities() {
+
+		Group walter = new Group(new GroupKey("users", "0x1", "walter"));
+		Group mike = new Group(new GroupKey("users", "0x1", "mike"));
+
+		CassandraBatchOperations batchOperations = new CassandraBatchTemplate(cassandraTemplate);
+		batchOperations.insert(Arrays.asList(walter, mike)).execute();
+
+		Group loaded = cassandraTemplate.selectOneById(Group.class, walter.getId());
+		assertThat(loaded.getId().getUsername(), is(equalTo(walter.getId().getUsername())));
+	}
+
+	/**
+	 * @see DATACASS-288
+	 */
+	@Test
+	public void shouldUpdateEntities() {
+
+		Group walter = cassandraTemplate.insert(new Group(new GroupKey("users", "0x1", "walter")));
+		Group mike = cassandraTemplate.insert(new Group(new GroupKey("users", "0x1", "mike")));
+
+		walter.setEmail("walter@white.com");
+		mike.setEmail("mike@sauls.com");
+
+		CassandraBatchOperations batchOperations = new CassandraBatchTemplate(cassandraTemplate);
+		batchOperations.update(walter).update(mike).execute();
+
+		Group loaded = cassandraTemplate.selectOneById(Group.class, walter.getId());
+		assertThat(loaded.getEmail(), is(equalTo(walter.getEmail())));
+	}
+
+	/**
+	 * @see DATACASS-288
+	 */
+	@Test
+	public void shouldUpdateCollectionOfEntities() {
+
+		Group walter = cassandraTemplate.insert(new Group(new GroupKey("users", "0x1", "walter")));
+		Group mike = cassandraTemplate.insert(new Group(new GroupKey("users", "0x1", "mike")));
+
+		walter.setEmail("walter@white.com");
+		mike.setEmail("mike@sauls.com");
+
+		CassandraBatchOperations batchOperations = new CassandraBatchTemplate(cassandraTemplate);
+		batchOperations.update(Arrays.asList(walter, mike)).execute();
+
+		Group loaded = cassandraTemplate.selectOneById(Group.class, walter.getId());
+		assertThat(loaded.getEmail(), is(equalTo(walter.getEmail())));
+	}
+
+	/**
+	 * @see DATACASS-288
+	 */
+	@Test
+	public void shouldUpdatesCollectionOfEntities() {
+
+		FlatGroup walter = cassandraTemplate.insert(new FlatGroup("users", "0x1", "walter"));
+		FlatGroup mike = cassandraTemplate.insert(new FlatGroup("users", "0x1", "mike"));
+
+		walter.setEmail("walter@white.com");
+		mike.setEmail("mike@sauls.com");
+
+		CassandraBatchOperations batchOperations = new CassandraBatchTemplate(cassandraTemplate);
+		batchOperations.update(Arrays.asList(walter, mike)).execute();
+
+		FlatGroup loaded = cassandraTemplate.selectOneById(FlatGroup.class, walter);
+		assertThat(loaded.getEmail(), is(equalTo(walter.getEmail())));
+	}
+
+	/**
+	 * @see DATACASS-288
+	 */
+	@Test
+	public void shouldDeleteEntities() {
+
+		Group walter = cassandraTemplate.insert(new Group(new GroupKey("users", "0x1", "walter")));
+		Group mike = cassandraTemplate.insert(new Group(new GroupKey("users", "0x1", "mike")));
+
+		CassandraBatchOperations batchOperations = new CassandraBatchTemplate(cassandraTemplate);
+
+		batchOperations.delete(walter).delete(mike).execute();
+
+		Group loaded = cassandraTemplate.selectOneById(Group.class, walter.getId());
+		assertThat(loaded, is(nullValue()));
+	}
+
+	/**
+	 * @see DATACASS-288
+	 */
+	@Test
+	public void shouldDeleteCollectionOfEntities() {
+
+		Group walter = cassandraTemplate.insert(new Group(new GroupKey("users", "0x1", "walter")));
+		Group mike = cassandraTemplate.insert(new Group(new GroupKey("users", "0x1", "mike")));
+
+		CassandraBatchOperations batchOperations = new CassandraBatchTemplate(cassandraTemplate);
+
+		batchOperations.delete(Arrays.asList(walter, mike)).execute();
+
+		Group loaded = cassandraTemplate.selectOneById(Group.class, walter.getId());
+		assertThat(loaded, is(nullValue()));
+	}
+
+	/**
+	 * @see DATACASS-288
+	 */
+	@Test
+	public void shouldApplyTimestampToAllEntities() {
+
+		Group walter = new Group(new GroupKey("users", "0x1", "walter"));
+		Group mike = new Group(new GroupKey("users", "0x1", "mike"));
+
+		walter.setEmail("walter@white.com");
+		mike.setEmail("mike@sauls.com");
+
+		long timestamp = (System.currentTimeMillis() + TimeUnit.DAYS.toMillis(1)) * 1000;
+
+		CassandraBatchOperations batchOperations = new CassandraBatchTemplate(cassandraTemplate);
+		batchOperations.insert(walter).insert(mike).withTimestamp(timestamp).execute();
+
+		ResultSet resultSet = cassandraTemplate.query("SELECT writetime(email) FROM group;");
+
+		assertThat(resultSet.getAvailableWithoutFetching(), is(2));
+		for (Row row : resultSet) {
+			assertThat(row.getLong(0), is(timestamp));
+		}
+	}
+
+	/**
+	 * @see DATACASS-288
+	 */
+	@Test(expected = IllegalStateException.class)
+	public void shouldNotExecuteTwice() {
+
+		CassandraBatchOperations batchOperations = new CassandraBatchTemplate(cassandraTemplate);
+		batchOperations.insert(new Group(new GroupKey("users", "0x1", "walter"))).execute();
+
+		batchOperations.execute();
+		fail("Missing IllegalStateException");
+	}
+
+	/**
+	 * @see DATACASS-288
+	 */
+	@Test(expected = IllegalStateException.class)
+	public void shouldNotAllowModificationAfterExecution() {
+
+		CassandraBatchOperations batchOperations = new CassandraBatchTemplate(cassandraTemplate);
+		batchOperations.insert(new Group(new GroupKey("users", "0x1", "walter"))).execute();
+
+		batchOperations.update(new Group());
+		fail("Missing IllegalStateException");
+	}
+}

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

@@ -29,12 +29,15 @@ import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.mockito.Mock;
+import org.mockito.Mockito;
 import org.mockito.runners.MockitoJUnitRunner;
 import org.springframework.data.cassandra.convert.CassandraConverter;
+import org.springframework.data.cassandra.test.integration.simpletons.Book;
 
 import com.datastax.driver.core.ResultSet;
 import com.datastax.driver.core.Row;
 import com.datastax.driver.core.Session;
+import com.datastax.driver.core.querybuilder.Batch;
 import com.datastax.driver.core.querybuilder.Select;
 
 /**
@@ -167,4 +170,15 @@ public class CassandraTemplateUnitTests {
 		verify(mockSession, times(1)).execute(eq("SELECT * FROM Test"));
 		verifyZeroInteractions(mockCassandraConverter);
 	}
+
+	/**
+	 * @see DATACASS-288
+	 */
+	@Test
+	public void batchOperationsShouldCallSession() {
+
+		template.batchOps().insert(new Book()).execute();
+
+		verify(mockSession).execute(Mockito.any(Batch.class));
+	}
 }

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

@@ -0,0 +1,44 @@
+/*
+ * Copyright 2016 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.cassandra.domain;
+
+import lombok.Data;
+
+import org.springframework.cassandra.core.PrimaryKeyType;
+import org.springframework.data.cassandra.mapping.PrimaryKeyColumn;
+import org.springframework.data.cassandra.mapping.Table;
+
+/**
+ * @author Mark Paluch
+ * @see http://www.datastax.com/dev/blog/basic-rules-of-cassandra-data-modeling
+ */
+@Table
+@Data
+public class FlatGroup {
+
+	@PrimaryKeyColumn(ordinal = 1, type = PrimaryKeyType.PARTITIONED) private String groupname;
+	@PrimaryKeyColumn(name = "hash_prefix", ordinal = 2, type = PrimaryKeyType.PARTITIONED) private String hashPrefix;
+	@PrimaryKeyColumn(ordinal = 3, type = PrimaryKeyType.CLUSTERED) private String username;
+
+	private String email;
+	private int age;
+
+	public FlatGroup(String groupname, String hashPrefix, String username) {
+		this.groupname = groupname;
+		this.hashPrefix = hashPrefix;
+		this.username = username;
+	}
+}

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

@@ -0,0 +1,41 @@
+/*
+ * Copyright 2016 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.cassandra.domain;
+
+import org.springframework.data.cassandra.mapping.PrimaryKey;
+import org.springframework.data.cassandra.mapping.Table;
+
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/**
+ * @author Mark Paluch
+ * @see http://www.datastax.com/dev/blog/basic-rules-of-cassandra-data-modeling
+ */
+@Table
+@Data
+@NoArgsConstructor
+public class Group {
+
+	@PrimaryKey private GroupKey id;
+
+	private String email;
+	private int age;
+
+	public Group(GroupKey id) {
+		this.id = id;
+	}
+}

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

@@ -0,0 +1,41 @@
+/*
+ * Copyright 2016 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.cassandra.domain;
+
+import java.io.Serializable;
+
+import org.springframework.cassandra.core.PrimaryKeyType;
+import org.springframework.data.cassandra.mapping.PrimaryKeyClass;
+import org.springframework.data.cassandra.mapping.PrimaryKeyColumn;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/**
+ * @author Mark Paluch
+ * @see http://www.datastax.com/dev/blog/basic-rules-of-cassandra-data-modeling
+ */
+@PrimaryKeyClass
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
+public class GroupKey implements Serializable {
+
+	@PrimaryKeyColumn(ordinal = 1, type = PrimaryKeyType.PARTITIONED) private String groupname;
+	@PrimaryKeyColumn(name = "hash_prefix", ordinal = 2, type = PrimaryKeyType.PARTITIONED) private String hashPrefix;
+	@PrimaryKeyColumn(ordinal = 3, type = PrimaryKeyType.CLUSTERED) private String username;
+}

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/test/integration/simpletons/Book.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2014 the original author or authors
+ * Copyright 2013-2016 the original author or authors
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -20,12 +20,18 @@ import java.util.Date;
 import org.springframework.data.cassandra.mapping.PrimaryKey;
 import org.springframework.data.cassandra.mapping.Table;
 
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
 /**
  * Test POJO
  *
  * @author David Webb
+ * @author Mark Paluch
  */
 @Table("book")
+@Data
+@NoArgsConstructor
 public class Book {
 
 	@PrimaryKey private String isbn;
@@ -37,115 +43,7 @@ public class Book {
 	private boolean isInStock;
 	private BookCondition condition;
 
-	/**
-	 * @return Returns the isbn.
-	 */
-	public String getIsbn() {
-		return isbn;
-	}
-
-	/**
-	 * @return Returns the saleDate.
-	 */
-	public Date getSaleDate() {
-		return saleDate;
-	}
-
-	/**
-	 * @param saleDate The saleDate to set.
-	 */
-	public void setSaleDate(Date saleDate) {
-		this.saleDate = saleDate;
-	}
-
-	/**
-	 * @return Returns the isInStock.
-	 */
-	public boolean isInStock() {
-		return isInStock;
-	}
-
-	/**
-	 * @param isInStock The isInStock to set.
-	 */
-	public void setInStock(boolean isInStock) {
-		this.isInStock = isInStock;
-	}
-
-	/**
-	 * @param isbn The isbn to set.
-	 */
-	public void setIsbn(String isbn) {
+	public Book(String isbn) {
 		this.isbn = isbn;
 	}
-
-	/**
-	 * @return Returns the title.
-	 */
-	public String getTitle() {
-		return title;
-	}
-
-	/**
-	 * @param title The title to set.
-	 */
-	public void setTitle(String title) {
-		this.title = title;
-	}
-
-	/**
-	 * @return Returns the author.
-	 */
-	public String getAuthor() {
-		return author;
-	}
-
-	/**
-	 * @param author The author to set.
-	 */
-	public void setAuthor(String author) {
-		this.author = author;
-	}
-
-	/**
-	 * @return Returns the pages.
-	 */
-	public int getPages() {
-		return pages;
-	}
-
-	/**
-	 * @param pages The pages to set.
-	 */
-	public void setPages(int pages) {
-		this.pages = pages;
-	}
-
-	/**
-	 * @param condition The condition to set.
-	 */
-	public void setCondition(BookCondition condition) {
-		this.condition = condition;
-	}
-
-	/**
-	 * @return Returns the condition.
-	 */
-	public BookCondition getCondition() {
-		return condition;
-	}
-
-	/* (non-Javadoc)
-	 * @see java.lang.Object#toString()
-	 */
-	@Override
-	public String toString() {
-		StringBuilder sb = new StringBuilder();
-		sb.append("isbn -> " + isbn).append("\n");
-		sb.append("tile -> " + title).append("\n");
-		sb.append("author -> " + author).append("\n");
-		sb.append("pages -> " + pages).append("\n");
-		sb.append("condition -> " + condition).append("\n");
-		return sb.toString();
-	}
 }