diff --git a/pom.xml b/pom.xml
index 335689a37..6a52dc867 100644
--- a/pom.xml
+++ b/pom.xml
@@ -227,6 +227,7 @@
 				<groupId>org.apache.maven.plugins</groupId>
 				<artifactId>maven-surefire-plugin</artifactId>
 				<configuration>
+			    	<argLine>-Xmx2048m -XX:MaxPermSize=512m</argLine>
 					<parallel>methods</parallel>
 					<threadCount>10</threadCount>
 					<useFile>false</useFile>
@@ -246,6 +247,7 @@
 				<groupId>org.apache.maven.plugins</groupId>
 				<artifactId>maven-failsafe-plugin</artifactId>
 				<configuration>
+				    <argLine>-Xmx2048m -XX:MaxPermSize=512m</argLine>
 					<useFile>false</useFile>
 					<includes>
 						<include>**/test/integration/**/*.java</include>
diff --git a/spring-cassandra/src/test/java/org/springframework/cassandra/test/integration/AbstractEmbeddedCassandraIntegrationTest.java b/spring-cassandra/src/test/java/org/springframework/cassandra/test/integration/AbstractEmbeddedCassandraIntegrationTest.java
index c636965fc..22dd2544d 100644
--- a/spring-cassandra/src/test/java/org/springframework/cassandra/test/integration/AbstractEmbeddedCassandraIntegrationTest.java
+++ b/spring-cassandra/src/test/java/org/springframework/cassandra/test/integration/AbstractEmbeddedCassandraIntegrationTest.java
@@ -29,7 +29,7 @@ public abstract class AbstractEmbeddedCassandraIntegrationTest {
 	/**
 	 * Whether to clear the cluster before the next test.
 	 */
-	protected boolean clear = true;
+	protected boolean clear = false;
 	/**
 	 * Whether to connect to Cassandra.
 	 */
diff --git a/spring-cassandra/src/test/java/org/springframework/cassandra/test/integration/core/template/CassandraOperationsTest.java b/spring-cassandra/src/test/java/org/springframework/cassandra/test/integration/core/template/CassandraOperationsTest.java
index bd413e819..7b763431d 100644
--- a/spring-cassandra/src/test/java/org/springframework/cassandra/test/integration/core/template/CassandraOperationsTest.java
+++ b/spring-cassandra/src/test/java/org/springframework/cassandra/test/integration/core/template/CassandraOperationsTest.java
@@ -15,14 +15,16 @@
  */
 package org.springframework.cassandra.test.integration.core.template;
 
+import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
 
 import java.util.Collection;
 import java.util.LinkedList;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
-
-import junit.framework.Assert;
+import java.util.UUID;
 
 import org.cassandraunit.CassandraCQLUnit;
 import org.cassandraunit.dataset.cql.ClassPathCQLDataSet;
@@ -35,17 +37,26 @@ import org.springframework.cassandra.core.CassandraOperations;
 import org.springframework.cassandra.core.CassandraTemplate;
 import org.springframework.cassandra.core.HostMapper;
 import org.springframework.cassandra.core.PreparedStatementBinder;
+import org.springframework.cassandra.core.PreparedStatementCallback;
+import org.springframework.cassandra.core.PreparedStatementCreator;
 import org.springframework.cassandra.core.ResultSetExtractor;
+import org.springframework.cassandra.core.ResultSetFutureExtractor;
 import org.springframework.cassandra.core.RingMember;
+import org.springframework.cassandra.core.RowCallbackHandler;
 import org.springframework.cassandra.core.RowIterator;
+import org.springframework.cassandra.core.RowMapper;
+import org.springframework.cassandra.core.SessionCallback;
 import org.springframework.cassandra.test.integration.AbstractEmbeddedCassandraIntegrationTest;
 import org.springframework.dao.DataAccessException;
+import org.springframework.util.CollectionUtils;
 
 import com.datastax.driver.core.BoundStatement;
 import com.datastax.driver.core.Host;
 import com.datastax.driver.core.PreparedStatement;
 import com.datastax.driver.core.ResultSet;
+import com.datastax.driver.core.ResultSetFuture;
 import com.datastax.driver.core.Row;
+import com.datastax.driver.core.Session;
 import com.datastax.driver.core.exceptions.DriverException;
 
 /**
@@ -63,6 +74,8 @@ public class CassandraOperationsTest extends AbstractEmbeddedCassandraIntegratio
 	/*
 	 * Objects used for test data
 	 */
+	final String ISBN_NINES = "999999999";
+	final String TITLE_NINES = "Book of Nines";
 	final Object[] o1 = new Object[] { "1234", "Moby Dick", "Herman Manville", new Integer(456) };
 	final Object[] o2 = new Object[] { "2345", "War and Peace", "Russian Dude", new Integer(456) };
 	final Object[] o3 = new Object[] { "3456", "Jane Ayre", "Charlotte", new Integer(456) };
@@ -118,7 +131,7 @@ public class CassandraOperationsTest extends AbstractEmbeddedCassandraIntegratio
 		});
 
 		assertNotNull(ring);
