DATACASS-328 - Revise QueryOptions and WriteOptions.

Deprecate our org.springframework.cassandra.core.ConsistencyLevel enum. Having an own consistency level type leads to confusion and it's always behind the driver. We don't want to maintain that type, so we decided to deprecate the own type and use the driver consistency levels.

We allow now the use of the driver retry policies aside of our consistency level enumeration. For most cases, our enumeration is the simpler approach. Some retry policies (IdempotenceAwareRetryPolicy, LoggingRetryPolicies) require further configuration and cannot be applied with just using a static enum value. We now support ReadTimeout, FetchSize, and Tracing via QueryOptions and WriteOptions and provide builders for QueryOptions and WriteOptions.

Original pull request: #81.

spring-cql/src/main/java/org/springframework/cassandra/core/ConsistencyLevel.java

@@ -20,7 +20,9 @@ package org.springframework.cassandra.core;
  * 
  * @author David Webb
  * @author Antoine Toulme
+ * @deprecated Use the driver's {@link com.datastax.driver.core.ConsistencyLevel}.
  */
+@Deprecated
 public enum ConsistencyLevel {
 
 	ANY, ONE, TWO, THREE,

spring-cql/src/main/java/org/springframework/cassandra/core/ConsistencyLevelResolver.java

@@ -22,7 +22,9 @@ import org.springframework.util.Assert;
  * 
  * @author David Webb
  * @author Antoine Toulme
+ * @deprecated Use the driver's {@link com.datastax.driver.core.ConsistencyLevel}.
  */
+@Deprecated
 public final class ConsistencyLevelResolver {
 
 	/**

spring-cql/src/main/java/org/springframework/cassandra/core/CqlTemplate.java

@@ -58,7 +58,6 @@ import org.springframework.util.Assert;
 import com.datastax.driver.core.BoundStatement;
 import com.datastax.driver.core.CodecRegistry;
 import com.datastax.driver.core.ColumnDefinitions;
-import com.datastax.driver.core.ColumnDefinitions.Definition;
 import com.datastax.driver.core.Host;
 import com.datastax.driver.core.PreparedStatement;
 import com.datastax.driver.core.ProtocolVersion;
@@ -69,6 +68,7 @@ import com.datastax.driver.core.Session;
 import com.datastax.driver.core.SimpleStatement;
 import com.datastax.driver.core.Statement;
 import com.datastax.driver.core.TypeCodec;
+import com.datastax.driver.core.ColumnDefinitions.Definition;
 import com.datastax.driver.core.exceptions.DriverException;
 import com.datastax.driver.core.querybuilder.Batch;
 import com.datastax.driver.core.querybuilder.Delete;
@@ -93,6 +93,7 @@ import com.datastax.driver.core.querybuilder.Update;
  * @author Ryan Scheidter
  * @author Antoine Toulme
  * @author John Blum
+ * @author Mark Paluch
  * @see org.springframework.cassandra.core.CqlOperations
  * @see org.springframework.cassandra.support.CassandraAccessor
  */
@@ -140,11 +141,15 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
 
 		if (queryOptions != null) {
 
-			if (queryOptions.getConsistencyLevel() != null) {
+			if (queryOptions.getDriverConsistencyLevel() != null) {
+				preparedStatement.setConsistencyLevel(queryOptions.getDriverConsistencyLevel());
+			} else if (queryOptions.getConsistencyLevel() != null) {
 				preparedStatement.setConsistencyLevel(ConsistencyLevelResolver.resolve(queryOptions.getConsistencyLevel()));
 			}
 
-			if (queryOptions.getRetryPolicy() != null) {
+			if (queryOptions.getDriverRetryPolicy() != null) {
+				preparedStatement.setRetryPolicy(queryOptions.getDriverRetryPolicy());
+			} else if (queryOptions.getRetryPolicy() != null) {
 				preparedStatement.setRetryPolicy(RetryPolicyResolver.resolve(queryOptions.getRetryPolicy()));
 			}
 		}
@@ -163,13 +168,33 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
 
 		if (queryOptions != null) {
 
-			if (queryOptions.getConsistencyLevel() != null) {
+			if (queryOptions.getDriverConsistencyLevel() != null) {
+				statement.setConsistencyLevel(queryOptions.getDriverConsistencyLevel());
+			} else if (queryOptions.getConsistencyLevel() != null) {
 				statement.setConsistencyLevel(ConsistencyLevelResolver.resolve(queryOptions.getConsistencyLevel()));
 			}
 
-			if (queryOptions.getRetryPolicy() != null) {
+			if (queryOptions.getDriverRetryPolicy() != null) {
+				statement.setRetryPolicy(queryOptions.getDriverRetryPolicy());
+			} else if (queryOptions.getRetryPolicy() != null) {
 				statement.setRetryPolicy(RetryPolicyResolver.resolve(queryOptions.getRetryPolicy()));
 			}
+
+			if (queryOptions.getReadTimeout() != null) {
+				statement.setReadTimeoutMillis(queryOptions.getReadTimeout().intValue());
+			}
+
+			if (queryOptions.getFetchSize() != null) {
+				statement.setFetchSize(queryOptions.getFetchSize());
+			}
+
+			if (queryOptions.getTracing() != null) {
+				if (queryOptions.getTracing()) {
+					statement.enableTracing();
+				} else {
+					statement.disableTracing();
+				}
+			}
 		}
 
 		return statement;

spring-cql/src/main/java/org/springframework/cassandra/core/QueryOptions.java

@@ -15,48 +15,409 @@
  */
 package org.springframework.cassandra.core;
 
+import java.util.concurrent.TimeUnit;
+
+import org.springframework.util.Assert;
+
+import com.datastax.driver.core.SocketOptions;
+
 /**
- * Contains Query Options for Cassandra queries. This controls the Consistency Tuning and Retry Policy for a Query.
+ * Cassandra Query Options for queries. {@link QueryOptions} allow tuning of various query options on a per-request
+ * level. Only options that are set are applied to queries.
  * 
  * @author David Webb
+ * @author Mark Paluch
  */
 public class QueryOptions {
 
 	private ConsistencyLevel consistencyLevel;
-	private RetryPolicy retryPolicy;
+	private com.datastax.driver.core.ConsistencyLevel driverConsistencyLevel;
 
+	private RetryPolicy retryPolicy;
+	private com.datastax.driver.core.policies.RetryPolicy driverRetryPolicy;
+
+	private Long readTimeout;
+	private Integer fetchSize;
+	private Boolean tracing;
+
+	/**
+	 * Creates new {@link QueryOptions}.
+	 */
 	public QueryOptions() {}
 
+	/**
+	 * Creates new {@link QueryOptions} for the given {@link ConsistencyLevel} and {@link RetryPolicy}.
+	 * 
+	 * @param consistencyLevel the consistency level, may be {@literal null}.
+	 * @param retryPolicy the retry policy, may be {@literal null}.
+	 */
 	public QueryOptions(ConsistencyLevel consistencyLevel, RetryPolicy retryPolicy) {
 		setConsistencyLevel(consistencyLevel);
 		setRetryPolicy(retryPolicy);
 	}
 
 	/**
-	 * @return Returns the consistencyLevel.
+	 * Creates a new {@link QueryOptionsBuilder}.
+	 * 
+	 * @return a new {@link QueryOptionsBuilder}.
+	 * @since 1.5
 	 */
+	public static QueryOptionsBuilder builder() {
+		return new QueryOptionsBuilder();
+	}
+
+	/**
+	 * Returns the {@link ConsistencyLevel}.
+	 * 
+	 * @return the consistencyLevel.
+	 * @deprecated Use {@link #setConsistencyLevel(com.datastax.driver.core.ConsistencyLevel)}
+	 */
+	@Deprecated
 	public ConsistencyLevel getConsistencyLevel() {
 		return consistencyLevel;
 	}
 
 	/**
-	 * @param consistencyLevel The consistencyLevel to set.
+	 * Sets the driver {@link ConsistencyLevel}. Setting both ({@link ConsistencyLevel} and
+	 * {@link com.datastax.driver.core.ConsistencyLevel driver ConsistencyLevel}) consistency levels is not supported.
+	 * 
+	 * @param consistencyLevel the consistencyLevel to set.
+	 * @throws IllegalStateException if the {@link com.datastax.driver.core.ConsistencyLevel driver ConsistencyLevel} is
+	 *           set
 	 */
 	public void setConsistencyLevel(ConsistencyLevel consistencyLevel) {
+
+		if (this.driverConsistencyLevel != null && consistencyLevel != null) {
+			throw new IllegalStateException(
+					"ConsistencyLevel cannot not be set if the Driver ConsistencyLevel is already set");
+		}
+
 		this.consistencyLevel = consistencyLevel;
 	}
 
 	/**
-	 * @return Returns the retryPolicy.
+	 * Sets the driver {@link com.datastax.driver.core.ConsistencyLevel}. Setting both ({@link ConsistencyLevel} and
+	 * {@link com.datastax.driver.core.ConsistencyLevel driver ConsistencyLevel}) consistency levels is not supported.
+	 *
+	 * @param driverConsistencyLevel the driver {@link com.datastax.driver.core.ConsistencyLevel} to set.
+	 * @since 1.5
+	 * @throws IllegalStateException if the {@link ConsistencyLevel} is set
+	 */
+	public void setConsistencyLevel(com.datastax.driver.core.ConsistencyLevel driverConsistencyLevel) {
+
+		if (this.consistencyLevel != null && driverConsistencyLevel != null) {
+			throw new IllegalStateException(
+					"Driver ConsistencyLevel cannot not be set if the ConsistencyLevel is already set");
+		}
+
+		this.driverConsistencyLevel = driverConsistencyLevel;
+	}
+
+	/**
+	 * @return the the driver {@link com.datastax.driver.core.ConsistencyLevel}
+	 * @since 1.5
+	 */
+	protected com.datastax.driver.core.ConsistencyLevel getDriverConsistencyLevel() {
+		return driverConsistencyLevel;
+	}
+
+	/**
+	 * Returns the {@link RetryPolicy}.
+	 *
+	 * @return the retryPolicy or {@literal null} if the {@link RetryPolicy} is not set.
 	 */
 	public RetryPolicy getRetryPolicy() {
 		return retryPolicy;
 	}
 
 	/**
-	 * @param retryPolicy The retryPolicy to set.
+	 * Sets the {@link RetryPolicy}. Setting both ({@link RetryPolicy} and
+	 * {@link com.datastax.driver.core.policies.RetryPolicy driver RetryPolicy}) retry policies is not supported.
+	 *
+	 * @param retryPolicy the retryPolicy to set.
+	 * @throws IllegalStateException if the {@link com.datastax.driver.core.policies.RetryPolicy driver RetryPolicy} is
+	 *           set
 	 */
 	public void setRetryPolicy(RetryPolicy retryPolicy) {
+
+		if (this.driverRetryPolicy != null && retryPolicy != null) {
+			throw new IllegalStateException("RetryPolicy cannot not be set if the Driver RetryPolicy is already set");
+		}
+
 		this.retryPolicy = retryPolicy;
 	}
+
+	/**
+	 * Sets the {@link com.datastax.driver.core.policies.RetryPolicy}. Setting both ({@link RetryPolicy} and
+	 * {@link com.datastax.driver.core.policies.RetryPolicy driver RetryPolicy}) retry policies is not supported.
+	 *
+	 * @param driverRetryPolicy the driver {@link com.datastax.driver.core.policies.RetryPolicy} to set.
+	 * @since 1.5
+	 * @throws IllegalStateException if the {@link RetryPolicy} is set
+	 */
+	public void setRetryPolicy(com.datastax.driver.core.policies.RetryPolicy driverRetryPolicy) {
+
+		if (this.retryPolicy != null && driverRetryPolicy != null) {
+			throw new IllegalStateException("Driver RetryPolicy cannot not be set if the RetryPolicy is already set");
+		}
+
+		this.driverRetryPolicy = driverRetryPolicy;
+	}
+
+	/**
+	 * @return the driver {@link com.datastax.driver.core.policies.RetryPolicy}
+	 * @since 1.5
+	 */
+	protected com.datastax.driver.core.policies.RetryPolicy getDriverRetryPolicy() {
+		return driverRetryPolicy;
+	}
+
+	/**
+	 * Sets the read timeout in milliseconds. Overrides the default per-host read timeout (
+	 * {@link SocketOptions#getReadTimeoutMillis()}).
+	 *
+	 * @param readTimeout the read timeout in milliseconds. Negative values are not allowed. If it is {@code 0}, the read timeout
+	 *          will be disabled for this statement.
+	 * @since 1.5
+	 * @see SocketOptions#getReadTimeoutMillis()
+	 * @see com.datastax.driver.core.Cluster.Builder#withSocketOptions(SocketOptions)
+	 */
+	public void setReadTimeout(long readTimeout) {
+
+		Assert.isTrue(readTimeout >= 0, "ReadTimeout must be greater or equal to zero");
+		this.readTimeout = readTimeout;
+	}
+
+	/**
+	 * @return the read timeout in milliseconds. May be {@literal null} if not set.
+	 * @since 1.5
+	 */
+	protected Long getReadTimeout() {
+		return readTimeout;
+	}
+
+	/**
+	 * Sets the query fetch size for {@link com.datastax.driver.core.ResultSet} chunks.
+	 * <p>
+	 * The fetch size controls how much resulting rows will be retrieved simultaneously (the goal being to avoid loading
+	 * too much results in memory for queries yielding large results). Please note that while value as low as 1 can be
+	 * used, it is *highly* discouraged to use such a low value in practice as it will yield very poor performance.
+	 *
+	 * @param fetchSize the number of rows to fetch per chunking request. To disable chunking of the result set, use
+	 *          {@code fetchSize == Integer.MAX_VALUE}. Negative values are not allowed.
+	 * @since 1.5
+	 * @see com.datastax.driver.core.QueryOptions#getFetchSize()
+	 * @see com.datastax.driver.core.Cluster.Builder#withQueryOptions(com.datastax.driver.core.QueryOptions)
+	 */
+	public void setFetchSize(int fetchSize) {
+
+		Assert.isTrue(fetchSize >= 0, "FetchSize must be greater or equal to zero");
+		this.fetchSize = fetchSize;
+	}
+
+	/**
+	 * @return the number of rows to fetch per chunking request. May be {@literal null} if not set.
+	 * @since 1.5
+	 */
+	protected Integer getFetchSize() {
+		return fetchSize;
+	}
+
+	/**
+	 * Enables statement tracing.
+	 *
+	 * @param tracing {@literal true} to enable statement tracing to the executed statements.
+	 * @since 1.5
+	 */
+	public void setTracing(boolean tracing) {
+		this.tracing = tracing;
+	}
+
+	/**
+	 * @return whether to enable tracing. May be {@literal null} if not set.
+	 */
+	protected Boolean getTracing() {
+		return tracing;
+	}
+
+	/**
+	 * Builder for {@link QueryOptions}.
+	 *
+	 * @author Mark Paluch
+	 * @since 1.5
+	 */
+	public static class QueryOptionsBuilder {
+
+		private com.datastax.driver.core.ConsistencyLevel driverConsistencyLevel;
+
+		private RetryPolicy retryPolicy;
+		private com.datastax.driver.core.policies.RetryPolicy driverRetryPolicy;
+
+		private Long readTimeout;
+		private Integer fetchSize;
+		private Boolean tracing;
+
+		QueryOptionsBuilder() {}
+
+		/**
+		 * Sets the {@link RetryPolicy} to use. Setting both ({@link RetryPolicy} and
+		 * {@link com.datastax.driver.core.policies.RetryPolicy driver RetryPolicy}) retry policies is not supported.
+		 *
+		 * @param retryPolicy must not be {@literal null}.
+		 * @return {@code this} {@link QueryOptionsBuilder}
+		 * @throws IllegalStateException if the {@link com.datastax.driver.core.policies.RetryPolicy driver RetryPolicy} is
+		 *           set
+		 */
+		public QueryOptionsBuilder retryPolicy(RetryPolicy retryPolicy) {
+
+			Assert.notNull(retryPolicy, "RetryPolicy must not be null");
+			Assert.state(this.driverRetryPolicy == null,
+					"RetryPolicy cannot not be set if the Driver RetryPolicy is already set");
+
+			this.retryPolicy = retryPolicy;
+			return this;
+		}
+
+		/**
+		 * Sets the {@link com.datastax.driver.core.policies.RetryPolicy driver RetryPolicy} to use. Setting both (
+		 * {@link RetryPolicy} and {@link com.datastax.driver.core.policies.RetryPolicy driver RetryPolicy}) retry policies
+		 * is not supported.
+		 *
+		 * @param driverRetryPolicy must not be {@literal null}.
+		 * @return {@code this} {@link QueryOptionsBuilder}
+		 * @throws IllegalStateException if the {@link RetryPolicy} is set
+		 */
+		public QueryOptionsBuilder retryPolicy(com.datastax.driver.core.policies.RetryPolicy driverRetryPolicy) {
+
+			Assert.notNull(driverRetryPolicy, "Driver RetryPolicy must not be null");
+			Assert.state(this.retryPolicy == null, "Driver RetryPolicy cannot not be set if the RetryPolicy is already set");
+
+			this.driverRetryPolicy = driverRetryPolicy;
+			return this;
+		}
+
+		/**
+		 * Sets the {@link com.datastax.driver.core.ConsistencyLevel} to use.
+		 *
+		 * @param driverConsistencyLevel must not be {@literal null}.
+		 * @return {@code this} {@link QueryOptionsBuilder}
+		 */
+		public QueryOptionsBuilder consistencyLevel(com.datastax.driver.core.ConsistencyLevel driverConsistencyLevel) {
+
+			Assert.notNull(driverConsistencyLevel, "Driver ConsistencyLevel must not be null");
+
+			this.driverConsistencyLevel = driverConsistencyLevel;
+			return this;
+		}
+
+		/**
+		 * Sets the read timeout in milliseconds. Overrides the default per-host read timeout.
+		 *
+		 * @param readTimeout the read timeout in milliseconds. Negative values are not allowed. If it is {@code 0}, the read
+		 *          timeout will be disabled for this statement.
+		 * @return {@code this} {@link QueryOptionsBuilder}
+		 * @see SocketOptions#getReadTimeoutMillis()
+		 * @see com.datastax.driver.core.Cluster.Builder#withSocketOptions(SocketOptions)
+		 */
+		public QueryOptionsBuilder readTimeout(long readTimeout) {
+
+			Assert.isTrue(readTimeout >= 0, "ReadTimeout must be greater or equal to zero");
+
+			this.readTimeout = readTimeout;
+
+			return this;
+		}
+
+		/**
+		 * Sets the read timeout. Overrides the default per-host read timeout.
+		 *
+		 * @param readTimeout the read timeout value. Negative values are not allowed. If it is {@code 0}, the read timeout will be
+		 *          disabled for this statement.
+		 * @param timeUnit the {@link TimeUnit} for the supplied timeout; must not be {@literal null}.
+		 * @return {@code this} {@link QueryOptionsBuilder}
+		 * @see SocketOptions#getReadTimeoutMillis()
+		 * @see com.datastax.driver.core.Cluster.Builder#withSocketOptions(SocketOptions)
+		 */
+		public QueryOptionsBuilder readTimeout(long readTimeout, TimeUnit timeUnit) {
+
+			Assert.isTrue(readTimeout >= 0, "ReadTimeout must be greater or equal to zero");
+			Assert.notNull(timeUnit, "TimeUnit must not be null");
+
+			this.readTimeout = timeUnit.toMillis(readTimeout);
+
+			return this;
+		}
+
+		/**
+		 * Sets the query fetch size for {@link com.datastax.driver.core.ResultSet} chunks.
+		 * <p>
+		 * The fetch size controls how much resulting rows will be retrieved simultaneously (the goal being to avoid loading
+		 * too much results in memory for queries yielding large results). Please note that while value as low as 1 can be
+		 * used, it is *highly* discouraged to use such a low value in practice as it will yield very poor performance.
+		 *
+		 * @param fetchSize the number of rows to fetch per chunking request. To disable chunking of the result set, use
+		 *          {@code fetchSize == Integer.MAX_VALUE}. Negative values are not allowed.
+		 * @return {@code this} {@link QueryOptionsBuilder}
+		 * @see com.datastax.driver.core.QueryOptions#getFetchSize()
+		 * @see com.datastax.driver.core.Cluster.Builder#withQueryOptions(com.datastax.driver.core.QueryOptions)
+		 */
+		public QueryOptionsBuilder fetchSize(int fetchSize) {
+
+			Assert.isTrue(fetchSize >= 0, "FetchSize must be greater or equal to zero");
+
+			this.fetchSize = fetchSize;
+			return this;
+		}
+
+		/**
+		 * Enables statement tracing.
+		 *
+		 * @param tracing {@literal true} to enable statement tracing to the executed statements.
+		 * @return {@code this} {@link QueryOptionsBuilder}
+		 */
+		public QueryOptionsBuilder tracing(boolean tracing) {
+
+			this.tracing = tracing;
+			return this;
+		}
+
+		/**
+		 * Enables statement tracing.
+		 * 
+		 * @return {@code this} {@link QueryOptionsBuilder}
+		 */
+		public QueryOptionsBuilder withTracing() {
+			return tracing(true);
+		}
+
+		/**
+		 * Builds a new {@link QueryOptions} with the configured values.
+		 *
+		 * @return a new {@link QueryOptions} with the configured values
+		 */
+		public QueryOptions build() {
+			return applyOptions(new QueryOptions());
+		}
+
+		<T extends QueryOptions> T applyOptions(T queryOptions) {
+
+			queryOptions.setConsistencyLevel(driverConsistencyLevel);
+			queryOptions.setRetryPolicy(retryPolicy);
+			queryOptions.setRetryPolicy(driverRetryPolicy);
+
+			if (readTimeout != null) {
+				queryOptions.setReadTimeout(readTimeout);
+			}
+
+			if (fetchSize != null) {
+				queryOptions.setFetchSize(fetchSize);
+			}
+
+			if (tracing != null) {
+				queryOptions.setTracing(tracing);
+			}
+
+			return queryOptions;
+		}
+	}
 }

spring-cql/src/main/java/org/springframework/cassandra/core/RetryPolicy.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.
@@ -16,12 +16,20 @@
 package org.springframework.cassandra.core;
 
 /**
- * Retry Policies associated with Cassandra.
- * 
+ * Retry Policies associated with Cassandra. This enum type is a shortcut to the predefined
+ * {@link com.datastax.driver.core.policies.RetryPolicy policies} that come with the driver.
+ *
  * @author David Webb
+ * @author Mark Paluch
  */
 public enum RetryPolicy {
 
-	DEFAULT, DOWNGRADING_CONSISTENCY, FALLTHROUGH, LOGGING
+	DEFAULT, DOWNGRADING_CONSISTENCY, FALLTHROUGH,
+
+	/**
+	 * The {@link com.datastax.driver.core.policies.LoggingRetryPolicy} is just a decorator for other policies so it can't
+	 * be applied to {@link QueryOptions}/{@link WriteOptions}
+	 */
+	LOGGING
 
 }

spring-cql/src/main/java/org/springframework/cassandra/core/RetryPolicyResolver.java

@@ -1,12 +1,12 @@
 /*
  * Copyright 2013-2014 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.
@@ -21,7 +21,7 @@ import com.datastax.driver.core.policies.FallthroughRetryPolicy;
 
 /**
  * Determine driver query retry policy
- * 
+ *
  * @author David Webb
  */
 public final class RetryPolicyResolver {

spring-cql/src/main/java/org/springframework/cassandra/core/WriteOptions.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.
@@ -15,36 +15,180 @@
  */
 package org.springframework.cassandra.core;
 
+import java.util.concurrent.TimeUnit;
+
 /**
+ * Cassandra Write Options are an extension to {@link QueryOptions} for write operations. {@link WriteOptions}allow
+ * tuning of various query options on a per-request level. Only options that are set are applied to queries.
+ * 
  * @author David Webb
+ * @author Mark Paluch
+ * @see QueryOptions
  */
 public class WriteOptions extends QueryOptions {
 
 	private Integer ttl;
 
+	/**
+	 * Creates new {@link WriteOptions}.
+	 */
 	public WriteOptions() {}
 
+	/**
+	 * Creates new {@link WriteOptions} for the given {@link ConsistencyLevel} and {@link RetryPolicy}.
+	 * 
+	 * @param consistencyLevel the consistency level, may be {@literal null}.
+	 * @param retryPolicy the retry policy, may be {@literal null}.
+	 */
 	public WriteOptions(ConsistencyLevel consistencyLevel, RetryPolicy retryPolicy) {
 		this(consistencyLevel, retryPolicy, null);
 	}
 
+	/**
+	 * Creates new {@link WriteOptions} for the given {@link ConsistencyLevel}, {@link RetryPolicy} and {@code ttl}.
+	 * 
+	 * @param consistencyLevel the consistency level, may be {@literal null}.
+	 * @param retryPolicy the retry policy, may be {@literal null}.
+	 * @param ttl the ttl, may be {@literal null}.
+	 */
 	public WriteOptions(ConsistencyLevel consistencyLevel, RetryPolicy retryPolicy, Integer ttl) {
 		super(consistencyLevel, retryPolicy);
 		setTtl(ttl);
 	}
 
 	/**
-	 * @return Returns the ttl.
+	 * Creates a new {@link WriteOptionsBuilder}.
+	 *
+	 * @return a new {@link WriteOptionsBuilder}.
+	 * @since 1.5
+	 */
+	public static WriteOptionsBuilder builder() {
+		return new WriteOptionsBuilder();
+	}
+
+	/**
+	 * @return the time to live, if set.
 	 */
 	public Integer getTtl() {
 		return ttl;
 	}
 
 	/**
-	 * @param ttl The ttl to set.
+	 * Sets the time to live for write operations.
+	 * 
+	 * @param ttl the ttl to set.
 	 */
 	public void setTtl(Integer ttl) {
 		this.ttl = ttl;
 	}
 
+	/**
+	 * Builder for {@link QueryOptions}.
+	 *
+	 * @author Mark Paluch
+	 * @since 1.5
+	 */
+	public static class WriteOptionsBuilder extends QueryOptionsBuilder {
+
+		private Integer ttl;
+
+		private WriteOptionsBuilder() {}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.cassandra.core.QueryOptions.QueryOptionsBuilder#retryPolicy(org.springframework.cassandra.core.RetryPolicy)
+		 */
+		@Override
+		public WriteOptionsBuilder retryPolicy(com.datastax.driver.core.policies.RetryPolicy driverRetryPolicy) {
+			return (WriteOptionsBuilder) super.retryPolicy(driverRetryPolicy);
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.cassandra.core.QueryOptions.QueryOptionsBuilder#retryPolicy(org.springframework.cassandra.core.RetryPolicy)
+		 */
+		@Override
+		public WriteOptionsBuilder retryPolicy(RetryPolicy retryPolicy) {
+			return (WriteOptionsBuilder) super.retryPolicy(retryPolicy);
+		}
+
+		/* 
+		 * (non-Javadoc)
+		 * @see org.springframework.cassandra.core.QueryOptions.QueryOptionsBuilder#consistencyLevel(com.datastax.driver.core.ConsistencyLevel)
+		 */
+		@Override
+		public WriteOptionsBuilder consistencyLevel(com.datastax.driver.core.ConsistencyLevel driverConsistencyLevel) {
+			return (WriteOptionsBuilder) super.consistencyLevel(driverConsistencyLevel);
+		}
+
+		/* 
+		 * (non-Javadoc)
+		 * @see org.springframework.cassandra.core.QueryOptions.QueryOptionsBuilder#readTimeout(long)
+		 */
+		@Override
+		public WriteOptionsBuilder readTimeout(long readTimeout) {
+			return (WriteOptionsBuilder) super.readTimeout(readTimeout);
+		}
+
+		/* 
+		 * (non-Javadoc)
+		 * @see org.springframework.cassandra.core.QueryOptions.QueryOptionsBuilder#readTimeout(long, java.util.concurrent.TimeUnit)
+		 */
+		@Override
+		public WriteOptionsBuilder readTimeout(long readTimeout, TimeUnit timeUnit) {
+			return (WriteOptionsBuilder) super.readTimeout(readTimeout, timeUnit);
+		}
+
+		/* 
+		 * (non-Javadoc)
+		 * @see org.springframework.cassandra.core.QueryOptions.QueryOptionsBuilder#fetchSize(int)
+		 */
+		@Override
+		public WriteOptionsBuilder fetchSize(int fetchSize) {
+			return (WriteOptionsBuilder) super.fetchSize(fetchSize);
+		}
+
+		/* 
+		 * (non-Javadoc)
+		 * @see org.springframework.cassandra.core.QueryOptions.QueryOptionsBuilder#tracing(boolean)
+		 */
+		@Override
+		public WriteOptionsBuilder tracing(boolean tracing) {
+			return (WriteOptionsBuilder) super.tracing(tracing);
+		}
+
+		/* 
+		 * (non-Javadoc)
+		 * @see org.springframework.cassandra.core.QueryOptions.QueryOptionsBuilder#withTracing()
+		 */
+		@Override
+		public WriteOptionsBuilder withTracing() {
+			return (WriteOptionsBuilder) super.withTracing();
+		}
+
+		/**
+		 * Sets the time to live for write operations.
+		 *
+		 * @param ttl the time to live.
+		 * @return {@code this} {@link WriteOptionsBuilder}
+		 */
+		public WriteOptionsBuilder ttl(int ttl) {
+			this.ttl = ttl;
+			return this;
+		}
+
+		/**
+		 * Builds a new {@link WriteOptions} with the configured values.
+		 *
+		 * @return a new {@link WriteOptions} with the configured values
+		 */
+		public WriteOptions build() {
+
+			WriteOptions queryOptions = applyOptions(new WriteOptions());
+
+			queryOptions.setTtl(ttl);
+
+			return queryOptions;
+		}
+	}
 }

spring-cql/src/test/java/org/springframework/cassandra/core/CqlTemplateUnitTests.java

@@ -20,6 +20,7 @@ import static org.junit.Assert.*;
 import static org.mockito.Mockito.*;
 
 import java.util.Iterator;
+import java.util.concurrent.TimeUnit;
 
 import org.junit.Before;
 import org.junit.Rule;
@@ -27,6 +28,7 @@ import org.junit.Test;
 import org.junit.rules.ExpectedException;
 import org.junit.runner.RunWith;
 import org.mockito.Mock;
+import org.mockito.Mockito;
 import org.mockito.runners.MockitoJUnitRunner;
 import org.springframework.cassandra.support.CassandraExceptionTranslator;
 import org.springframework.cassandra.support.exception.CassandraReadTimeoutException;
@@ -36,31 +38,44 @@ import org.springframework.dao.IncorrectResultSizeDataAccessException;
 
 import com.datastax.driver.core.ColumnDefinitions;
 import com.datastax.driver.core.ConsistencyLevel;
+import com.datastax.driver.core.PreparedStatement;
 import com.datastax.driver.core.ResultSet;
 import com.datastax.driver.core.Row;
 import com.datastax.driver.core.Session;
+import com.datastax.driver.core.Statement;
 import com.datastax.driver.core.exceptions.DriverException;
 import com.datastax.driver.core.exceptions.ReadTimeoutException;
+import com.datastax.driver.core.policies.FallthroughRetryPolicy;
+import com.datastax.driver.core.querybuilder.Insert;
 import com.datastax.driver.core.querybuilder.Select;
+import com.datastax.driver.core.querybuilder.Update;
+import com.datastax.driver.core.querybuilder.Using;
 
 /**
  * The CqlTemplateUnitTests class is a test suite of test cases testing the contract and functionality of the
  * {@link CqlTemplate} class.
  *
  * @author John Blum
+ * @author Mark Paluch
  */
 // TODO: add many more unit tests until SUT test coverage is 100%!
 @RunWith(MockitoJUnitRunner.class)
 @SuppressWarnings("unchecked")
 public class CqlTemplateUnitTests {
 
-	@Rule
-	public ExpectedException exception = ExpectedException.none();
+	@Rule public ExpectedException exception = ExpectedException.none();
 
 	private CqlTemplate template;
 
-	@Mock
-	private Session mockSession;
+	@Mock private Session mockSession;
+
+	@Mock private PreparedStatement mockPreparedStatement;
+
+	@Mock private Statement mockStatement;
+
+	@Mock private Insert mockInsert;
+
+	@Mock private Update mockUpdate;
 
 	@Before
 	public void setup() {
@@ -71,7 +86,8 @@ public class CqlTemplateUnitTests {
 	@Test
 	public void doExecuteInSessionCallbackIsCalled() {
 		String result = template.doExecute(new SessionCallback<String>() {
-			@Override public String doInSession(Session session) throws DataAccessException {
+			@Override
+			public String doInSession(Session session) throws DataAccessException {
 				session.execute("test");
 				return "test";
 			}
@@ -91,7 +107,8 @@ public class CqlTemplateUnitTests {
 		exception.expectCause(org.hamcrest.Matchers.isA(ReadTimeoutException.class));
 
 		template.doExecute(new SessionCallback<String>() {
-			@Override public String doInSession(Session session) throws DataAccessException {
+			@Override
+			public String doInSession(Session session) throws DataAccessException {
 				throw new ReadTimeoutException(ConsistencyLevel.ALL, 0, 1, true);
 			}
 		});
@@ -107,7 +124,8 @@ public class CqlTemplateUnitTests {
 		exception.expectMessage(containsString("test"));
 
 		template.doExecute(new SessionCallback<String>() {
-			@Override public String doInSession(Session session) throws DataAccessException {
+			@Override
+			public String doInSession(Session session) throws DataAccessException {
 				throw new DriverException("test");
 			}
 		});
@@ -123,7 +141,8 @@ public class CqlTemplateUnitTests {
 		exception.expectMessage(containsString("test"));
 
 		template.doExecute(new SessionCallback<String>() {
-			@Override public String doInSession(Session session) throws DataAccessException {
+			@Override
+			public String doInSession(Session session) throws DataAccessException {
 				throw new Error("test");
 			}
 		});
@@ -406,4 +425,144 @@ public class CqlTemplateUnitTests {
 
 		template.processOne(null, String.class);
 	}
+
+	/**
+	 * @see DATACASS-202
+	 */
+	@Test
+	public void addPreparedStatementOptionsShouldAddDriverQueryOptions() {
+
+		QueryOptions queryOptions = QueryOptions.builder() //
+				.consistencyLevel(ConsistencyLevel.EACH_QUORUM) //
+				.retryPolicy(FallthroughRetryPolicy.INSTANCE) //
+				.build();
+
+		template.addPreparedStatementOptions(mockPreparedStatement, queryOptions);
+
+		verify(mockPreparedStatement).setConsistencyLevel(ConsistencyLevel.EACH_QUORUM);
+		verify(mockPreparedStatement).setRetryPolicy(FallthroughRetryPolicy.INSTANCE);
+	}
+
+	/**
+	 * @see DATACASS-202
+	 */
+	@Test
+	public void addPreparedStatementOptionsShouldAddOurQueryOptions() {
+
+		QueryOptions queryOptions = QueryOptions.builder() //
+				.retryPolicy(RetryPolicy.FALLTHROUGH).build();
+		queryOptions.setConsistencyLevel(org.springframework.cassandra.core.ConsistencyLevel.LOCAL_QUOROM);
+
+		template.addPreparedStatementOptions(mockPreparedStatement, queryOptions);
+
+		verify(mockPreparedStatement).setRetryPolicy(FallthroughRetryPolicy.INSTANCE);
+		verify(mockPreparedStatement).setConsistencyLevel(ConsistencyLevel.LOCAL_QUORUM);
+	}
+
+	/**
+	 * @see DATACASS-202
+	 */
+	@Test
+	public void addStatementQueryOptionsShouldAddDriverQueryOptions() {
+
+		QueryOptions queryOptions = QueryOptions.builder().consistencyLevel(ConsistencyLevel.EACH_QUORUM) //
+				.retryPolicy(FallthroughRetryPolicy.INSTANCE) //
+				.build();
+
+		template.addQueryOptions(mockStatement, queryOptions);
+
+		verify(mockStatement).setConsistencyLevel(ConsistencyLevel.EACH_QUORUM);
+		verify(mockStatement).setRetryPolicy(FallthroughRetryPolicy.INSTANCE);
+	}
+
+	/**
+	 * @see DATACASS-202
+	 */
+	@Test
+	public void addStatementQueryOptionsShouldAddOurQueryOptions() {
+
+		QueryOptions queryOptions = QueryOptions.builder() //
+				.retryPolicy(RetryPolicy.FALLTHROUGH) //
+				.build();
+		queryOptions.setConsistencyLevel(org.springframework.cassandra.core.ConsistencyLevel.LOCAL_QUOROM);
+
+		template.addQueryOptions(mockStatement, queryOptions);
+
+		verify(mockStatement).setRetryPolicy(FallthroughRetryPolicy.INSTANCE);
+		verify(mockStatement).setConsistencyLevel(ConsistencyLevel.LOCAL_QUORUM);
+	}
+
+	/**
+	 * @see DATACASS-202
+	 */
+	@Test
+	public void addStatementQueryOptionsShouldNotAddOptions() {
+
+		QueryOptions queryOptions = QueryOptions.builder().build();
+
+		template.addQueryOptions(mockStatement, queryOptions);
+
+		verifyZeroInteractions(mockStatement);
+	}
+
+	/**
+	 * @see DATACASS-202
+	 */
+	@Test
+	public void addStatementQueryOptionsShouldAddGenericQueryOptions() {
+
+		QueryOptions queryOptions = QueryOptions.builder() //
+				.readTimeout(1, TimeUnit.MINUTES) //
+				.fetchSize(10) //
+				.withTracing() //
+				.build();
+
+		template.addQueryOptions(mockStatement, queryOptions);
+
+		verify(mockStatement).setReadTimeoutMillis(60 * 1000);
+		verify(mockStatement).setFetchSize(10);
+		verify(mockStatement).enableTracing();
+	}
+
+	/**
+	 * @see DATACASS-202
+	 */
+	@Test
+	public void addInsertWriteOptionsShouldAddDriverQueryOptions() {
+
+		WriteOptions writeOptions = WriteOptions.builder() //
+				.consistencyLevel(ConsistencyLevel.EACH_QUORUM) //
+				.retryPolicy(FallthroughRetryPolicy.INSTANCE) //
+				.readTimeout(10) //
+				.ttl(10) //
+				.build();
+
+		template.addWriteOptions(mockInsert, writeOptions);
+
+		verify(mockInsert).setConsistencyLevel(ConsistencyLevel.EACH_QUORUM);
+		verify(mockInsert).setRetryPolicy(FallthroughRetryPolicy.INSTANCE);
+		verify(mockInsert).setReadTimeoutMillis(10);
+		verify(mockInsert).using(Mockito.any(Using.class));
+	}
+
+	/**
+	 * @see DATACASS-202
+	 */
+	@Test
+	public void addUpdateWriteOptionsShouldAddDriverQueryOptions() {
+
+		WriteOptions writeOptions = WriteOptions.builder() //
+				.consistencyLevel(ConsistencyLevel.EACH_QUORUM) //
+				.retryPolicy(FallthroughRetryPolicy.INSTANCE) //
+				.ttl(10) //
+				.tracing(false)
+				.build();
+
+		template.addWriteOptions(mockUpdate, writeOptions);
+
+		verify(mockUpdate).setConsistencyLevel(ConsistencyLevel.EACH_QUORUM);
+		verify(mockUpdate).setRetryPolicy(FallthroughRetryPolicy.INSTANCE);
+		verify(mockUpdate).using(Mockito.any(Using.class));
+		verify(mockUpdate).disableTracing();
+	}
 }

spring-cql/src/test/java/org/springframework/cassandra/core/QueryOptionsUnitTests.java

@@ -0,0 +1,153 @@
+/*
+ * 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.cassandra.core;
+
+import static org.hamcrest.MatcherAssert.*;
+import static org.hamcrest.Matchers.*;
+
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Test;
+
+import com.datastax.driver.core.ConsistencyLevel;
+import com.datastax.driver.core.policies.DefaultRetryPolicy;
+import com.datastax.driver.core.policies.FallthroughRetryPolicy;
+import com.datastax.driver.core.policies.LoggingRetryPolicy;
+
+/**
+ * Unit tests for {@link QueryOptions}.
+ * 
+ * @author Mark Paluch
+ */
+public class QueryOptionsUnitTests {
+
+	/**
+	 * @see DATACASS-202
+	 */
+	@Test
+	public void buildQueryOptions() {
+
+		QueryOptions queryOptions = QueryOptions.builder() //
+				.consistencyLevel(ConsistencyLevel.ANY) //
+				.retryPolicy(RetryPolicy.DEFAULT) //
+				.readTimeout(1, TimeUnit.SECONDS)//
+				.fetchSize(10)//
+				.tracing(true)//
+				.build(); //
+
+		assertThat((Class) queryOptions.getClass(), is(equalTo((Class) QueryOptions.class)));
+		assertThat(queryOptions.getRetryPolicy(), is(RetryPolicy.DEFAULT));
+		assertThat(queryOptions.getConsistencyLevel(), is(nullValue()));
+		assertThat(queryOptions.getDriverConsistencyLevel(), is(ConsistencyLevel.ANY));
+		assertThat(queryOptions.getReadTimeout(), is(1000L));
+		assertThat(queryOptions.getFetchSize(), is(10));
+		assertThat(queryOptions.getTracing(), is(true));
+	}
+
+	/**
+	 * @see DATACASS-202
+	 */
+	@Test
+	public void buildQueryOptionsWithDriverRetryPolicy() {
+
+		QueryOptions writeOptions = QueryOptions.builder() //
+				.retryPolicy(new LoggingRetryPolicy(DefaultRetryPolicy.INSTANCE)) //
+				.build(); //
+
+		assertThat(writeOptions.getRetryPolicy(), is(nullValue()));
+		assertThat(writeOptions.getDriverRetryPolicy(), is(instanceOf(LoggingRetryPolicy.class)));
+	}
+
+	/**
+	 * @see DATACASS-202
+	 */
+	@Test
+	public void buildQueryOptionsWithRetryPolicy() {
+
+		QueryOptions writeOptions = QueryOptions.builder() //
+				.retryPolicy(RetryPolicy.DOWNGRADING_CONSISTENCY) //
+				.build(); //
+
+		assertThat(writeOptions.getRetryPolicy(), is(RetryPolicy.DOWNGRADING_CONSISTENCY));
+		assertThat(writeOptions.getDriverRetryPolicy(), is(nullValue()));
+	}
+
+	/**
+	 * @see DATACASS-202
+	 */
+	@Test(expected = IllegalStateException.class)
+	public void builderShouldRejectSettingOurAndDriverRetryPolicy() {
+
+		QueryOptions.builder() //
+				.retryPolicy(RetryPolicy.DEFAULT).retryPolicy(FallthroughRetryPolicy.INSTANCE);
+	}
+
+	/**
+	 * @see DATACASS-202
+	 */
+	@Test(expected = IllegalStateException.class)
+	public void builderShouldRejectSettingDriverAndOurRetryPolicy() {
+
+		QueryOptions.builder() //
+				.retryPolicy(FallthroughRetryPolicy.INSTANCE)//
+				.retryPolicy(RetryPolicy.DEFAULT);
+
+	}
+
+	/**
+	 * @see DATACASS-202
+	 */
+	@Test(expected = IllegalStateException.class)
+	public void shouldRejectSettingOurAndDriverRetryPolicy() {
+
+		QueryOptions queryOptions = new QueryOptions();
+		queryOptions.setRetryPolicy(RetryPolicy.DEFAULT);
+		queryOptions.setRetryPolicy(FallthroughRetryPolicy.INSTANCE);
+	}
+
+	/**
+	 * @see DATACASS-202
+	 */
+	@Test(expected = IllegalStateException.class)
+	public void shouldRejectSettingDriverAndOurRetryPolicy() {
+
+		QueryOptions queryOptions = new QueryOptions();
+		queryOptions.setRetryPolicy(FallthroughRetryPolicy.INSTANCE);
+		queryOptions.setRetryPolicy(RetryPolicy.DEFAULT);
+	}
+
+	/**
+	 * @see DATACASS-202
+	 */
+	@Test(expected = IllegalStateException.class)
+	public void shouldRejectSettingOurAndDriverConsistencyLevel() {
+
+		QueryOptions queryOptions = new QueryOptions();
+		queryOptions.setConsistencyLevel(org.springframework.cassandra.core.ConsistencyLevel.ANY);
+		queryOptions.setConsistencyLevel(ConsistencyLevel.ANY);
+	}
+
+	/**
+	 * @see DATACASS-202
+	 */
+	@Test(expected = IllegalStateException.class)
+	public void shouldRejectSettingDriverAndOurConsistencyLevel() {
+
+		QueryOptions queryOptions = new QueryOptions();
+		queryOptions.setConsistencyLevel(ConsistencyLevel.ANY);
+		queryOptions.setConsistencyLevel(org.springframework.cassandra.core.ConsistencyLevel.ANY);
+	}
+}

spring-cql/src/test/java/org/springframework/cassandra/core/WriteOptionsUnitTests.java

@@ -0,0 +1,123 @@
+/*
+ * 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.cassandra.core;
+
+import static org.hamcrest.MatcherAssert.*;
+import static org.hamcrest.Matchers.*;
+
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Test;
+
+import com.datastax.driver.core.policies.FallthroughRetryPolicy;
+
+/**
+ * Unit tests for {@link WriteOptions}.
+ * 
+ * @author Mark Paluch
+ */
+public class WriteOptionsUnitTests {
+
+	/**
+	 * @see DATACASS-202
+	 */
+	@Test
+	public void buildWriteOptions() {
+
+		WriteOptions writeOptions = WriteOptions.builder() //
+				.consistencyLevel(com.datastax.driver.core.ConsistencyLevel.ANY) //
+				.ttl(123) //
+				.retryPolicy(RetryPolicy.DEFAULT) //
+				.readTimeout(1)//
+				.fetchSize(10)//
+				.withTracing()//
+				.build(); //
+
+		assertThat(writeOptions.getTtl(), is(123));
+		assertThat(writeOptions.getRetryPolicy(), is(RetryPolicy.DEFAULT));
+		assertThat(writeOptions.getConsistencyLevel(), is(nullValue()));
+		assertThat(writeOptions.getDriverConsistencyLevel(), is(com.datastax.driver.core.ConsistencyLevel.ANY));
+		assertThat(writeOptions.getReadTimeout(), is(1L));
+		assertThat(writeOptions.getFetchSize(), is(10));
+		assertThat(writeOptions.getTracing(), is(true));
+	}
+
+	/**
+	 * @see DATACASS-202
+	 */
+	@Test
+	public void buildReadTimeoutOptionsWriteOptions() {
+
+		WriteOptions writeOptions = WriteOptions.builder() //
+				.readTimeout(1, TimeUnit.MINUTES)//
+				.build(); //
+
+		assertThat(writeOptions.getReadTimeout(), is(60L * 1000L));
+		assertThat(writeOptions.getFetchSize(), is(nullValue()));
+		assertThat(writeOptions.getTracing(), is(nullValue()));
+	}
+
+	/**
+	 * @see DATACASS-202
+	 */
+	@Test
+	public void buildQueryOptionsWithDriverRetryPolicy() {
+
+		QueryOptions writeOptions = QueryOptions.builder() //
+				.retryPolicy(FallthroughRetryPolicy.INSTANCE) //
+				.build(); //
+
+		assertThat(writeOptions.getRetryPolicy(), is(nullValue()));
+		assertThat(writeOptions.getDriverRetryPolicy(),
+				is(equalTo((com.datastax.driver.core.policies.RetryPolicy) FallthroughRetryPolicy.INSTANCE)));
+	}
+
+	/**
+	 * @see DATACASS-202
+	 */
+	@Test
+	public void buildQueryOptionsWithRetryPolicy() {
+
+		QueryOptions writeOptions = QueryOptions.builder() //
+				.retryPolicy(RetryPolicy.DOWNGRADING_CONSISTENCY) //
+				.build(); //
+
+		assertThat(writeOptions.getRetryPolicy(), is(RetryPolicy.DOWNGRADING_CONSISTENCY));
+		assertThat(writeOptions.getDriverRetryPolicy(), is(nullValue()));
+	}
+
+	/**
+	 * @see DATACASS-202
+	 */
+	@Test(expected = IllegalStateException.class)
+	public void builderShouldRejectSettingOurAndDriverRetryPolicy() {
+
+		WriteOptions.builder() //
+				.retryPolicy(RetryPolicy.DEFAULT).retryPolicy(FallthroughRetryPolicy.INSTANCE);
+	}
+
+	/**
+	 * @see DATACASS-202
+	 */
+	@Test(expected = IllegalStateException.class)
+	public void builderShouldRejectSettingDriverAndOurRetryPolicy() {
+
+		WriteOptions.builder() //
+				.retryPolicy(FallthroughRetryPolicy.INSTANCE)//
+				.retryPolicy(RetryPolicy.DEFAULT);
+
+	}
+}

spring-cql/src/test/java/org/springframework/cassandra/test/integration/core/CqlOperationsIntegrationTests.java

@@ -15,6 +15,7 @@
  */
 package org.springframework.cassandra.test.integration.core;
 
+import static org.hamcrest.Matchers.*;
 import static org.junit.Assert.*;
 
 import java.util.Collection;
@@ -1087,7 +1088,27 @@ public class CqlOperationsIntegrationTests extends AbstractKeyspaceCreatingInteg
 
 		Truncate truncate = QueryBuilder.truncate(tableName);
 		cqlTemplate.execute(truncate);
+	}
 
+	/**
+	 * @see DATACASS-202
+	 */
+	@Test
+	public void queryShouldApplyFetchSize() {
+
+		insertTestObjectArray();
+
+		final String cql = "select * from book";
+
+		ResultSet oneByOneResultSet = cqlTemplate.query(cql, QueryOptions.builder().fetchSize(1).build());
+
+		assertThat(oneByOneResultSet.isFullyFetched(), is(false));
+		assertThat(oneByOneResultSet.getAvailableWithoutFetching(), is(1));
+
+		ResultSet fullResultSet = cqlTemplate.query(cql, QueryOptions.builder().fetchSize(10).build());
+
+		assertThat(fullResultSet.isFullyFetched(), is(true));
+		assertThat(fullResultSet.getAvailableWithoutFetching(), is(4));
 	}
 
 	/**

spring-cql/src/test/java/org/springframework/cassandra/test/integration/core/async/AsynchronousCqlOperationsIntegrationTests.java

@@ -269,7 +269,7 @@ public class AsynchronousCqlOperationsIntegrationTests extends AbstractKeyspaceC
 		new AsynchronousQueryListenerTestTemplate() {
 			@Override
 			void doAsyncQuery(Book b, QueryListener listener) {
-				cqlOperations.queryAsynchronously(cql(b), listener, new QueryOptions(cl, RetryPolicy.LOGGING));
+				cqlOperations.queryAsynchronously(cql(b), listener, new QueryOptions(cl, RetryPolicy.DEFAULT));
 			}
 		}.test();
 	}