DATACASS-767 - Polishing.

Reformat code. Add since tags. Extend keyspace association to PreparedStatements so that simple CQL statements get associated with the keyspace upon prepare. Remove check if a statement is a bound statement as prepared statements should not fail with applyStatementSettings(…).

Fix keyspace configuration in …Options objects as Insert/Update/Delete options should consider the keyspace when using the builder. Extend tests.

Simplify Cassandra version check. Ensure the discovery session gets closed after its use.

Original pull request: #177.

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

@@ -46,11 +46,12 @@ public class DeleteOptions extends WriteOptions {
 	private final @Nullable Filter ifCondition;
 
 	private DeleteOptions(@Nullable ConsistencyLevel consistencyLevel, ExecutionProfileResolver executionProfileResolver,
-			@Nullable Integer pageSize, @Nullable ConsistencyLevel serialConsistencyLevel, Duration timeout, Duration ttl,
-			@Nullable Long timestamp, @Nullable Boolean tracing, boolean ifExists, @Nullable Filter ifCondition) {
+			@Nullable CqlIdentifier keyspace, @Nullable Integer pageSize, @Nullable ConsistencyLevel serialConsistencyLevel,
+			Duration timeout, Duration ttl, @Nullable Long timestamp, @Nullable Boolean tracing, boolean ifExists,
+			@Nullable Filter ifCondition) {
 
-		super(consistencyLevel, executionProfileResolver, pageSize, serialConsistencyLevel, timeout, ttl, timestamp,
-				tracing);
+		super(consistencyLevel, executionProfileResolver, keyspace, pageSize, serialConsistencyLevel, timeout, ttl,
+				timestamp, tracing);
 
 		this.ifExists = ifExists;
 		this.ifCondition = ifCondition;
@@ -197,6 +198,16 @@ public class DeleteOptions extends WriteOptions {
 			return this;
 		}
 
+		/* (non-Javadoc)
+		 * @see org.springframework.data.cassandra.core.cql.WriteOptions.WriteOptionsBuilder#keyspace()
+		 */
+		@Override
+		public DeleteOptionsBuilder keyspace(CqlIdentifier keyspace) {
+
+			super.keyspace(keyspace);
+			return this;
+		}
+
 		/* (non-Javadoc)
 		 * @see org.springframework.data.cassandra.core.cql.WriteOptions.WriteOptionsBuilder#pageSize(int)
 		 */
@@ -307,15 +318,6 @@ public class DeleteOptions extends WriteOptions {
 			return this;
 		}
 
-		/* (non-Javadoc)
-		 * @see org.springframework.data.cassandra.core.cql.WriteOptions.WriteOptionsBuilder#keyspace()
-		 */
-		@Override
-		public DeleteOptionsBuilder keyspace(CqlIdentifier keyspace) {
-
-			super.keyspace(keyspace);
-			return this;
-		}
 
 		/**
 		 * Use light-weight transactions by applying {@code IF EXISTS}. Replaces a previous {@link #ifCondition(Filter)}.
@@ -378,7 +380,7 @@ public class DeleteOptions extends WriteOptions {
 		 */
 		public DeleteOptions build() {
 
-			return new DeleteOptions(this.consistencyLevel, this.executionProfileResolver, this.pageSize,
+			return new DeleteOptions(this.consistencyLevel, this.executionProfileResolver, this.keyspace, this.pageSize,
 					this.serialConsistencyLevel, this.timeout, this.ttl, this.timestamp, this.tracing, this.ifExists,
 					this.ifCondition);
 		}

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

@@ -38,16 +38,17 @@ public class InsertOptions extends WriteOptions {
 
 	private static final InsertOptions EMPTY = new InsertOptionsBuilder().build();
 
-	private boolean ifNotExists;
+	private final boolean ifNotExists;
 
-	private boolean insertNulls;
+	private final boolean insertNulls;
 
 	private InsertOptions(@Nullable ConsistencyLevel consistencyLevel, ExecutionProfileResolver executionProfileResolver,
-			@Nullable Integer pageSize, @Nullable ConsistencyLevel serialConsistencyLevel, Duration timeout, Duration ttl,
-			@Nullable Long timestamp, @Nullable Boolean tracing, boolean ifNotExists, boolean insertNulls) {
+			@Nullable CqlIdentifier keyspace, @Nullable Integer pageSize, @Nullable ConsistencyLevel serialConsistencyLevel,
+			Duration timeout, Duration ttl, @Nullable Long timestamp, @Nullable Boolean tracing, boolean ifNotExists,
+			boolean insertNulls) {
 
-		super(consistencyLevel, executionProfileResolver, pageSize, serialConsistencyLevel, timeout, ttl, timestamp,
-				tracing);
+		super(consistencyLevel, executionProfileResolver, keyspace, pageSize, serialConsistencyLevel, timeout, ttl,
+				timestamp, tracing);
 
 		this.ifNotExists = ifNotExists;
 		this.insertNulls = insertNulls;
@@ -197,6 +198,16 @@ public class InsertOptions extends WriteOptions {
 			return (InsertOptionsBuilder) super.fetchSize(fetchSize);
 		}
 
+		/* (non-Javadoc)
+		 * @see org.springframework.data.cassandra.core.cql.WriteOptions.WriteOptionsBuilder#keyspace()
+		 */
+		@Override
+		public InsertOptionsBuilder keyspace(CqlIdentifier keyspace) {
+
+			super.keyspace(keyspace);
+			return this;
+		}
+
 		/* (non-Javadoc)
 		 * @see org.springframework.data.cassandra.core.cql.WriteOptions.WriteOptionsBuilder#pageSize(int)
 		 */
@@ -306,16 +317,6 @@ public class InsertOptions extends WriteOptions {
 			return this;
 		}
 
-		/* (non-Javadoc)
-		 * @see org.springframework.data.cassandra.core.cql.WriteOptions.WriteOptionsBuilder#keyspace()
-		 */
-		@Override
-		public InsertOptionsBuilder keyspace(CqlIdentifier keyspace) {
-
-			super.keyspace(keyspace);
-			return this;
-		}
-
 		/**
 		 * Use light-weight transactions by applying {@code IF NOT EXISTS}.
 		 *
@@ -372,7 +373,7 @@ public class InsertOptions extends WriteOptions {
 		 * @return a new {@link InsertOptions} with the configured values
 		 */
 		public InsertOptions build() {
-			return new InsertOptions(this.consistencyLevel, this.executionProfileResolver, this.pageSize,
+			return new InsertOptions(this.consistencyLevel, this.executionProfileResolver, this.keyspace, this.pageSize,
 					this.serialConsistencyLevel, this.timeout, this.ttl, this.timestamp, this.tracing, this.ifNotExists,
 					this.insertNulls);
 		}

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

@@ -47,11 +47,12 @@ public class UpdateOptions extends WriteOptions {
 	private final @Nullable Filter ifCondition;
 
 	private UpdateOptions(@Nullable ConsistencyLevel consistencyLevel, ExecutionProfileResolver executionProfileResolver,
-			@Nullable Integer pageSize, @Nullable ConsistencyLevel serialConsistencyLevel, Duration timeout, Duration ttl,
-			@Nullable Long timestamp, @Nullable Boolean tracing, boolean ifExists, @Nullable Filter ifCondition) {
+			@Nullable CqlIdentifier keyspace, @Nullable Integer pageSize, @Nullable ConsistencyLevel serialConsistencyLevel,
+			Duration timeout, Duration ttl, @Nullable Long timestamp, @Nullable Boolean tracing, boolean ifExists,
+			@Nullable Filter ifCondition) {
 
-		super(consistencyLevel, executionProfileResolver, pageSize, serialConsistencyLevel, timeout, ttl, timestamp,
-				tracing);
+		super(consistencyLevel, executionProfileResolver, keyspace, pageSize, serialConsistencyLevel, timeout, ttl,
+				timestamp, tracing);
 
 		this.ifExists = ifExists;
 		this.ifCondition = ifCondition;
@@ -204,6 +205,16 @@ public class UpdateOptions extends WriteOptions {
 			return this;
 		}
 
+		/* (non-Javadoc)
+		 * @see org.springframework.data.cassandra.core.cql.WriteOptions.WriteOptionsBuilder#keyspace()
+		 */
+		@Override
+		public UpdateOptionsBuilder keyspace(CqlIdentifier keyspace) {
+
+			super.keyspace(keyspace);
+			return this;
+		}
+
 		/* (non-Javadoc)
 		 * @see org.springframework.data.cassandra.core.cql.WriteOptions.WriteOptionsBuilder#pageSize(int)
 		 */
@@ -314,16 +325,6 @@ public class UpdateOptions extends WriteOptions {
 			return this;
 		}
 
-		/* (non-Javadoc)
-		 * @see org.springframework.data.cassandra.core.cql.WriteOptions.WriteOptionsBuilder#keyspace()
-		 */
-		@Override
-		public UpdateOptionsBuilder keyspace(CqlIdentifier keyspace) {
-
-			super.keyspace(keyspace);
-			return this;
-		}
-
 		/**
 		 * Use light-weight transactions by applying {@code IF EXISTS}. Replaces a previous {@link #ifCondition(Filter)}.
 		 *
@@ -386,8 +387,7 @@ public class UpdateOptions extends WriteOptions {
 		 * @return a new {@link UpdateOptions} with the configured values
 		 */
 		public UpdateOptions build() {
-
-			return new UpdateOptions(this.consistencyLevel, this.executionProfileResolver, this.pageSize,
+			return new UpdateOptions(this.consistencyLevel, this.executionProfileResolver, this.keyspace, this.pageSize,
 					this.serialConsistencyLevel, this.timeout, this.ttl, this.timestamp, this.tracing, this.ifExists,
 					this.ifCondition);
 		}

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

@@ -22,13 +22,6 @@ import java.util.concurrent.CompletionStage;
 import java.util.concurrent.ExecutionException;
 import java.util.function.Function;
 
-import com.datastax.oss.driver.api.core.CqlSession;
-import com.datastax.oss.driver.api.core.DriverException;
-import com.datastax.oss.driver.api.core.cql.AsyncResultSet;
-import com.datastax.oss.driver.api.core.cql.PreparedStatement;
-import com.datastax.oss.driver.api.core.cql.ResultSet;
-import com.datastax.oss.driver.api.core.cql.Statement;
-
 import org.springframework.dao.DataAccessException;
 import org.springframework.dao.support.DataAccessUtils;
 import org.springframework.dao.support.PersistenceExceptionTranslator;
@@ -39,6 +32,14 @@ import org.springframework.util.Assert;
 import org.springframework.util.concurrent.ListenableFuture;
 import org.springframework.util.concurrent.SettableListenableFuture;
 
+import com.datastax.oss.driver.api.core.CqlSession;
+import com.datastax.oss.driver.api.core.DriverException;
+import com.datastax.oss.driver.api.core.cql.AsyncResultSet;
+import com.datastax.oss.driver.api.core.cql.PreparedStatement;
+import com.datastax.oss.driver.api.core.cql.ResultSet;
+import com.datastax.oss.driver.api.core.cql.SimpleStatement;
+import com.datastax.oss.driver.api.core.cql.Statement;
+
 /**
  * <b>This is the central class in the CQL core package for asynchronous Cassandra data access.</b> It simplifies the
  * use of CQL and helps to avoid common errors. It executes core CQL workflow, leaving application code to provide CQL
@@ -745,7 +746,8 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera
 	 * @return the new {@link AsyncPreparedStatementCreator} to use
 	 */
 	protected AsyncPreparedStatementCreator newAsyncPreparedStatementCreator(String cql) {
-		return new SimpleAsyncPreparedStatementCreator(cql,
+		return new SimpleAsyncPreparedStatementCreator(
+				(SimpleStatement) applyStatementSettings(SimpleStatement.newInstance(cql)),
 				ex -> translateExceptionIfPossible("PrepareStatement", cql, ex));
 	}
 
@@ -806,25 +808,23 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera
 
 		private final PersistenceExceptionTranslator exceptionTranslator;
 
-		private final String cql;
+		private final SimpleStatement statement;
 
-		private SimpleAsyncPreparedStatementCreator(String cql, PersistenceExceptionTranslator exceptionTranslator) {
+		private SimpleAsyncPreparedStatementCreator(SimpleStatement statement,
+				PersistenceExceptionTranslator exceptionTranslator) {
 
-			Assert.hasText(cql, "CQL must not be empty");
-
-			this.cql = cql;
+			this.statement = statement;
 			this.exceptionTranslator = exceptionTranslator;
 		}
 
 		@Override
-		public String getCql() {
-			return this.cql;
+		public ListenableFuture<PreparedStatement> createPreparedStatement(CqlSession session) throws DriverException {
+			return new CassandraFutureAdapter<>(session.prepareAsync(this.statement), exceptionTranslator);
 		}
 
 		@Override
-		public ListenableFuture<PreparedStatement> createPreparedStatement(CqlSession session) throws DriverException {
-			return new CassandraFutureAdapter<>(session.prepareAsync(getCql()),
-					exceptionTranslator::translateExceptionIfPossible);
+		public String getCql() {
+			return this.statement.getQuery();
 		}
 	}
 

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

@@ -32,7 +32,6 @@ import com.datastax.oss.driver.api.core.ConsistencyLevel;
 import com.datastax.oss.driver.api.core.CqlIdentifier;
 import com.datastax.oss.driver.api.core.CqlSession;
 import com.datastax.oss.driver.api.core.cql.BatchStatement;
-import com.datastax.oss.driver.api.core.cql.BoundStatement;
 import com.datastax.oss.driver.api.core.cql.SimpleStatement;
 import com.datastax.oss.driver.api.core.cql.Statement;
 import com.datastax.oss.driver.api.core.retry.RetryPolicy;
@@ -64,6 +63,12 @@ public class CassandraAccessor implements InitializingBean {
 	 */
 	private ExecutionProfileResolver executionProfileResolver = ExecutionProfileResolver.none();
 
+	/**
+	 * If this variable is set to a value, it will be used for setting the {@code keyspace} property on statements used
+	 * for query processing.
+	 */
+	private @Nullable CqlIdentifier keyspace;
+
 	/**
 	 * If this variable is set to a non-negative value, it will be used for setting the {@code pageSize} property on
 	 * statements used for query processing.
@@ -84,12 +89,6 @@ public class CassandraAccessor implements InitializingBean {
 
 	private @Nullable SessionFactory sessionFactory;
 
-	/**
-	 * If this variable is set to a value, it will be used for setting the {@code keyspace} property on statements used
-	 * for query processing.
-	 */
-	private @Nullable CqlIdentifier keyspace;
-
 	/**
 	 * Ensures the Cassandra {@link CqlSession} and exception translator has been propertly set.
 	 */
@@ -200,6 +199,31 @@ public class CassandraAccessor implements InitializingBean {
 		return getPageSize();
 	}
 
+	/**
+	 * Set the {@link CqlIdentifier keyspace} to be applied on statement-level for this template. If not set, the default
+	 * {@link CqlSession} keyspace will be used.
+	 *
+	 * @param keyspace the keyspace to apply, must not be {@literal null}.
+	 * @see SimpleStatement#setKeyspace(CqlIdentifier)
+	 * @see BatchStatement#setKeyspace(CqlIdentifier)
+	 * @since 3.1
+	 */
+	public void setKeyspace(CqlIdentifier keyspace) {
+
+		Assert.notNull(keyspace, "Keyspace must not be null");
+
+		this.keyspace = keyspace;
+	}
+
+	/**
+	 * @return the {@link CqlIdentifier keyspace} to be applied on statement-level for this template.
+	 * @since 3.1
+	 */
+	@Nullable
+	public CqlIdentifier getKeyspace() {
+		return this.keyspace;
+	}
+
 	/**
 	 * Set the page size for this template. This is important for processing large result sets: Setting this higher than
 	 * the default value will increase processing speed at the cost of memory consumption; setting this lower can avoid
@@ -297,24 +321,6 @@ public class CassandraAccessor implements InitializingBean {
 		return this.sessionFactory;
 	}
 
-	/**
-	 * Set the keyspace for this template. If it is null, then the default {@link CqlSession} level keyspace will be used.
-	 *
-	 * @see SimpleStatement#setKeyspace(CqlIdentifier)
-	 * @see BatchStatement#setKeyspace(CqlIdentifier)
-	 */
-	public void setKeyspace(@Nullable CqlIdentifier keyspace) {
-		this.keyspace = keyspace;
-	}
-
-	/**
-	 * @return the {@link CqlIdentifier} keyspace for this template.
-	 */
-	@Nullable
-	public CqlIdentifier getKeyspace() {
-		return this.keyspace;
-	}
-
 	/**
 	 * Create a {@link SimpleStatement} given {@code cql}.
 	 *
@@ -357,9 +363,6 @@ public class CassandraAccessor implements InitializingBean {
 		}
 
 		if (keyspace != null) {
-			if (statementToUse instanceof BoundStatement) {
-				throw new IllegalArgumentException("Keyspace cannot be set for a BoundStatement");
-			}
 			if (statementToUse instanceof BatchStatement) {
 				statementToUse = ((BatchStatement) statementToUse).setKeyspace(keyspace);
 			}

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

@@ -32,6 +32,7 @@ import com.datastax.oss.driver.api.core.DriverException;
 import com.datastax.oss.driver.api.core.cql.PreparedStatement;
 import com.datastax.oss.driver.api.core.cql.ResultSet;
 import com.datastax.oss.driver.api.core.cql.Row;
+import com.datastax.oss.driver.api.core.cql.SimpleStatement;
 import com.datastax.oss.driver.api.core.cql.Statement;
 import com.datastax.oss.driver.api.core.metadata.Node;
 
@@ -726,7 +727,8 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
 	 * @return the new {@link PreparedStatementCreator} to use
 	 */
 	protected PreparedStatementCreator newPreparedStatementCreator(String cql) {
-		return new SimplePreparedStatementCreator(cql);
+		return new SimplePreparedStatementCreator(
+				(SimpleStatement) applyStatementSettings(SimpleStatement.newInstance(cql)));
 	}
 
 	/**

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

@@ -43,6 +43,8 @@ public class QueryOptions {
 
 	private final ExecutionProfileResolver executionProfileResolver;
 
+	private final @Nullable CqlIdentifier keyspace;
+
 	private final @Nullable Integer pageSize;
 
 	private final @Nullable ConsistencyLevel serialConsistencyLevel;
@@ -51,25 +53,17 @@ public class QueryOptions {
 
 	private final @Nullable Boolean tracing;
 
-	private final @Nullable CqlIdentifier keyspace;
-
 	protected QueryOptions(@Nullable ConsistencyLevel consistencyLevel, ExecutionProfileResolver executionProfileResolver,
-			@Nullable Integer pageSize, @Nullable ConsistencyLevel serialConsistencyLevel, Duration timeout,
-			@Nullable Boolean tracing) {
-		this(consistencyLevel, executionProfileResolver, pageSize, serialConsistencyLevel, timeout, tracing, null);
-	}
-
-	protected QueryOptions(@Nullable ConsistencyLevel consistencyLevel, ExecutionProfileResolver executionProfileResolver,
-			@Nullable Integer pageSize, @Nullable ConsistencyLevel serialConsistencyLevel, Duration timeout,
-			@Nullable Boolean tracing, @Nullable CqlIdentifier keyspace) {
+			@Nullable CqlIdentifier keyspace, @Nullable Integer pageSize, @Nullable ConsistencyLevel serialConsistencyLevel,
+			Duration timeout, @Nullable Boolean tracing) {
 
 		this.consistencyLevel = consistencyLevel;
 		this.executionProfileResolver = executionProfileResolver;
+		this.keyspace = keyspace;
 		this.pageSize = pageSize;
 		this.serialConsistencyLevel = serialConsistencyLevel;
 		this.timeout = timeout;
 		this.tracing = tracing;
-		this.keyspace = keyspace;
 	}
 
 	/**
@@ -167,8 +161,9 @@ public class QueryOptions {
 	}
 
 	/**
-	 * @return the keyspace associated with the query. If it is null, it means that the default keyspace from
-	 *         {@link CqlSession} will be used.
+	 * @return the keyspace associated with the query. If it is {@literal null}, it means that either keyspace configured
+	 *         on the statement or from the {@link CqlSession} will be used.
+	 * @since 3.1
 	 */
 	@Nullable
 	public CqlIdentifier getKeyspace() {
@@ -247,6 +242,8 @@ public class QueryOptions {
 
 		protected ExecutionProfileResolver executionProfileResolver = ExecutionProfileResolver.none();
 
+		protected @Nullable CqlIdentifier keyspace;
+
 		protected @Nullable Integer pageSize;
 
 		protected @Nullable ConsistencyLevel serialConsistencyLevel;
@@ -255,19 +252,17 @@ public class QueryOptions {
 
 		protected @Nullable Boolean tracing;
 
-		protected @Nullable CqlIdentifier keyspace;
-
 		QueryOptionsBuilder() {}
 
 		QueryOptionsBuilder(QueryOptions queryOptions) {
 
 			this.consistencyLevel = queryOptions.consistencyLevel;
 			this.executionProfileResolver = queryOptions.executionProfileResolver;
+			this.keyspace = queryOptions.keyspace;
 			this.pageSize = queryOptions.pageSize;
 			this.serialConsistencyLevel = queryOptions.serialConsistencyLevel;
 			this.timeout = queryOptions.timeout;
 			this.tracing = queryOptions.tracing;
-			this.keyspace = queryOptions.keyspace;
 		}
 
 		/**
@@ -330,6 +325,23 @@ public class QueryOptions {
 			return pageSize(fetchSize);
 		}
 
+		/**
+		 * Sets the {@link CqlIdentifier keyspace} to use. If left unconfigured, then the keyspace set on the statement or
+		 * {@link CqlSession} will be used.
+		 *
+		 * @param keyspace the specific keyspace to use to run a statement, must not be {@literal null}.
+		 * @return {@code this} {@link QueryOptionsBuilder}.
+		 * @since 3.1
+		 */
+		public QueryOptionsBuilder keyspace(CqlIdentifier keyspace) {
+
+			Assert.notNull(keyspace, "Keyspace must not be null");
+
+			this.keyspace = keyspace;
+
+			return this;
+		}
+
 		/**
 		 * Sets the query fetch size for {@link com.datastax.oss.driver.api.core.cql.ResultSet} chunks.
 		 * <p>
@@ -459,27 +471,14 @@ public class QueryOptions {
 			return tracing(true);
 		}
 
-		/**
-		 * Sets the keyspace to use.
-		 *
-		 * @param keyspace if it is null, then the default {@link CqlSession} level keyspace will be used.
-		 * @return {@code this} {@link QueryOptionsBuilder}
-		 */
-		public QueryOptionsBuilder keyspace(CqlIdentifier keyspace) {
-
-			this.keyspace = keyspace;
-
-			return this;
-		}
-
 		/**
 		 * Builds a new {@link QueryOptions} with the configured values.
 		 *
 		 * @return a new {@link QueryOptions} with the configured values
 		 */
 		public QueryOptions build() {
-			return new QueryOptions(this.consistencyLevel, this.executionProfileResolver, this.pageSize,
-					this.serialConsistencyLevel, this.timeout, this.tracing, this.keyspace);
+			return new QueryOptions(this.consistencyLevel, this.executionProfileResolver, this.keyspace, this.pageSize,
+					this.serialConsistencyLevel, this.timeout, this.tracing);
 		}
 	}
 }

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

@@ -23,6 +23,7 @@ import java.util.Optional;
 import java.util.function.Function;
 
 import org.reactivestreams.Publisher;
+
 import org.springframework.dao.DataAccessException;
 import org.springframework.dao.support.DataAccessUtils;
 import org.springframework.data.cassandra.ReactiveResultSet;
@@ -84,12 +85,6 @@ import com.datastax.oss.driver.api.core.retry.RetryPolicy;
  */
 public class ReactiveCqlTemplate extends ReactiveCassandraAccessor implements ReactiveCqlOperations {
 
-	/**
-	 * If this variable is set to a non-negative value, it will be used for setting the {@code pageSize} property on
-	 * statements used for query processing.
-	 */
-	private int pageSize = -1;
-
 	/**
 	 * If this variable is set to a value, it will be used for setting the {@code consistencyLevel} property on statements
 	 * used for query processing.
@@ -102,18 +97,24 @@ public class ReactiveCqlTemplate extends ReactiveCassandraAccessor implements Re
 	 */
 	private ExecutionProfileResolver executionProfileResolver = ExecutionProfileResolver.none();
 
-	/**
-	 * If this variable is set to a value, it will be used for setting the serial {@code consistencyLevel} property on
-	 * statements used for query processing.
-	 */
-	private @Nullable ConsistencyLevel serialConsistencyLevel;
-
 	/**
 	 * If this variable is set to a value, it will be used for setting the {@code keyspace} property on statements used
 	 * for query processing.
 	 */
 	private @Nullable CqlIdentifier keyspace;
 
+	/**
+	 * If this variable is set to a non-negative value, it will be used for setting the {@code pageSize} property on
+	 * statements used for query processing.
+	 */
+	private int pageSize = -1;
+
+	/**
+	 * If this variable is set to a value, it will be used for setting the serial {@code consistencyLevel} property on
+	 * statements used for query processing.
+	 */
+	private @Nullable ConsistencyLevel serialConsistencyLevel;
+
 	/**
 	 * Construct a new {@link ReactiveCqlTemplate Note: The {@link ReactiveSessionFactory} has to be set before using the
 	 * instance.
@@ -226,6 +227,31 @@ public class ReactiveCqlTemplate extends ReactiveCassandraAccessor implements Re
 		return getPageSize();
 	}
 
+	/**
+	 * Set the {@link CqlIdentifier keyspace} to be applied on statement-level for this template. If not set, the default
+	 * {@link CqlSession} keyspace will be used.
+	 *
+	 * @param keyspace the keyspace to apply, must not be {@literal null}.
+	 * @see SimpleStatement#setKeyspace(CqlIdentifier)
+	 * @see BatchStatement#setKeyspace(CqlIdentifier)
+	 * @since 3.1
+	 */
+	public void setKeyspace(CqlIdentifier keyspace) {
+
+		Assert.notNull(keyspace, "Keyspace must not be null");
+
+		this.keyspace = keyspace;
+	}
+
+	/**
+	 * @return the {@link CqlIdentifier keyspace} to be applied on statement-level for this template.
+	 * @since 3.1
+	 */
+	@Nullable
+	public CqlIdentifier getKeyspace() {
+		return this.keyspace;
+	}
+
 	/**
 	 * Set the page size for this template. This is important for processing large result sets: Setting this higher than
 	 * the default value will increase processing speed at the cost of memory consumption; setting this lower can avoid
@@ -265,24 +291,6 @@ public class ReactiveCqlTemplate extends ReactiveCassandraAccessor implements Re
 		return this.serialConsistencyLevel;
 	}
 
-	/**
-	 * Set the keyspace for this template. If it is null, then the default {@link CqlSession} level keyspace will be used.
-	 *
-	 * @see SimpleStatement#setKeyspace(CqlIdentifier)
-	 * @see BatchStatement#setKeyspace(CqlIdentifier)
-	 */
-	public void setKeyspace(@Nullable CqlIdentifier keyspace) {
-		this.keyspace = keyspace;
-	}
-
-	/**
-	 * @return the {@link CqlIdentifier} keyspace for this template.
-	 */
-	@Nullable
-	public CqlIdentifier getKeyspace() {
-		return this.keyspace;
-	}
-
 	// -------------------------------------------------------------------------
 	// Methods dealing with a plain org.springframework.data.cassandra.core.cql.ReactiveSession
 	// -------------------------------------------------------------------------
@@ -754,7 +762,8 @@ public class ReactiveCqlTemplate extends ReactiveCassandraAccessor implements Re
 	 * @since 2.0.8
 	 */
 	protected ReactivePreparedStatementCreator newReactivePreparedStatementCreator(String cql) {
-		return new SimpleReactivePreparedStatementCreator(cql);
+		return new SimpleReactivePreparedStatementCreator(
+				(SimpleStatement) applyStatementSettings(SimpleStatement.newInstance(cql)));
 	}
 
 	/**
@@ -865,9 +874,6 @@ public class ReactiveCqlTemplate extends ReactiveCassandraAccessor implements Re
 		}
 
 		if (keyspace != null) {
-			if (statementToUse instanceof BoundStatement) {
-				throw new IllegalArgumentException("Keyspace cannot be set for a BoundStatement");
-			}
 			if (statementToUse instanceof BatchStatement) {
 				statementToUse = ((BatchStatement) statementToUse).setKeyspace(keyspace);
 			}
@@ -922,23 +928,20 @@ public class ReactiveCqlTemplate extends ReactiveCassandraAccessor implements Re
 
 	static class SimpleReactivePreparedStatementCreator implements ReactivePreparedStatementCreator, CqlProvider {
 
-		private final String cql;
+		private final SimpleStatement statement;
 
-		SimpleReactivePreparedStatementCreator(String cql) {
-
-			Assert.notNull(cql, "CQL must not be null");
-
-			this.cql = cql;
+		SimpleReactivePreparedStatementCreator(SimpleStatement statement) {
+			this.statement = statement;
 		}
 
 		@Override
 		public Mono<PreparedStatement> createPreparedStatement(ReactiveSession session) throws DriverException {
-			return session.prepare(cql);
+			return session.prepare(this.statement);
 		}
 
 		@Override
 		public String getCql() {
-			return cql;
+			return this.statement.getQuery();
 		}
 	}
 }

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

@@ -15,11 +15,12 @@
  */
 package org.springframework.data.cassandra.core.cql;
 
+import org.springframework.util.Assert;
+
 import com.datastax.oss.driver.api.core.CqlSession;
 import com.datastax.oss.driver.api.core.DriverException;
 import com.datastax.oss.driver.api.core.cql.PreparedStatement;
-
-import org.springframework.util.Assert;
+import com.datastax.oss.driver.api.core.cql.SimpleStatement;
 
 /**
  * Trivial implementation of {@link PreparedStatementCreator}. This prepared statement creator simply prepares a
@@ -34,7 +35,7 @@ import org.springframework.util.Assert;
  */
 public class SimplePreparedStatementCreator implements PreparedStatementCreator, CqlProvider {
 
-	private final String cql;
+	private final SimpleStatement statement;
 
 	/**
 	 * Create a {@link SimplePreparedStatementCreator} given {@code cql}.
@@ -45,14 +46,20 @@ public class SimplePreparedStatementCreator implements PreparedStatementCreator,
 
 		Assert.notNull(cql, "CQL is required to create a PreparedStatement");
 
-		this.cql = cql;
+		this.statement = SimpleStatement.newInstance(cql);
 	}
 
-	/* (non-Javadoc)
-	 * @see org.springframework.data.cassandra.core.cqlProvider#getCql()
+	/**
+	 * Create a {@link SimplePreparedStatementCreator} given {@code cql}.
+	 *
+	 * @param statement must not be {@literal null}.
+	 * @since 3.1
 	 */
-	public String getCql() {
-		return this.cql;
+	public SimplePreparedStatementCreator(SimpleStatement statement) {
+
+		Assert.notNull(statement, "CQL is required to create a PreparedStatement");
+
+		this.statement = statement;
 	}
 
 	/* (non-Javadoc)
@@ -60,6 +67,13 @@ public class SimplePreparedStatementCreator implements PreparedStatementCreator,
 	 */
 	@Override
 	public PreparedStatement createPreparedStatement(CqlSession session) throws DriverException {
-		return session.prepare(this.cql);
+		return session.prepare(this.statement);
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.cassandra.core.cqlProvider#getCql()
+	 */
+	public String getCql() {
+		return this.statement.getQuery();
 	}
 }

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

@@ -41,13 +41,15 @@ public class WriteOptions extends QueryOptions {
 	private static final WriteOptions EMPTY = new WriteOptionsBuilder().build();
 
 	private final Duration ttl;
+
 	private final @Nullable Long timestamp;
 
 	protected WriteOptions(@Nullable ConsistencyLevel consistencyLevel, ExecutionProfileResolver executionProfileResolver,
-			@Nullable Integer pageSize, @Nullable ConsistencyLevel serialConsistencyLevel, Duration timeout, Duration ttl,
+			@Nullable CqlIdentifier keyspace, @Nullable Integer pageSize, @Nullable ConsistencyLevel serialConsistencyLevel,
+			Duration timeout, Duration ttl,
 			@Nullable Long timestamp, @Nullable Boolean tracing) {
 
-		super(consistencyLevel, executionProfileResolver, pageSize, serialConsistencyLevel, timeout, tracing);
+		super(consistencyLevel, executionProfileResolver, keyspace, pageSize, serialConsistencyLevel, timeout, tracing);
 
 		this.ttl = ttl;
 		this.timestamp = timestamp;
@@ -150,6 +152,7 @@ public class WriteOptions extends QueryOptions {
 	public static class WriteOptionsBuilder extends QueryOptionsBuilder {
 
 		protected Duration ttl = Duration.ofMillis(-1);
+
 		protected Long timestamp = null;
 
 		protected WriteOptionsBuilder() {}
@@ -201,6 +204,16 @@ public class WriteOptions extends QueryOptions {
 			return this;
 		}
 
+		/* (non-Javadoc)
+		 * @see org.springframework.data.cassandra.core.cql.QueryOptions.QueryOptionsBuilder#keyspace()
+		 */
+		@Override
+		public WriteOptionsBuilder keyspace(CqlIdentifier keyspace) {
+
+			super.keyspace(keyspace);
+			return this;
+		}
+
 		/* (non-Javadoc)
 		 * @see org.springframework.data.cassandra.core.cql.QueryOptions.QueryOptionsBuilder#pageSize(int)
 		 */
@@ -272,16 +285,6 @@ public class WriteOptions extends QueryOptions {
 			return this;
 		}
 
-		/* (non-Javadoc)
-		 * @see org.springframework.data.cassandra.core.cql.QueryOptions.QueryOptionsBuilder#keyspace()
-		 */
-		@Override
-		public WriteOptionsBuilder keyspace(CqlIdentifier keyspace) {
-
-			super.keyspace(keyspace);
-			return this;
-		}
-
 		/**
 		 * Sets the time to live in seconds for write operations.
 		 *
@@ -353,7 +356,7 @@ public class WriteOptions extends QueryOptions {
 		 * @return a new {@link WriteOptions} with the configured values
 		 */
 		public WriteOptions build() {
-			return new WriteOptions(this.consistencyLevel, this.executionProfileResolver, this.pageSize,
+			return new WriteOptions(this.consistencyLevel, this.executionProfileResolver, this.keyspace, this.pageSize,
 					this.serialConsistencyLevel, this.timeout, this.ttl, this.timestamp, this.tracing);
 		}
 	}

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

@@ -16,6 +16,7 @@
 package org.springframework.data.cassandra.core;
 
 import static org.assertj.core.api.Assertions.*;
+import static org.assertj.core.api.Assumptions.assumeThat;
 import static org.junit.Assume.*;
 import static org.springframework.data.cassandra.core.query.Criteria.*;
 
@@ -438,7 +439,8 @@ class CassandraTemplateIntegrationTests extends AbstractKeyspaceCreatingIntegrat
 
 	@Test // DATACASS-767
 	void selectByQueryWithKeyspaceShouldRetrieveData() {
-		assumeTrue(cassandraVersion.isGreaterThanOrEqualTo(CASSANDRA_4));
+
+		assumeThat(cassandraVersion.isGreaterThanOrEqualTo(CASSANDRA_4)).isTrue();
 
 		QueryOptions queryOptions = QueryOptions.builder().keyspace(CqlIdentifier.fromCql(keyspace)).build();
 
@@ -451,7 +453,8 @@ class CassandraTemplateIntegrationTests extends AbstractKeyspaceCreatingIntegrat
 
 	@Test // DATACASS-767
 	void selectByQueryWithNonExistingKeyspaceShouldThrowThatKeyspaceDoesNotExists() {
-		assumeTrue(cassandraVersion.isGreaterThanOrEqualTo(CASSANDRA_4));
+
+		assumeThat(cassandraVersion.isGreaterThanOrEqualTo(CASSANDRA_4)).isTrue();
 
 		QueryOptions queryOptions = QueryOptions.builder().keyspace(CqlIdentifier.fromCql("non_existing")).build();
 

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

@@ -20,6 +20,7 @@ import static org.assertj.core.api.Assertions.*;
 import java.time.Duration;
 import java.time.Instant;
 
+import com.datastax.oss.driver.api.core.CqlIdentifier;
 import org.junit.jupiter.api.Test;
 
 import org.springframework.data.cassandra.core.query.Query;
@@ -33,7 +34,7 @@ import com.datastax.oss.driver.api.core.DefaultConsistencyLevel;
  */
 class DeleteOptionsUnitTests {
 
-	@Test // DATACASS-575, DATACASS-708
+	@Test // DATACASS-575, DATACASS-708, DATACASS-767
 	void shouldConfigureDeleteOptions() {
 
 		Instant now = Instant.ofEpochSecond(1234);
@@ -44,13 +45,14 @@ class DeleteOptionsUnitTests {
 				.withIfExists() //
 				.executionProfile("foo") //
 				.serialConsistencyLevel(DefaultConsistencyLevel.LOCAL_ONE) //
+				.keyspace(CqlIdentifier.fromCql("my_keyspace"))
 				.build();
 
-
 		assertThat(deleteOptions.getTtl()).isEqualTo(Duration.ofSeconds(10));
 		assertThat(deleteOptions.getTimestamp()).isEqualTo(now.toEpochMilli() * 1000);
 		assertThat(deleteOptions.isIfExists()).isTrue();
 		assertThat(deleteOptions.getIfCondition()).isNull();
+		assertThat(deleteOptions.getKeyspace()).isEqualTo(CqlIdentifier.fromCql("my_keyspace"));
 	}
 
 	@Test // DATACASS-575

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

@@ -24,6 +24,7 @@ import java.time.ZoneOffset;
 
 import org.junit.jupiter.api.Test;
 
+import com.datastax.oss.driver.api.core.CqlIdentifier;
 import com.datastax.oss.driver.api.core.DefaultConsistencyLevel;
 
 /**
@@ -34,7 +35,7 @@ import com.datastax.oss.driver.api.core.DefaultConsistencyLevel;
  */
 class InsertOptionsUnitTests {
 
-	@Test // DATACASS-250, DATACASS-155, DATACASS-708
+	@Test // DATACASS-250, DATACASS-155, DATACASS-708, DATACASS-767
 	void shouldConfigureInsertOptions() {
 
 		Instant now = LocalDateTime.now().toInstant(ZoneOffset.UTC);
@@ -45,11 +46,13 @@ class InsertOptionsUnitTests {
 				.withIfNotExists() //
 				.executionProfile("foo") //
 				.serialConsistencyLevel(DefaultConsistencyLevel.LOCAL_ONE) //
+				.keyspace(CqlIdentifier.fromCql("my_keyspace"))
 				.build();
 
 		assertThat(insertOptions.getTtl()).isEqualTo(Duration.ofSeconds(10));
 		assertThat(insertOptions.getTimestamp()).isEqualTo(now.toEpochMilli() * 1000);
 		assertThat(insertOptions.isIfNotExists()).isTrue();
+		assertThat(insertOptions.getKeyspace()).isEqualTo(CqlIdentifier.fromCql("my_keyspace"));
 	}
 
 	@Test // DATACASS-56

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

@@ -24,6 +24,7 @@ import org.junit.jupiter.api.Test;
 
 import org.springframework.data.cassandra.core.query.Query;
 
+import com.datastax.oss.driver.api.core.CqlIdentifier;
 import com.datastax.oss.driver.api.core.DefaultConsistencyLevel;
 
 /**
@@ -34,7 +35,7 @@ import com.datastax.oss.driver.api.core.DefaultConsistencyLevel;
  */
 class UpdateOptionsUnitTests {
 
-	@Test // DATACASS-250, DATACASS-155, DATACASS-708
+	@Test // DATACASS-250, DATACASS-155, DATACASS-708, DATACASS-767
 	void shouldConfigureUpdateOptions() {
 
 		Instant now = Instant.ofEpochSecond(1234);
@@ -45,12 +46,14 @@ class UpdateOptionsUnitTests {
 				.executionProfile("foo") //
 				.serialConsistencyLevel(DefaultConsistencyLevel.LOCAL_ONE) //
 				.withIfExists() //
+				.keyspace(CqlIdentifier.fromCql("my_keyspace")) //
 				.build();
 
 		assertThat(updateOptions.getTtl()).isEqualTo(Duration.ofSeconds(10));
 		assertThat(updateOptions.getTimestamp()).isEqualTo(now.toEpochMilli() * 1000);
 		assertThat(updateOptions.isIfExists()).isTrue();
 		assertThat(updateOptions.getIfCondition()).isNull();
+		assertThat(updateOptions.getKeyspace()).isEqualTo(CqlIdentifier.fromCql("my_keyspace"));
 	}
 
 	@Test // DATACASS-56, DATACASS-155

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

@@ -46,6 +46,7 @@ import org.springframework.scheduling.annotation.AsyncResult;
 import org.springframework.util.concurrent.ListenableFuture;
 
 import com.datastax.oss.driver.api.core.ConsistencyLevel;
+import com.datastax.oss.driver.api.core.CqlIdentifier;
 import com.datastax.oss.driver.api.core.CqlSession;
 import com.datastax.oss.driver.api.core.NoNodeAvailableException;
 import com.datastax.oss.driver.api.core.cql.AsyncResultSet;
@@ -64,7 +65,7 @@ import com.datastax.oss.driver.api.core.servererrors.InvalidQueryException;
  */
 @ExtendWith(MockitoExtension.class)
 @MockitoSettings(strictness = Strictness.LENIENT)
-public class AsyncCqlTemplateUnitTests {
+class AsyncCqlTemplateUnitTests {
 
 	@Mock CqlSession session;
 	@Mock AsyncResultSet resultSet;
@@ -133,10 +134,24 @@ public class AsyncCqlTemplateUnitTests {
 		});
 	}
 
+	@Test // DATACASS-767
+	void executePreparedStatementShouldApplyKeyspace() {
+
+		doTestStrings(null, null, CqlIdentifier.fromCql("ks1"), cqlTemplate -> {
+			cqlTemplate.execute("SELECT * from USERS", (session, ps) -> session.executeAsync(ps.bind("A")));
+		});
+
+		ArgumentCaptor<SimpleStatement> captor = ArgumentCaptor.forClass(SimpleStatement.class);
+		verify(session).prepareAsync(captor.capture());
+
+		SimpleStatement statement = captor.getValue();
+		assertThat(statement.getKeyspace()).isEqualTo(CqlIdentifier.fromCql("ks1"));
+	}
+
 	@Test // DATACASS-292
 	void executeCqlWithArgumentsShouldCallExecution() {
 
-		doTestStrings(5, ConsistencyLevel.ONE, asyncCqlTemplate -> {
+		doTestStrings(5, ConsistencyLevel.ONE, null, asyncCqlTemplate -> {
 
 			asyncCqlTemplate.execute("SELECT * from USERS");
 
@@ -172,7 +187,7 @@ public class AsyncCqlTemplateUnitTests {
 	@Test // DATACASS-292
 	void queryWithResultSetExtractorWithArgumentsShouldCallExecution() {
 
-		doTestStrings(5, ConsistencyLevel.ONE, asyncCqlTemplate -> {
+		doTestStrings(5, ConsistencyLevel.ONE, null, asyncCqlTemplate -> {
 
 			List<String> rows = getUninterruptibly(
 					asyncCqlTemplate.query("SELECT * from USERS", (row, index) -> row.getString(0)));
@@ -313,7 +328,7 @@ public class AsyncCqlTemplateUnitTests {
 	@Test // DATACASS-292
 	void executeStatementWithArgumentsShouldCallExecution() {
 
-		doTestStrings(5, ConsistencyLevel.ONE, asyncCqlTemplate -> {
+		doTestStrings(5, ConsistencyLevel.ONE, null, asyncCqlTemplate -> {
 
 			asyncCqlTemplate.execute(SimpleStatement.newInstance("SELECT * from USERS"));
 
@@ -350,7 +365,7 @@ public class AsyncCqlTemplateUnitTests {
 	@Test // DATACASS-292
 	void queryWithResultSetStatementExtractorWithArgumentsShouldCallExecution() {
 
-		doTestStrings(5, ConsistencyLevel.ONE, asyncCqlTemplate -> {
+		doTestStrings(5, ConsistencyLevel.ONE, null, asyncCqlTemplate -> {
 
 			ListenableFuture<List<String>> future = asyncCqlTemplate.query(SimpleStatement.newInstance("SELECT * from USERS"),
 					(row, index) -> row.getString(0));
@@ -657,7 +672,7 @@ public class AsyncCqlTemplateUnitTests {
 	@Test // DATACASS-292
 	void queryForObjectPreparedStatementShouldBeEmpty() throws Exception {
 
-		when(session.prepareAsync("SELECT * FROM user WHERE username = ?"))
+		when(session.prepareAsync(any(SimpleStatement.class)))
 				.thenReturn(new TestPreparedStatementFuture(preparedStatement));
 		when(preparedStatement.bind("Walter")).thenReturn(boundStatement);
 		when(session.executeAsync(boundStatement)).thenReturn(new TestResultSetFuture(resultSet));
@@ -679,7 +694,7 @@ public class AsyncCqlTemplateUnitTests {
 	@Test // DATACASS-292
 	void queryForObjectPreparedStatementShouldReturnRecord() {
 
-		when(session.prepareAsync("SELECT * FROM user WHERE username = ?"))
+		when(session.prepareAsync(any(SimpleStatement.class)))
 				.thenReturn(new TestPreparedStatementFuture(preparedStatement));
 		when(preparedStatement.bind("Walter")).thenReturn(boundStatement);
 		when(session.executeAsync(boundStatement)).thenReturn(new TestResultSetFuture(resultSet));
@@ -693,7 +708,7 @@ public class AsyncCqlTemplateUnitTests {
 	@Test // DATACASS-292
 	void queryForObjectPreparedStatementShouldFailReturningManyRecords() throws Exception {
 
-		when(session.prepareAsync("SELECT * FROM user WHERE username = ?"))
+		when(session.prepareAsync(any(SimpleStatement.class)))
 				.thenReturn(new TestPreparedStatementFuture(preparedStatement));
 		when(preparedStatement.bind("Walter")).thenReturn(boundStatement);
 		when(session.executeAsync(boundStatement)).thenReturn(new TestResultSetFuture(resultSet));
@@ -714,7 +729,7 @@ public class AsyncCqlTemplateUnitTests {
 	@Test // DATACASS-292
 	void queryForObjectPreparedStatementWithTypeShouldReturnRecord() {
 
-		when(session.prepareAsync("SELECT * FROM user WHERE username = ?"))
+		when(session.prepareAsync(any(SimpleStatement.class)))
 				.thenReturn(new TestPreparedStatementFuture(preparedStatement));
 		when(preparedStatement.bind("Walter")).thenReturn(boundStatement);
 		when(session.executeAsync(boundStatement)).thenReturn(new TestResultSetFuture(resultSet));
@@ -731,7 +746,7 @@ public class AsyncCqlTemplateUnitTests {
 	@Test // DATACASS-292
 	void queryForListPreparedStatementWithTypeShouldReturnRecord() {
 
-		when(session.prepareAsync("SELECT * FROM user WHERE username = ?"))
+		when(session.prepareAsync(any(SimpleStatement.class)))
 				.thenReturn(new TestPreparedStatementFuture(preparedStatement));
 		when(preparedStatement.bind("Walter")).thenReturn(boundStatement);
 		when(session.executeAsync(boundStatement)).thenReturn(new TestResultSetFuture(resultSet));
@@ -749,7 +764,7 @@ public class AsyncCqlTemplateUnitTests {
 	@Test // DATACASS-292
 	void updatePreparedStatementShouldReturnApplied() {
 
-		when(session.prepareAsync("UPDATE user SET username = ?"))
+		when(session.prepareAsync(any(SimpleStatement.class)))
 				.thenReturn(new TestPreparedStatementFuture(preparedStatement));
 		when(preparedStatement.bind("Walter")).thenReturn(boundStatement);
 		when(session.executeAsync(boundStatement)).thenReturn(new TestResultSetFuture(resultSet));
@@ -761,10 +776,11 @@ public class AsyncCqlTemplateUnitTests {
 	}
 
 	private void doTestStrings(Consumer<AsyncCqlTemplate> cqlTemplateConsumer) {
-		doTestStrings(null, null, cqlTemplateConsumer);
+		doTestStrings(null, null, null, cqlTemplateConsumer);
 	}
 
 	private void doTestStrings(@Nullable Integer fetchSize, @Nullable ConsistencyLevel consistencyLevel,
+			@Nullable CqlIdentifier keyspace,
 			Consumer<AsyncCqlTemplate> cqlTemplateConsumer) {
 
 		String[] results = { "Walter", "Hank", " Jesse" };
@@ -773,6 +789,8 @@ public class AsyncCqlTemplateUnitTests {
 		when(this.resultSet.currentPage()).thenReturn(Arrays.asList(row, row, row));
 		when(this.row.getString(0)).thenReturn(results[0], results[1], results[2]);
 		when(this.session.prepareAsync(anyString())).thenReturn(new TestPreparedStatementFuture(this.preparedStatement));
+		when(this.session.prepareAsync(any(SimpleStatement.class)))
+				.thenReturn(new TestPreparedStatementFuture(this.preparedStatement));
 
 		AsyncCqlTemplate template = new AsyncCqlTemplate();
 		template.setSession(this.session);
@@ -780,10 +798,15 @@ public class AsyncCqlTemplateUnitTests {
 		if (fetchSize != null) {
 			template.setFetchSize(fetchSize);
 		}
+
 		if (consistencyLevel != null) {
 			template.setConsistencyLevel(consistencyLevel);
 		}
 
+		if (keyspace != null) {
+			template.setKeyspace(keyspace);
+		}
+
 		cqlTemplateConsumer.accept(template);
 
 		ArgumentCaptor<Statement> statementArgumentCaptor = ArgumentCaptor.forClass(Statement.class);

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

@@ -20,7 +20,6 @@ import static org.assertj.core.api.Assertions.*;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
-
 import org.mockito.Mock;
 import org.mockito.junit.jupiter.MockitoExtension;
 
@@ -33,6 +32,7 @@ import com.datastax.oss.driver.api.core.CqlSession;
  *
  * @author John Blum
  * @author Mark Paluch
+ * @author Tomasz Lelek
  */
 @ExtendWith(MockitoExtension.class)
 class CassandraAccessorUnitTests {
@@ -107,8 +107,10 @@ class CassandraAccessorUnitTests {
 
 	@Test // DATACASS-767
 	void setAndGetKeyspace() {
+
 		CqlIdentifier keyspace = CqlIdentifier.fromCql("ks1");
 		cassandraAccessor.setKeyspace(keyspace);
+
 		assertThat(cassandraAccessor.getKeyspace()).isEqualTo(keyspace);
 	}
 

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

@@ -28,12 +28,12 @@ import java.util.function.Consumer;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
-
 import org.mockito.ArgumentCaptor;
 import org.mockito.Mock;
 import org.mockito.junit.jupiter.MockitoExtension;
 import org.mockito.junit.jupiter.MockitoSettings;
 import org.mockito.quality.Strictness;
+
 import org.springframework.dao.EmptyResultDataAccessException;
 import org.springframework.dao.IncorrectResultSizeDataAccessException;
 import org.springframework.data.cassandra.CassandraConnectionFailureException;
@@ -459,6 +459,23 @@ class CqlTemplateUnitTests {
 		});
 	}
 
+	@Test // DATACASS-767
+	void executePreparedStatementShouldApplyKeyspace() {
+
+		when(preparedStatement.bind("A")).thenReturn(boundStatement);
+		when(boundStatement.getKeyspace()).thenReturn(CqlIdentifier.fromCql("ks1"));
+
+		doTestStrings(null, null, null, null, CqlIdentifier.fromCql("ks1"), cqlTemplate -> {
+			cqlTemplate.execute("SELECT * from USERS", (session, ps) -> session.execute(ps.bind("A")));
+		});
+
+		ArgumentCaptor<SimpleStatement> captor = ArgumentCaptor.forClass(SimpleStatement.class);
+		verify(session).prepare(captor.capture());
+
+		SimpleStatement statement = captor.getValue();
+		assertThat(statement.getKeyspace()).isEqualTo(CqlIdentifier.fromCql("ks1"));
+	}
+
 	@Test // DATACASS-292
 	void executePreparedStatementWithCallbackShouldCallExecution() {
 
@@ -596,7 +613,7 @@ class CqlTemplateUnitTests {
 	@Test // DATACASS-292
 	void queryForObjectPreparedStatementShouldBeEmpty() {
 
-		when(session.prepare("SELECT * FROM user WHERE username = ?")).thenReturn(preparedStatement);
+		when(session.prepare(any(SimpleStatement.class))).thenReturn(preparedStatement);
 		when(preparedStatement.bind("Walter")).thenReturn(boundStatement);
 		when(session.execute(boundStatement)).thenReturn(resultSet);
 		when(resultSet.iterator()).thenReturn(Collections.emptyIterator());
@@ -613,7 +630,7 @@ class CqlTemplateUnitTests {
 	@Test // DATACASS-292
 	void queryForObjectPreparedStatementShouldReturnRecord() {
 
-		when(session.prepare("SELECT * FROM user WHERE username = ?")).thenReturn(preparedStatement);
+		when(session.prepare(any(SimpleStatement.class))).thenReturn(preparedStatement);
 		when(preparedStatement.bind("Walter")).thenReturn(boundStatement);
 		when(session.execute(boundStatement)).thenReturn(resultSet);
 		when(resultSet.iterator()).thenReturn(Collections.singleton(row).iterator());
@@ -625,7 +642,7 @@ class CqlTemplateUnitTests {
 	@Test // DATACASS-292
 	void queryForObjectPreparedStatementShouldFailReturningManyRecords() {
 
-		when(session.prepare("SELECT * FROM user WHERE username = ?")).thenReturn(preparedStatement);
+		when(session.prepare(any(SimpleStatement.class))).thenReturn(preparedStatement);
 		when(preparedStatement.bind("Walter")).thenReturn(boundStatement);
 		when(session.execute(boundStatement)).thenReturn(resultSet);
 		when(resultSet.iterator()).thenReturn(Arrays.asList(row, row).iterator());
@@ -642,7 +659,7 @@ class CqlTemplateUnitTests {
 	@Test // DATACASS-292
 	void queryForObjectPreparedStatementWithTypeShouldReturnRecord() {
 
-		when(session.prepare("SELECT * FROM user WHERE username = ?")).thenReturn(preparedStatement);
+		when(session.prepare(any(SimpleStatement.class))).thenReturn(preparedStatement);
 		when(preparedStatement.bind("Walter")).thenReturn(boundStatement);
 		when(session.execute(boundStatement)).thenReturn(resultSet);
 		when(resultSet.iterator()).thenReturn(Collections.singleton(row).iterator());
@@ -658,7 +675,7 @@ class CqlTemplateUnitTests {
 	@Test // DATACASS-292
 	void queryForListPreparedStatementWithTypeShouldReturnRecord() {
 
-		when(session.prepare("SELECT * FROM user WHERE username = ?")).thenReturn(preparedStatement);
+		when(session.prepare(any(SimpleStatement.class))).thenReturn(preparedStatement);
 		when(preparedStatement.bind("Walter")).thenReturn(boundStatement);
 		when(session.execute(boundStatement)).thenReturn(resultSet);
 		when(resultSet.iterator()).thenReturn(Arrays.asList(row, row).iterator());
@@ -674,7 +691,7 @@ class CqlTemplateUnitTests {
 	@Test // DATACASS-292
 	void updatePreparedStatementShouldReturnApplied() {
 
-		when(session.prepare("UPDATE user SET username = ?")).thenReturn(preparedStatement);
+		when(session.prepare(any(SimpleStatement.class))).thenReturn(preparedStatement);
 		when(preparedStatement.bind("Walter")).thenReturn(boundStatement);
 		when(session.execute(boundStatement)).thenReturn(resultSet);
 		when(resultSet.wasApplied()).thenReturn(true);
@@ -687,27 +704,27 @@ class CqlTemplateUnitTests {
 	@Test // DATACASS-767
 	void executeCqlWithKeyspaceShouldCallExecution() {
 
-		doTestStrings(5, DefaultConsistencyLevel.ONE, null, "foo", cqlTemplate -> {
+		doTestStrings(5, DefaultConsistencyLevel.ONE, null, "foo", CqlIdentifier.fromCql("some_keyspace"), cqlTemplate -> {
 
 			cqlTemplate.execute("SELECT * from USERS");
 
 			verify(session).execute(any(Statement.class));
-		}, CqlIdentifier.fromCql("some_keyspace"));
+		});
 	}
 
 	private void doTestStrings(Consumer<CqlTemplate> cqlTemplateConsumer) {
-		doTestStrings(null, null, null, null, cqlTemplateConsumer, null);
+		doTestStrings(null, null, null, null, null, cqlTemplateConsumer);
 	}
 
 	private void doTestStrings(@Nullable Integer fetchSize, @Nullable ConsistencyLevel consistencyLevel,
 			@Nullable ConsistencyLevel serialConsistencyLevel, @Nullable String executionProfile,
 			Consumer<CqlTemplate> cqlTemplateConsumer) {
-		doTestStrings(fetchSize, consistencyLevel, serialConsistencyLevel, executionProfile, cqlTemplateConsumer, null);
+		doTestStrings(fetchSize, consistencyLevel, serialConsistencyLevel, executionProfile, null, cqlTemplateConsumer);
 	}
 
 	private void doTestStrings(@Nullable Integer fetchSize, @Nullable ConsistencyLevel consistencyLevel,
 			@Nullable ConsistencyLevel serialConsistencyLevel, @Nullable String executionProfile,
-			Consumer<CqlTemplate> cqlTemplateConsumer, @Nullable CqlIdentifier keyspace) {
+			@Nullable CqlIdentifier keyspace, Consumer<CqlTemplate> cqlTemplateConsumer) {
 
 		String[] results = { "Walter", "Hank", " Jesse" };
 
@@ -717,6 +734,7 @@ class CqlTemplateUnitTests {
 
 		when(this.row.getString(0)).thenReturn(results[0], results[1], results[2]);
 		when(this.session.prepare(anyString())).thenReturn(preparedStatement);
+		when(this.session.prepare(any(SimpleStatement.class))).thenReturn(preparedStatement);
 
 		CqlTemplate template = new CqlTemplate();
 		template.setSession(this.session);

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

@@ -35,8 +35,13 @@ class QueryOptionsUnitTests {
 	@Test // DATACASS-202
 	void buildQueryOptions() {
 
-		QueryOptions queryOptions = QueryOptions.builder().consistencyLevel(DefaultConsistencyLevel.ANY)
-				.timeout(Duration.ofSeconds(1)).pageSize(10).tracing(true).keyspace(CqlIdentifier.fromCql("ks1")).build();
+		QueryOptions queryOptions = QueryOptions.builder() //
+				.consistencyLevel(DefaultConsistencyLevel.ANY) //
+				.timeout(Duration.ofSeconds(1)) //
+				.pageSize(10) //
+				.tracing(true) //
+				.keyspace(CqlIdentifier.fromCql("ks1")) //
+				.build();
 
 		assertThat(queryOptions.getClass()).isEqualTo(QueryOptions.class);
 		assertThat(queryOptions.getConsistencyLevel()).isEqualTo(DefaultConsistencyLevel.ANY);
@@ -49,13 +54,17 @@ class QueryOptionsUnitTests {
 	@Test // DATACASS-56
 	void buildQueryOptionsMutate() {
 
-		QueryOptions queryOptions = QueryOptions.builder().consistencyLevel(DefaultConsistencyLevel.ANY)
-				.timeout(Duration.ofSeconds(1)).pageSize(10).tracing(true).keyspace(CqlIdentifier.fromCql("ks1")).build();
+		QueryOptions queryOptions = QueryOptions.builder() //
+				.consistencyLevel(DefaultConsistencyLevel.ANY) //
+				.timeout(Duration.ofSeconds(1)) //
+				.pageSize(10) //
+				.tracing(true) //
+				.keyspace(CqlIdentifier.fromCql("ks1")) //
+				.build();
 
 		QueryOptions mutated = queryOptions.mutate().timeout(Duration.ofSeconds(5)).build();
 
-		assertThat(mutated).isNotNull();
-		assertThat(mutated).isNotSameAs(queryOptions);
+		assertThat(mutated).isNotNull().isNotSameAs(queryOptions);
 		assertThat(mutated.getClass()).isEqualTo(QueryOptions.class);
 		assertThat(mutated.getConsistencyLevel()).isEqualTo(DefaultConsistencyLevel.ANY);
 		assertThat(mutated.getTimeout()).isEqualTo(Duration.ofSeconds(5));

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

@@ -23,7 +23,6 @@ import java.util.concurrent.TimeUnit;
 
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
-
 import org.mockito.Mock;
 import org.mockito.junit.jupiter.MockitoExtension;
 
@@ -99,6 +98,7 @@ class QueryOptionsUtilUnitTests {
 
 	@Test // DATACASS-767
 	void addKeyspaceOptionsOnSimpleStatementShouldAddDriverQueryOptions() {
+
 		when(simpleStatement.setKeyspace(any(CqlIdentifier.class))).thenReturn(simpleStatement);
 
 		QueryOptions queryOptions = QueryOptions.builder() //
@@ -111,6 +111,7 @@ class QueryOptionsUtilUnitTests {
 
 	@Test // DATACASS-767
 	void addKeyspaceOptionsOnBatchStatementShouldAddDriverQueryOptions() {
+
 		when(batchStatement.setKeyspace(any(CqlIdentifier.class))).thenReturn(batchStatement);
 
 		QueryOptions queryOptions = QueryOptions.builder() //
@@ -123,12 +124,12 @@ class QueryOptionsUtilUnitTests {
 
 	@Test // DATACASS-767
 	void addKeyspaceOptionsOnBoundStatementShouldThrowException() {
+
 		QueryOptions queryOptions = QueryOptions.builder() //
 				.keyspace(CqlIdentifier.fromCql("ks1")).build();
 
 		assertThatThrownBy(() -> QueryOptionsUtil.addQueryOptions(boundStatement, queryOptions))
 				.isInstanceOf(IllegalArgumentException.class)
 				.hasMessageContaining("Keyspace cannot be set for a BoundStatement");
-
 	}
 }

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

@@ -28,12 +28,12 @@ import java.util.function.Consumer;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
-
 import org.mockito.ArgumentCaptor;
 import org.mockito.Mock;
 import org.mockito.junit.jupiter.MockitoExtension;
 import org.mockito.junit.jupiter.MockitoSettings;
 import org.mockito.quality.Strictness;
+
 import org.springframework.dao.IncorrectResultSizeDataAccessException;
 import org.springframework.data.cassandra.CassandraConnectionFailureException;
 import org.springframework.data.cassandra.CassandraInvalidQueryException;
@@ -550,6 +550,24 @@ class ReactiveCqlTemplateUnitTests {
 		});
 	}
 
+	@Test // DATACASS-767
+	void executePreparedStatementShouldApplyKeyspace() {
+
+		when(preparedStatement.bind("A")).thenReturn(boundStatement);
+		when(boundStatement.getKeyspace()).thenReturn(CqlIdentifier.fromCql("ks1"));
+
+		doTestStrings(null, null, null, null, CqlIdentifier.fromCql("ks1"), cqlTemplate -> {
+			cqlTemplate.execute("SELECT * from USERS", (session, ps) -> session.execute(ps.bind("A")))
+					.as(StepVerifier::create).expectNextCount(1).verifyComplete();
+		});
+
+		ArgumentCaptor<SimpleStatement> captor = ArgumentCaptor.forClass(SimpleStatement.class);
+		verify(session).prepare(captor.capture());
+
+		SimpleStatement statement = captor.getValue();
+		assertThat(statement.getKeyspace()).isEqualTo(CqlIdentifier.fromCql("ks1"));
+	}
+
 	@Test // DATACASS-335
 	void executePreparedStatementWithCallbackShouldCallExecution() {
 
@@ -566,7 +584,7 @@ class ReactiveCqlTemplateUnitTests {
 	@Test // DATACASS-335
 	void executePreparedStatementCallbackShouldExecuteDeferred() {
 
-		when(session.prepare(anyString())).thenReturn(Mono.just(preparedStatement));
+		when(session.prepare(any(SimpleStatement.class))).thenReturn(Mono.just(preparedStatement));
 		when(preparedStatement.bind()).thenReturn(boundStatement);
 		when(session.execute(boundStatement)).thenReturn(Mono.just(reactiveResultSet));
 
@@ -577,7 +595,7 @@ class ReactiveCqlTemplateUnitTests {
 
 		flux.as(StepVerifier::create).expectNext(reactiveResultSet).verifyComplete();
 
-		verify(session).prepare(anyString());
+		verify(session).prepare(any(SimpleStatement.class));
 		verify(session).execute(boundStatement);
 	}
 
@@ -670,7 +688,7 @@ class ReactiveCqlTemplateUnitTests {
 	@Test // DATACASS-335
 	void queryForObjectPreparedStatementShouldBeEmpty() {
 
-		when(session.prepare("SELECT * FROM user WHERE username = ?")).thenReturn(Mono.just(preparedStatement));
+		when(session.prepare(any(SimpleStatement.class))).thenReturn(Mono.just(preparedStatement));
 		when(preparedStatement.bind("Walter")).thenReturn(boundStatement);
 		when(session.execute(boundStatement)).thenReturn(Mono.just(reactiveResultSet));
 		when(reactiveResultSet.rows()).thenReturn(Flux.empty());
@@ -684,7 +702,7 @@ class ReactiveCqlTemplateUnitTests {
 	@Test // DATACASS-335
 	void queryForObjectPreparedStatementShouldReturnRecord() {
 
-		when(session.prepare("SELECT * FROM user WHERE username = ?")).thenReturn(Mono.just(preparedStatement));
+		when(session.prepare(any(SimpleStatement.class))).thenReturn(Mono.just(preparedStatement));
 		when(preparedStatement.bind("Walter")).thenReturn(boundStatement);
 		when(session.execute(boundStatement)).thenReturn(Mono.just(reactiveResultSet));
 		when(reactiveResultSet.rows()).thenReturn(Flux.just(row));
@@ -698,7 +716,7 @@ class ReactiveCqlTemplateUnitTests {
 	@Test // DATACASS-335
 	void queryForObjectPreparedStatementShouldFailReturningManyRecords() {
 
-		when(session.prepare("SELECT * FROM user WHERE username = ?")).thenReturn(Mono.just(preparedStatement));
+		when(session.prepare(any(SimpleStatement.class))).thenReturn(Mono.just(preparedStatement));
 		when(preparedStatement.bind("Walter")).thenReturn(boundStatement);
 		when(session.execute(boundStatement)).thenReturn(Mono.just(reactiveResultSet));
 		when(reactiveResultSet.rows()).thenReturn(Flux.just(row, row));
@@ -712,7 +730,7 @@ class ReactiveCqlTemplateUnitTests {
 	@Test // DATACASS-335
 	void queryForObjectPreparedStatementWithTypeShouldReturnRecord() {
 
-		when(session.prepare("SELECT * FROM user WHERE username = ?")).thenReturn(Mono.just(preparedStatement));
+		when(session.prepare(any(SimpleStatement.class))).thenReturn(Mono.just(preparedStatement));
 		when(preparedStatement.bind("Walter")).thenReturn(boundStatement);
 		when(session.execute(boundStatement)).thenReturn(Mono.just(reactiveResultSet));
 		when(reactiveResultSet.rows()).thenReturn(Flux.just(row));
@@ -728,7 +746,7 @@ class ReactiveCqlTemplateUnitTests {
 	@Test // DATACASS-335
 	void queryForFluxPreparedStatementWithTypeShouldReturnRecord() {
 
-		when(session.prepare("SELECT * FROM user WHERE username = ?")).thenReturn(Mono.just(preparedStatement));
+		when(session.prepare(any(SimpleStatement.class))).thenReturn(Mono.just(preparedStatement));
 		when(preparedStatement.bind("Walter")).thenReturn(boundStatement);
 		when(session.execute(boundStatement)).thenReturn(Mono.just(reactiveResultSet));
 		when(reactiveResultSet.rows()).thenReturn(Flux.just(row, row));
@@ -744,7 +762,7 @@ class ReactiveCqlTemplateUnitTests {
 	@Test // DATACASS-335
 	void queryForRowsPreparedStatementReturnRows() {
 
-		when(session.prepare("SELECT * FROM user WHERE username = ?")).thenReturn(Mono.just(preparedStatement));
+		when(session.prepare(any(SimpleStatement.class))).thenReturn(Mono.just(preparedStatement));
 		when(preparedStatement.bind("Walter")).thenReturn(boundStatement);
 		when(session.execute(boundStatement)).thenReturn(Mono.just(reactiveResultSet));
 		when(reactiveResultSet.rows()).thenReturn(Flux.just(row, row));
@@ -757,7 +775,7 @@ class ReactiveCqlTemplateUnitTests {
 	@Test // DATACASS-335
 	void updatePreparedStatementShouldReturnApplied() {
 
-		when(session.prepare("UPDATE user SET username = ?")).thenReturn(Mono.just(preparedStatement));
+		when(session.prepare(any(SimpleStatement.class))).thenReturn(Mono.just(preparedStatement));
 		when(preparedStatement.bind("Walter")).thenReturn(boundStatement);
 		when(session.execute(boundStatement)).thenReturn(Mono.just(reactiveResultSet));
 		when(reactiveResultSet.wasApplied()).thenReturn(true);
@@ -770,7 +788,7 @@ class ReactiveCqlTemplateUnitTests {
 	@Test // DATACASS-335
 	void updatePreparedStatementArgsPublisherShouldReturnApplied() {
 
-		when(session.prepare("UPDATE user SET username = ?")).thenReturn(Mono.just(preparedStatement));
+		when(session.prepare(any(SimpleStatement.class))).thenReturn(Mono.just(preparedStatement));
 		when(preparedStatement.bind("Walter")).thenReturn(boundStatement);
 		when(preparedStatement.bind("Hank")).thenReturn(boundStatement);
 		when(session.execute(boundStatement)).thenReturn(Mono.just(reactiveResultSet));
@@ -786,18 +804,18 @@ class ReactiveCqlTemplateUnitTests {
 	}
 
 	private void doTestStrings(Consumer<ReactiveCqlTemplate> cqlTemplateConsumer) {
-		doTestStrings(null, null, null, null, cqlTemplateConsumer, null);
+		doTestStrings(null, null, null, null, null, cqlTemplateConsumer);
 	}
 
 	private void doTestStrings(@Nullable Integer fetchSize, @Nullable ConsistencyLevel consistencyLevel,
 			@Nullable ConsistencyLevel serialConsistencyLevel, @Nullable String executionProfile,
 			Consumer<ReactiveCqlTemplate> cqlTemplateConsumer) {
-		doTestStrings(fetchSize, consistencyLevel, serialConsistencyLevel, executionProfile, cqlTemplateConsumer, null);
+		doTestStrings(fetchSize, consistencyLevel, serialConsistencyLevel, executionProfile, null, cqlTemplateConsumer);
 	}
 
 	private void doTestStrings(@Nullable Integer fetchSize, @Nullable ConsistencyLevel consistencyLevel,
 			@Nullable ConsistencyLevel serialConsistencyLevel, @Nullable String executionProfile,
-			Consumer<ReactiveCqlTemplate> cqlTemplateConsumer, @Nullable CqlIdentifier keyspace) {
+			@Nullable CqlIdentifier keyspace, Consumer<ReactiveCqlTemplate> cqlTemplateConsumer) {
 
 		String[] results = { "Walter", "Hank", " Jesse" };
 
@@ -806,6 +824,7 @@ class ReactiveCqlTemplateUnitTests {
 
 		when(this.row.getString(0)).thenReturn(results[0], results[1], results[2]);
 		when(this.session.prepare(anyString())).thenReturn(Mono.just(this.preparedStatement));
+		when(this.session.prepare(any(SimpleStatement.class))).thenReturn(Mono.just(this.preparedStatement));
 
 		ReactiveCqlTemplate template = new ReactiveCqlTemplate();
 		template.setSessionFactory(this.sessionFactory);
@@ -852,6 +871,7 @@ class ReactiveCqlTemplateUnitTests {
 		if (executionProfile != null) {
 			assertThat(statement.getExecutionProfileName()).isEqualTo(executionProfile);
 		}
+
 		if (keyspace != null) {
 			assertThat(statement.getKeyspace()).isEqualTo(keyspace);
 		}

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

@@ -24,6 +24,7 @@ import java.time.ZoneOffset;
 
 import org.junit.jupiter.api.Test;
 
+import com.datastax.oss.driver.api.core.CqlIdentifier;
 import com.datastax.oss.driver.api.core.DefaultConsistencyLevel;
 
 /**
@@ -33,17 +34,17 @@ import com.datastax.oss.driver.api.core.DefaultConsistencyLevel;
  */
 class WriteOptionsUnitTests {
 
-	@Test // DATACASS-202
+	@Test // DATACASS-202, DATACASS-767
 	void buildWriteOptions() {
 
-		WriteOptions writeOptions = WriteOptions.builder()
-				.consistencyLevel(DefaultConsistencyLevel.ANY)
-				.ttl(123)
-				.timestamp(1519000753)
-				.readTimeout(1)
-				.pageSize(10)
-				.withTracing()
-				.build();
+		WriteOptions writeOptions = WriteOptions.builder() //
+				.consistencyLevel(DefaultConsistencyLevel.ANY) //
+				.ttl(123) //
+				.timestamp(1519000753) //
+				.readTimeout(1) //
+				.pageSize(10) //
+				.withTracing() //
+				.keyspace(CqlIdentifier.fromCql("my_keyspace")).build();
 
 		assertThat(writeOptions.getTtl()).isEqualTo(Duration.ofSeconds(123));
 		assertThat(writeOptions.getTimestamp()).isEqualTo(1519000753);
@@ -51,6 +52,7 @@ class WriteOptionsUnitTests {
 		assertThat(writeOptions.getTimeout()).isEqualTo(Duration.ofMillis(1));
 		assertThat(writeOptions.getPageSize()).isEqualTo(10);
 		assertThat(writeOptions.getTracing()).isTrue();
+		assertThat(writeOptions.getKeyspace()).isEqualTo(CqlIdentifier.fromCql("my_keyspace"));
 	}
 
 	@Test // DATACASS-202
@@ -63,18 +65,17 @@ class WriteOptionsUnitTests {
 		assertThat(writeOptions.getTracing()).isNull();
 	}
 
-
 	@Test // DATACASS-56
 	void buildWriteOptionsMutate() {
 		Instant now = LocalDateTime.now().toInstant(ZoneOffset.UTC);
 
-		WriteOptions writeOptions = WriteOptions.builder()
-				.consistencyLevel(DefaultConsistencyLevel.ANY)
-				.ttl(123)
-				.timestamp(now)
-				.readTimeout(1)
-				.pageSize(10)
-				.withTracing()
+		WriteOptions writeOptions = WriteOptions.builder() //
+				.consistencyLevel(DefaultConsistencyLevel.ANY) //
+				.ttl(123) //
+				.timestamp(now) //
+				.readTimeout(1) //
+				.pageSize(10) //
+				.withTracing() //
 				.build();
 
 		WriteOptions mutated = writeOptions.mutate().timeout(Duration.ofMillis(100)).build();

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/test/util/CassandraDelegate.java

@@ -28,11 +28,13 @@ import org.springframework.data.cassandra.support.CqlDataSet;
 import org.springframework.util.Assert;
 import org.springframework.util.SocketUtils;
 import org.springframework.util.StringUtils;
+
 import org.testcontainers.containers.CassandraContainer;
 
 import com.datastax.oss.driver.api.core.CqlIdentifier;
 import com.datastax.oss.driver.api.core.CqlSession;
 import com.datastax.oss.driver.api.core.CqlSessionBuilder;
+import com.datastax.oss.driver.api.core.Version;
 import com.datastax.oss.driver.api.core.config.DefaultDriverOption;
 import com.datastax.oss.driver.api.core.config.DriverConfigLoader;
 import com.datastax.oss.driver.api.core.metadata.Node;
@@ -302,8 +304,9 @@ class CassandraDelegate {
 		CqlSessionBuilder builder = CqlSession.builder().addContactPoint(InetSocketAddress.createUnresolved(host, port))
 				.withLocalDatacenter("datacenter1");
 
-		CqlSession cqlSession = builder.build();
-		if (cassandraVersionGreaterThanOrEqualTo4(cqlSession)) {
+		Version cassandraVersion = getCassandraVersion(builder);
+
+		if (cassandraVersion.getMajor() >= 4) {
 			return builder.withConfigLoader(
 					DriverConfigLoader.programmaticBuilder().withString(DefaultDriverOption.PROTOCOL_VERSION, "V5").build());
 		} else {
@@ -311,9 +314,17 @@ class CassandraDelegate {
 		}
 	}
 
-	private boolean cassandraVersionGreaterThanOrEqualTo4(CqlSession cqlSession) {
-		return cqlSession.getMetadata().getNodes().values().stream().map(Node::getCassandraVersion)
-				.allMatch(v -> v != null && v.getMajor() >= 4);
+	private Version getCassandraVersion(CqlSessionBuilder builder) {
+
+		try (CqlSession cqlSession = builder.build()) {
+
+			return cqlSession.getMetadata().getNodes() //
+					.values() //
+					.stream() //
+					.map(Node::getCassandraVersion) //
+					.findFirst() //
+					.orElseThrow(() -> new IllegalStateException("Cannot determine Cassandra version"));
+		}
 	}
 
 	private String resolveHost() {

src/main/asciidoc/new-features.adoc

@@ -8,6 +8,7 @@ This chapter summarizes changes and new features for each release.
 
 * <<cassandra.auditing,Reactive auditing>> enabled through `@EnableReactiveCassandraAuditing`. `@EnableCassandraAuditing` no longer registers `ReactiveAuditingEntityCallback`.
 * Reactive SpEL support in `@Query` query methods.
+* Configuration of the keyspace per `Statement` through `CqlTemplate` and `QueryOptions`.
 
 [[new-features.3-0-0]]
 == What's new in Spring Data for Apache Cassandra 3.0