-		Assert.assertTrue(ring.size() > 0);
+		assertTrue(ring.size() > 0);
 
 		for (MyHost h : ring) {
 			log.info("hostMapperTest Host -> " + h.someName);
@@ -133,32 +146,20 @@ public class CassandraOperationsTest extends AbstractEmbeddedCassandraIntegratio
 
 		List<List<?>> values = new LinkedList<List<?>>();
 
-		List<Object> l1 = new LinkedList<Object>();
-		l1.add("1234");
-		l1.add("Moby Dick");
-		l1.add("Herman Manville");
-		l1.add(new Integer(456));
-
-		values.add(l1);
-
-		List<Object> l2 = new LinkedList<Object>();
-		l2.add("2345");
-		l2.add("War and Peace");
-		l2.add("Russian Dude");
-		l2.add(new Integer(456));
-
-		values.add(l2);
-
-		// values.add(new Object[] { "3456", "Jane Ayre", "Charlotte", new Integer(456) });
+		values.add(new LinkedList<Object>(CollectionUtils.arrayToList(o1)));
+		values.add(new LinkedList<Object>(CollectionUtils.arrayToList(o2)));
+		values.add(new LinkedList<Object>(CollectionUtils.arrayToList(o3)));
 
 		cassandraTemplate.ingest(cql, values);
 
 		// Assert that the rows were inserted into Cassandra
-		Book b1 = getBook("1234");
-		Book b2 = getBook("2345");
+		Book b1 = getBook((String) o1[0]);
+		Book b2 = getBook((String) o2[0]);
+		Book b3 = getBook((String) o3[0]);
 
-		Assert.assertEquals(b1.getIsbn(), l1.get(0));
-		Assert.assertEquals(b2.getIsbn(), l2.get(0));
+		assertBook(b1, objectToBook(o1));
+		assertBook(b2, objectToBook(o2));
+		assertBook(b3, objectToBook(o3));
 	}
 
 	@Test
@@ -178,9 +179,9 @@ public class CassandraOperationsTest extends AbstractEmbeddedCassandraIntegratio
 		Book b2 = getBook("2345");
 		Book b3 = getBook("3456");
 
-		Assert.assertEquals(b1.getIsbn(), values[0][0]);
-		Assert.assertEquals(b2.getTitle(), values[1][1]);
-		Assert.assertEquals(b3.getAuthor(), values[2][2]);
+		assertBook(b1, objectToBook(o1));
+		assertBook(b2, objectToBook(o2));
+		assertBook(b3, objectToBook(o3));
 	}
 
 	/**
@@ -234,11 +235,878 @@ public class CassandraOperationsTest extends AbstractEmbeddedCassandraIntegratio
 		Book b2 = getBook("2345");
 		Book b3 = getBook("3456");
 
-		Assert.assertEquals(b1.getIsbn(), o1[0]);
-		Assert.assertEquals(b2.getTitle(), o2[1]);
-		Assert.assertEquals(b3.getAuthor(), o3[2]);
+		assertBook(b1, objectToBook(o1));
+		assertBook(b2, objectToBook(o2));
+		assertBook(b3, objectToBook(o3));
+
 	}
 
+	@Test
+	public void executeTestSessionCallback() {
+
+		final String isbn = UUID.randomUUID().toString();
+		final String title = "Spring Data Cassandra Cookbook";
+		final String author = "David Webb";
+		final Integer pages = 1;
+
+		cassandraTemplate.execute(new SessionCallback<Object>() {
+
+			@Override
+			public Object doInSession(Session s) throws DataAccessException {
+
+				String cql = "insert into book (isbn, title, author, pages) values (?, ?, ?, ?)";
+
+				PreparedStatement ps = s.prepare(cql);
+				BoundStatement bs = ps.bind(isbn, title, author, pages);
+
+				s.execute(bs);
+
+				return null;
+
+			}
+		});
+
+		Book b = getBook(isbn);
+
+		assertBook(b, isbn, title, author, pages);
+
+	}
+
+	@Test
+	public void executeTestCqlString() {
+
+		final String isbn = UUID.randomUUID().toString();
+		final String title = "Spring Data Cassandra Cookbook";
+		final String author = "David Webb";
+		final Integer pages = 1;
+
+		cassandraTemplate.execute("insert into book (isbn, title, author, pages) values ('" + isbn + "', '" + title
+				+ "', '" + author + "', " + pages + ")");
+
+		Book b = getBook(isbn);
+
+		assertBook(b, isbn, title, author, pages);
+
+	}
+
+	@Test
+	public void executeAsynchronouslyTestCqlString() {
+
+		final String isbn = UUID.randomUUID().toString();
+		final String title = "Spring Data Cassandra Cookbook";
+		final String author = "David Webb";
+		final Integer pages = 1;
+
+		cassandraTemplate.executeAsynchronously("insert into book (isbn, title, author, pages) values ('" + isbn + "', '"
+				+ title + "', '" + author + "', " + pages + ")");
+
+		try {
+			Thread.sleep(2000);
+		} catch (InterruptedException e) {
+			e.printStackTrace();
+		}
+
+		Book b = getBook(isbn);
+
+		assertBook(b, isbn, title, author, pages);
+
+	}
+
+	@Test
+	public void queryTestCqlStringResultSetExtractor() {
+
+		final String isbn = "999999999";
+
+		Book b1 = cassandraTemplate.query("select * from book where isbn='" + isbn + "'", new ResultSetExtractor<Book>() {
+
+			@Override
+			public Book extractData(ResultSet rs) throws DriverException, DataAccessException {
+				Row r = rs.one();
+				assertNotNull(r);
+
+				Book b = rowToBook(r);
+
+				return b;
+			}
+		});
+
+		Book b2 = getBook(isbn);
+
+		assertBook(b1, b2);
+
+	}
+
+	@Test
+	public void queryAsynchronouslyTestCqlStringResultSetExtractor() {
+
+		final String isbn = "999999999";
+
+		Book b1 = cassandraTemplate.queryAsynchronously("select * from book where isbn='" + isbn + "'",
+				new ResultSetFutureExtractor<Book>() {
+
+					@Override
+					public Book extractData(ResultSetFuture rs) throws DriverException, DataAccessException {
+
+						ResultSet frs = rs.getUninterruptibly();
+						Row r = frs.one();
+						assertNotNull(r);
+
+						Book b = rowToBook(r);
+
+						return b;
+					}
+				});
+
+		Book b2 = getBook(isbn);
+
+		assertBook(b1, b2);
+
+	}
+
+	@Test
+	public void queryTestCqlStringRowCallbackHandler() {
+
+		final String isbn = "999999999";
+
+		final Book b1 = getBook(isbn);
+
+		cassandraTemplate.query("select * from book where isbn='" + isbn + "'", new RowCallbackHandler() {
+
+			@Override
+			public void processRow(Row row) throws DriverException {
+
+				assertNotNull(row);
+
+				Book b = rowToBook(row);
+
+				assertBook(b1, b);
+
+			}
+		});
+
+	}
+
+	@Test
+	public void processTestResultSetRowCallbackHandler() {
+
+		final String isbn = "999999999";
+
+		final Book b1 = getBook(isbn);
+
+		ResultSet rs = cassandraTemplate.queryAsynchronously("select * from book where isbn='" + isbn + "'",
+				new ResultSetFutureExtractor<ResultSet>() {
+
+					@Override
+					public ResultSet extractData(ResultSetFuture rs) throws DriverException, DataAccessException {
+
+						ResultSet frs = rs.getUninterruptibly();
+						return frs;
+					}
+				});
+
+		assertNotNull(rs);
+
+		cassandraTemplate.process(rs, new RowCallbackHandler() {
+
+			@Override
+			public void processRow(Row row) throws DriverException {
+
+				assertNotNull(row);
+
+				Book b = rowToBook(row);
+
+				assertBook(b1, b);
+
+			}
+
+		});
+
+	}
+
+	@Test
+	public void queryTestCqlStringRowMapper() {
+
+		// Insert our 3 test books.
+		ingestionTestObjectArray();
+
+		List<Book> books = cassandraTemplate.query("select * from book where isbn in ('1234','2345','3456')",
+				new RowMapper<Book>() {
+
+					@Override
+					public Book mapRow(Row row, int rowNum) throws DriverException {
+						Book b = rowToBook(row);
+						return b;
+					}
+				});
+
+		log.debug("Size of Book List -> " + books.size());
+		assertEquals(books.size(), 3);
+		assertBook(books.get(0), getBook(books.get(0).getIsbn()));
+		assertBook(books.get(1), getBook(books.get(1).getIsbn()));
+		assertBook(books.get(2), getBook(books.get(2).getIsbn()));
+	}
+
+	@Test
+	public void processTestResultSetRowMapper() {
+
+		// Insert our 3 test books.
+		ingestionTestObjectArray();
+
+		ResultSet rs = cassandraTemplate.queryAsynchronously("select * from book where isbn in ('1234','2345','3456')",
+				new ResultSetFutureExtractor<ResultSet>() {
+
+					@Override
+					public ResultSet extractData(ResultSetFuture rs) throws DriverException, DataAccessException {
+
+						ResultSet frs = rs.getUninterruptibly();
+						return frs;
+					}
+				});
+
+		assertNotNull(rs);
+
+		List<Book> books = cassandraTemplate.process(rs, new RowMapper<Book>() {
+
+			@Override
+			public Book mapRow(Row row, int rowNum) throws DriverException {
+				Book b = rowToBook(row);
+				return b;
+			}
+		});
+
+		log.debug("Size of Book List -> " + books.size());
+		assertEquals(books.size(), 3);
+		assertBook(books.get(0), getBook(books.get(0).getIsbn()));
+		assertBook(books.get(1), getBook(books.get(1).getIsbn()));
+		assertBook(books.get(2), getBook(books.get(2).getIsbn()));
+
+	}
+
+	@Test
+	public void queryForObjectTestCqlStringRowMapper() {
+
+		Book book = cassandraTemplate.queryForObject("select * from book where isbn in ('" + ISBN_NINES + "')",
+				new RowMapper<Book>() {
+					@Override
+					public Book mapRow(Row row, int rowNum) throws DriverException {
+						Book b = rowToBook(row);
+						return b;
+					}
+				});
+
+		assertNotNull(book);
+		assertBook(book, getBook(ISBN_NINES));
+	}
+
+	/**
+	 * Test that CQL for QueryForObject must only return 1 row or an IllegalArgumentException is thrown.
+	 */
+	@Test(expected = IllegalArgumentException.class)
+	public void queryForObjectTestCqlStringRowMapperNotOneRowReturned() {
+
+		// Insert our 3 test books.
+		ingestionTestObjectArray();
+
+		Book book = cassandraTemplate.queryForObject("select * from book where isbn in ('1234','2345','3456')",
+				new RowMapper<Book>() {
+					@Override
+					public Book mapRow(Row row, int rowNum) throws DriverException {
+						Book b = rowToBook(row);
+						return b;
+					}
+				});
+	}
+
+	@Test
+	public void processOneTestResultSetRowMapper() {
+
+		// Insert our 3 test books.
+		ingestionTestObjectArray();
+
+		ResultSet rs = cassandraTemplate.queryAsynchronously("select * from book where isbn in ('" + ISBN_NINES + "')",
+				new ResultSetFutureExtractor<ResultSet>() {
+
+					@Override
+					public ResultSet extractData(ResultSetFuture rs) throws DriverException, DataAccessException {
+
+						ResultSet frs = rs.getUninterruptibly();
+						return frs;
+					}
+				});
+
+		assertNotNull(rs);
+
+		Book book = cassandraTemplate.processOne(rs, new RowMapper<Book>() {
+			@Override
+			public Book mapRow(Row row, int rowNum) throws DriverException {
+				Book b = rowToBook(row);
+				return b;
+			}
+		});
+
+		assertNotNull(book);
+		assertBook(book, getBook(ISBN_NINES));
+	}
+
+	@Test
+	public void quertForObjectTestCqlStringRequiredType() {
+
+		String title = cassandraTemplate.queryForObject("select title from book where isbn in ('" + ISBN_NINES + "')",
+				String.class);
+
+		assertEquals(title, TITLE_NINES);
+
+	}
+
+	@Test(expected = ClassCastException.class)
+	public void queryForObjectTestCqlStringRequiredTypeInvalid() {
+
+		Float title = cassandraTemplate.queryForObject("select title from book where isbn in ('" + ISBN_NINES + "')",
+				Float.class);
+
+	}
+
+	@Test
+	public void processOneTestResultSetType() {
+
+		ResultSet rs = cassandraTemplate.queryAsynchronously("select title from book where isbn in ('" + ISBN_NINES + "')",
+				new ResultSetFutureExtractor<ResultSet>() {
+
+					@Override
+					public ResultSet extractData(ResultSetFuture rs) throws DriverException, DataAccessException {
+
+						ResultSet frs = rs.getUninterruptibly();
+						return frs;
+					}
+				});
+
+		assertNotNull(rs);
+
+		String title = cassandraTemplate.processOne(rs, String.class);
+
+		assertNotNull(title);
+		assertEquals(title, TITLE_NINES);
+	}
+
+	@Test
+	public void queryForMapTestCqlString() {
+
+		Map<String, Object> rsMap = cassandraTemplate
+				.queryForMap("select * from book where isbn in ('" + ISBN_NINES + "')");
+
+		log.debug(rsMap.toString());
+
+		Book b1 = objectToBook(rsMap.get("isbn"), rsMap.get("title"), rsMap.get("author"), rsMap.get("pages"));
+
+		Book b2 = getBook(ISBN_NINES);
+
+		assertBook(b1, b2);
+
+	}
+
+	@Test
+	public void processMapTestResultSet() {
+
+		ResultSet rs = cassandraTemplate.queryAsynchronously("select * from book where isbn in ('" + ISBN_NINES + "')",
+				new ResultSetFutureExtractor<ResultSet>() {
+
+					@Override
+					public ResultSet extractData(ResultSetFuture rs) throws DriverException, DataAccessException {
+
+						ResultSet frs = rs.getUninterruptibly();
+						return frs;
+					}
+				});
+
+		assertNotNull(rs);
+
+		Map<String, Object> rsMap = cassandraTemplate.processMap(rs);
+
+		log.debug("Size of Book List -> " + rsMap.size());
+
+		Book b1 = objectToBook(rsMap.get("isbn"), rsMap.get("title"), rsMap.get("author"), rsMap.get("pages"));
+
+		Book b2 = getBook(ISBN_NINES);
+
+		assertBook(b1, b2);
+
+	}
+
+	@Test
+	public void queryForListTestCqlStringType() {
+
+		// Insert our 3 test books.
+		ingestionTestObjectArray();
+
+		List<String> titles = cassandraTemplate.queryForList("select title from book where isbn in ('1234','2345','3456')",
+				String.class);
+
+		log.debug(titles.toString());
+
+		assertNotNull(titles);
+		assertEquals(titles.size(), 3);
+
+	}
+
+	@Test
+	public void processListTestResultSetType() {
+
+		// Insert our 3 test books.
+		ingestionTestObjectArray();
+
+		ResultSet rs = cassandraTemplate.queryAsynchronously("select * from book where isbn in ('1234','2345','3456')",
+				new ResultSetFutureExtractor<ResultSet>() {
+
+					@Override
+					public ResultSet extractData(ResultSetFuture rs) throws DriverException, DataAccessException {
+
+						ResultSet frs = rs.getUninterruptibly();
+						return frs;
+					}
+				});
+
+		assertNotNull(rs);
+
+		List<String> titles = cassandraTemplate.processList(rs, String.class);
+
+		log.debug(titles.toString());
+
+		assertNotNull(titles);
+		assertEquals(titles.size(), 3);
+	}
+
+	@Test
+	public void queryForListOfMapCqlString() {
+
+		// Insert our 3 test books.
+		ingestionTestObjectArray();
+
+		List<Map<String, Object>> results = cassandraTemplate
+				.queryForListOfMap("select * from book where isbn in ('1234','2345','3456')");
+
+		log.debug(results.toString());
+
+		assertEquals(results.size(), 3);
+
+	}
+
+	@Test
+	public void processListOfMapTestResultSet() {
+
+		// Insert our 3 test books.
+		ingestionTestObjectArray();
+
+		ResultSet rs = cassandraTemplate.queryAsynchronously("select * from book where isbn in ('1234','2345','3456')",
+				new ResultSetFutureExtractor<ResultSet>() {
+
+					@Override
+					public ResultSet extractData(ResultSetFuture rs) throws DriverException, DataAccessException {
+
+						ResultSet frs = rs.getUninterruptibly();
+						return frs;
+					}
+				});
+
+		assertNotNull(rs);
+
+		List<Map<String, Object>> results = cassandraTemplate.processListOfMap(rs);
+
+		log.debug(results.toString());
+
+		assertEquals(results.size(), 3);
+
+	}
+
+	@Test
+	public void executeTestCqlStringPreparedStatementCallback() {
+
+		String cql = "insert into book (isbn, title, author, pages) values (?, ?, ?, ?)";
+
+		BoundStatement statement = cassandraTemplate.execute(cql, new PreparedStatementCallback<BoundStatement>() {
+
+			@Override
+			public BoundStatement doInPreparedStatement(PreparedStatement ps) throws DriverException, DataAccessException {
+				BoundStatement bs = ps.bind();
+				return bs;
+			}
+		});
+
+		assertNotNull(statement);
+
+	}
+
+	@Test
+	public void executeTestPreparedStatementCreatorPreparedStatementCallback() {
+
+		final String cql = "insert into book (isbn, title, author, pages) values (?, ?, ?, ?)";
+
+		BoundStatement statement = cassandraTemplate.execute(new PreparedStatementCreator() {
+
+			@Override
+			public PreparedStatement createPreparedStatement(Session session) throws DriverException {
+				return session.prepare(cql);
+			}
+		}, new PreparedStatementCallback<BoundStatement>() {
+
+			@Override
+			public BoundStatement doInPreparedStatement(PreparedStatement ps) throws DriverException, DataAccessException {
+				BoundStatement bs = ps.bind();
+				return bs;
+			}
+		});
+
+		assertNotNull(statement);
+
+	}
+
+	@Test
+	public void queryTestCqlStringPreparedStatementBinderResultSetExtractor() {
+
+		final String cql = "select * from book where isbn = ?";
+		final String isbn = "999999999";
+
+		Book b1 = cassandraTemplate.query(cql, new PreparedStatementBinder() {
+
+			@Override
+			public BoundStatement bindValues(PreparedStatement ps) throws DriverException {
+				return ps.bind(isbn);
+			}
+		}, new ResultSetExtractor<Book>() {
+
+			@Override
+			public Book extractData(ResultSet rs) throws DriverException, DataAccessException {
+				Row r = rs.one();
+				assertNotNull(r);
+
+				Book b = rowToBook(r);
+
+				return b;
+			}
+		});
+
+		Book b2 = getBook(isbn);
+
+		assertBook(b1, b2);
+	}
+
+	@Test
+	public void queryTestCqlStringPreparedStatementBinderRowCallbackHandler() {
+
+		final String cql = "select * from book where isbn = ?";
+		final String isbn = "999999999";
+
+		cassandraTemplate.query(cql, new PreparedStatementBinder() {
+
+			@Override
+			public BoundStatement bindValues(PreparedStatement ps) throws DriverException {
+				return ps.bind(isbn);
+			}
+		}, new RowCallbackHandler() {
+
+			@Override
+			public void processRow(Row row) throws DriverException {
+
+				Book b = rowToBook(row);
+
+				Book b2 = getBook(isbn);
+
+				assertBook(b, b2);
+
+			}
+		});
+
+	}
+
+	@Test
+	public void queryTestCqlStringPreparedStatementBinderRowMapper() {
+
+		final String cql = "select * from book where isbn = ?";
+		final String isbn = "999999999";
+
+		List<Book> books = cassandraTemplate.query(cql, new PreparedStatementBinder() {
+
+			@Override
+			public BoundStatement bindValues(PreparedStatement ps) throws DriverException {
+				return ps.bind(isbn);
+			}
+		}, new RowMapper<Book>() {
+
+			@Override
+			public Book mapRow(Row row, int rowNum) throws DriverException {
+				return rowToBook(row);
+			}
+		});
+
+		Book b2 = getBook(isbn);
+
+		assertEquals(books.size(), 1);
+		assertBook(books.get(0), b2);
+	}
+
+	@Test
+	public void queryTestPreparedStatementCreatorResultSetExtractor() {
+
+		ingestionTestObjectArray();
+
+		final String cql = "select * from book";
+
+		List<Book> books = cassandraTemplate.query(new PreparedStatementCreator() {
+
+			@Override
+			public PreparedStatement createPreparedStatement(Session session) throws DriverException {
+				return session.prepare(cql);
+			}
+		}, new ResultSetExtractor<List<Book>>() {
+
+			@Override
+			public List<Book> extractData(ResultSet rs) throws DriverException, DataAccessException {
+
+				List<Book> books = new LinkedList<Book>();
+
+				for (Row row : rs.all()) {
+					books.add(rowToBook(row));
+				}
+
+				return books;
+			}
+		});
+
+		log.debug("Size of all Books -> " + books.size());
+
+		assertTrue(books.size() > 0);
+	}
+
+	@Test
+	public void queryTestPreparedStatementCreatorRowCallbackHandler() {
+
+		ingestionTestObjectArray();
+
+		final String cql = "select * from book";
+
+		cassandraTemplate.query(new PreparedStatementCreator() {
+
+			@Override
+			public PreparedStatement createPreparedStatement(Session session) throws DriverException {
+				return session.prepare(cql);
+			}
+		}, new RowCallbackHandler() {
+
+			@Override
+			public void processRow(Row row) throws DriverException {
+
+				Book b = rowToBook(row);
+
+				log.debug("Title -> " + b.getTitle());
+
+			}
+		});
+
+	}
+
+	@Test
+	public void queryTestPreparedStatementCreatorRowMapper() {
+
+		ingestionTestObjectArray();
+
+		final String cql = "select * from book";
+
+		List<Book> books = cassandraTemplate.query(new PreparedStatementCreator() {
+
+			@Override
+			public PreparedStatement createPreparedStatement(Session session) throws DriverException {
+				return session.prepare(cql);
+			}
+		}, new RowMapper<Book>() {
+
+			@Override
+			public Book mapRow(Row row, int rowNum) throws DriverException {
+				return rowToBook(row);
+			}
+		});
+
+		log.debug("Size of all Books -> " + books.size());
+
+		assertTrue(books.size() > 0);
+	}
+
+	@Test
+	public void queryTestPreparedStatementCreatorPreparedStatementBinderResultSetExtractor() {
+
+		final String cql = "select * from book where isbn = ?";
+		final String isbn = "999999999";
+
+		List<Book> books = cassandraTemplate.query(new PreparedStatementCreator() {
+
+			@Override
+			public PreparedStatement createPreparedStatement(Session session) throws DriverException {
+				return session.prepare(cql);
+			}
+		}, new PreparedStatementBinder() {
+
+			@Override
+			public BoundStatement bindValues(PreparedStatement ps) throws DriverException {
+				return ps.bind(isbn);
+			}
+		}, new ResultSetExtractor<List<Book>>() {
+
+			@Override
+			public List<Book> extractData(ResultSet rs) throws DriverException, DataAccessException {
+				List<Book> books = new LinkedList<Book>();
+
+				for (Row row : rs.all()) {
+					books.add(rowToBook(row));
+				}
+
+				return books;
+			}
+		});
+
+		Book b2 = getBook(isbn);
+
+		log.debug("Book list Size -> " + books.size());
+
+		assertEquals(books.size(), 1);
+		assertBook(books.get(0), b2);
+	}
+
+	@Test
+	public void queryTestPreparedStatementCreatorPreparedStatementBinderRowCallbackHandler() {
+
+		final String cql = "select * from book where isbn = ?";
+		final String isbn = "999999999";
+
+		cassandraTemplate.query(new PreparedStatementCreator() {
+
+			@Override
+			public PreparedStatement createPreparedStatement(Session session) throws DriverException {
+				return session.prepare(cql);
+			}
+		}, new PreparedStatementBinder() {
+
+			@Override
+			public BoundStatement bindValues(PreparedStatement ps) throws DriverException {
+				return ps.bind(isbn);
+			}
+		}, new RowCallbackHandler() {
+
+			@Override
+			public void processRow(Row row) throws DriverException {
+				Book b = rowToBook(row);
+				Book b2 = getBook(isbn);
+				assertBook(b, b2);
+			}
+		});
+
+	}
+
+	@Test
+	public void queryTestPreparedStatementCreatorPreparedStatementBinderRowMapper() {
+
+		final String cql = "select * from book where isbn = ?";
+		final String isbn = "999999999";
+
+		List<Book> books = cassandraTemplate.query(new PreparedStatementCreator() {
+
+			@Override
+			public PreparedStatement createPreparedStatement(Session session) throws DriverException {
+				return session.prepare(cql);
+			}
+		}, new PreparedStatementBinder() {
+
+			@Override
+			public BoundStatement bindValues(PreparedStatement ps) throws DriverException {
+				return ps.bind(isbn);
+			}
+		}, new RowMapper<Book>() {
+
+			@Override
+			public Book mapRow(Row row, int rowNum) throws DriverException {
+				return rowToBook(row);
+			}
+		});
+
+		Book b2 = getBook(isbn);
+
+		assertEquals(books.size(), 1);
+		assertBook(books.get(0), b2);
+	}
+
+	/**
+	 * Assert that a Book matches the arguments expected
+	 * 
+	 * @param b
+	 * @param orderedElements
+	 */
+	private void assertBook(Book b, Object... orderedElements) {
+
+		assertEquals(b.getIsbn(), orderedElements[0]);
+		assertEquals(b.getTitle(), orderedElements[1]);
+		assertEquals(b.getAuthor(), orderedElements[2]);
+		assertEquals(b.getPages(), orderedElements[3]);
+
+	}
+
+	private Book rowToBook(Row row) {
+		Book b = new Book();
+		b.setIsbn(row.getString("isbn"));
+		b.setTitle(row.getString("title"));
+		b.setAuthor(row.getString("author"));
+		b.setPages(row.getInt("pages"));
+		return b;
+	}
+
+	/**
+	 * Convert Object[] to a Book
+	 * 
+	 * @param bookElements
+	 * @return
+	 */
+	private Book objectToBook(Object... bookElements) {
+		Book b = new Book();
+		b.setIsbn((String) bookElements[0]);
+		b.setTitle((String) bookElements[1]);
+		b.setAuthor((String) bookElements[2]);
+		b.setPages((Integer) bookElements[3]);
+		return b;
+	}
+
+	/**
+	 * Convert List<Object> to a Book
+	 * 
+	 * @param bookElements
+	 * @return
+	 */
+	private Book listToBook(List<Object> bookElements) {
+		Book b = new Book();
+		b.setIsbn((String) bookElements.get(0));
+		b.setTitle((String) bookElements.get(1));
+		b.setAuthor((String) bookElements.get(2));
+		b.setPages((Integer) bookElements.get(3));
+		return b;
+
+	}
+
+	/**
+	 * Assert that 2 Book objects are the same
+	 * 
+	 * @param b
+	 * @param orderedElements
+	 */
+	private void assertBook(Book b1, Book b2) {
+
+		assertEquals(b1.getIsbn(), b2.getIsbn());
+		assertEquals(b1.getTitle(), b2.getTitle());
+		assertEquals(b1.getAuthor(), b2.getAuthor());
+		assertEquals(b1.getPages(), b2.getPages());
+
+	}
+
+	/**
+	 * Get a Book from Cassandra for assertions.
+	 * 
+	 * @param isbn
+	 * @return
+	 */
 	public Book getBook(final String isbn) {
 
 		Book b = this.cassandraTemplate.query("select * from book where isbn = ?", new PreparedStatementBinder() {