DATACASS-238 - Add Cluster configuration options.

We now support configuration of the cluster name, AddressTranslator, MaxSchemaAgreementWaitSeconds, SpeculativeExecutionPolicy and TimestampGenerator in the Cassandra Cluster factory. The cluster name is derived from the bean name, if not configured otherwise. Related tickets: DATACASS-120, DATACASS-316, DATACASS-317, DATACASS-319, DATACASS-320. Original pull request: #79. Related pull request: #80.
2016-07-25 00:26:49 -07:00
parent d0610b2812
commit dfc7e3a716
11 changed files with 778 additions and 198 deletions
--- a/spring-cql/src/main/java/org/springframework/cassandra/config/CassandraCqlClusterFactoryBean.java
+++ b/spring-cql/src/main/java/org/springframework/cassandra/config/CassandraCqlClusterFactoryBean.java
@@ -22,6 +22,7 @@ import java.util.Set;

 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.BeanNameAware;
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.beans.factory.FactoryBean;
 import org.springframework.beans.factory.InitializingBean;
@@ -49,12 +50,15 @@ import com.datastax.driver.core.QueryOptions;
 import com.datastax.driver.core.SSLOptions;
 import com.datastax.driver.core.Session;
 import com.datastax.driver.core.SocketOptions;
+import com.datastax.driver.core.TimestampGenerator;
+import com.datastax.driver.core.policies.AddressTranslator;
 import com.datastax.driver.core.policies.LoadBalancingPolicy;
 import com.datastax.driver.core.policies.ReconnectionPolicy;
 import com.datastax.driver.core.policies.RetryPolicy;
+import com.datastax.driver.core.policies.SpeculativeExecutionPolicy;

 /**
- * Convenient {@link org.springframework.beans.factory.FactoryBean} for configuring a Cassandra {@link Cluster}.
+ * {@link org.springframework.beans.factory.FactoryBean} for configuring a Cassandra {@link Cluster}.
 *
 * @author Alex Shvid
 * @author Matthew T. Adams
@@ -69,164 +73,171 @@ import com.datastax.driver.core.policies.RetryPolicy;
 * @see org.springframework.beans.factory.FactoryBean
 * @see com.datastax.driver.core.Cluster
 */
-public class CassandraCqlClusterFactoryBean
-		implements FactoryBean<Cluster>, InitializingBean, DisposableBean, PersistenceExceptionTranslator {
+@SuppressWarnings("unused")
+public class CassandraCqlClusterFactoryBean implements FactoryBean<Cluster>, InitializingBean, DisposableBean,
+		BeanNameAware, PersistenceExceptionTranslator {
+
+	public static final boolean DEFAULT_JMX_REPORTING_ENABLED = true;
+	public static final boolean DEFAULT_METRICS_ENABLED = true;
+	public static final boolean DEFAULT_SSL_ENABLED = false;
+
+	public static final int DEFAULT_MAX_SCHEMA_AGREEMENT_WAIT_SECONDS = 10;
+	public static final int DEFAULT_PORT = 9042;

 	public static final String DEFAULT_CONTACT_POINTS = "localhost";
-	public static final boolean DEFAULT_METRICS_ENABLED = true;
-	public static final boolean DEFAULT_JMX_REPORTING_ENABLED = true;
-	public static final boolean DEFAULT_SSL_ENABLED = false;
-	public static final int DEFAULT_PORT = 9042;

 	protected static final Logger log = LoggerFactory.getLogger(CassandraCqlClusterFactoryBean.class);

+	private boolean jmxReportingEnabled = DEFAULT_JMX_REPORTING_ENABLED;
+	private boolean metricsEnabled = DEFAULT_METRICS_ENABLED;
+	private boolean sslEnabled = DEFAULT_SSL_ENABLED;
+
+	private int maxSchemaAgreementWaitSeconds = DEFAULT_MAX_SCHEMA_AGREEMENT_WAIT_SECONDS;
+	private int port = DEFAULT_PORT;
+
+	private AddressTranslator addressTranslator;
+
+	private AuthProvider authProvider;
+
 	private Cluster cluster;

-	/*
-	 * Attributes needed for cluster builder
-	 */
-	private String contactPoints = DEFAULT_CONTACT_POINTS;
-	private int port = CassandraCqlClusterFactoryBean.DEFAULT_PORT;
-
-	// Protocol options
 	private CompressionType compressionType;
-	private SSLOptions sslOptions;
-	private boolean sslEnabled = DEFAULT_SSL_ENABLED;
-	private AuthProvider authProvider;
-	private String username;
-	private String password;
-	private NettyOptions nettyOptions;
-	private ProtocolVersion protocolVersion;
-
-	// Policies
-	private LoadBalancingPolicy loadBalancingPolicy;
-	private ReconnectionPolicy reconnectionPolicy;
-	private RetryPolicy retryPolicy;
-
-	private PoolingOptions poolingOptions;
-	private QueryOptions queryOptions;
-	private SocketOptions socketOptions;
-
-	private boolean metricsEnabled = DEFAULT_METRICS_ENABLED;
-	private boolean jmxReportingEnabled = DEFAULT_JMX_REPORTING_ENABLED;

 	private Host.StateListener hostStateListener;
+
 	private LatencyTracker latencyTracker;

-	// Startup and shutdown actions
-	private Set<KeyspaceActionSpecification<?>> keyspaceSpecifications = new HashSet<KeyspaceActionSpecification<?>>();
 	private List<CreateKeyspaceSpecification> keyspaceCreations = new ArrayList<CreateKeyspaceSpecification>();
 	private List<DropKeyspaceSpecification> keyspaceDrops = new ArrayList<DropKeyspaceSpecification>();
 	private List<String> startupScripts = new ArrayList<String>();
 	private List<String> shutdownScripts = new ArrayList<String>();

+	private LoadBalancingPolicy loadBalancingPolicy;
+
+	private NettyOptions nettyOptions;
+
 	private final PersistenceExceptionTranslator exceptionTranslator = new CassandraExceptionTranslator();

-	/* (non-Javadoc)
-	 * @see org.springframework.beans.factory.FactoryBean#getObject()
-	 */
-	@Override
-	public Cluster getObject() {
-		return cluster;
-	}
+	private PoolingOptions poolingOptions;

-	/* (non-Javadoc)
-	 * @see org.springframework.beans.factory.FactoryBean#getObjectType()
-	 */
-	@Override
-	public Class<? extends Cluster> getObjectType() {
-		return Cluster.class;
-	}
+	private ProtocolVersion protocolVersion;

-	/* (non-Javadoc)
-	 * @see org.springframework.beans.factory.FactoryBean#isSingleton()
-	 */
-	@Override
-	public boolean isSingleton() {
-		return true;
-	}
+	private QueryOptions queryOptions;

-	/* (non-Javadoc)
-	 * @see org.springframework.dao.support.PersistenceExceptionTranslator#translateExceptionIfPossible(java.lang.RuntimeException)
-	 */
-	@Override
-	public DataAccessException translateExceptionIfPossible(RuntimeException ex) {
-		return exceptionTranslator.translateExceptionIfPossible(ex);
-	}
+	private ReconnectionPolicy reconnectionPolicy;

-	/* (non-Javadoc)
+	private RetryPolicy retryPolicy;
+
+	private Set<KeyspaceActionSpecification<?>> keyspaceSpecifications = new HashSet<KeyspaceActionSpecification<?>>();
+
+	private SpeculativeExecutionPolicy speculativeExecutionPolicy;
+
+	private SocketOptions socketOptions;
+
+	private SSLOptions sslOptions;
+
+	private String beanName;
+	private String clusterName;
+	private String contactPoints = DEFAULT_CONTACT_POINTS;
+	private String password;
+	private String username;
+
+	private TimestampGenerator timestampGenerator;
+
+	/*
+	 * (non-Javadoc)
 	 * @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet()
 	 */
 	@Override
 	public void afterPropertiesSet() throws Exception {

 		if (!StringUtils.hasText(contactPoints)) {
-			throw new IllegalArgumentException("at least one server is required");
+			throw new IllegalArgumentException("At least one server is required");
 		}

-		Cluster.Builder builder = Cluster.builder();
+		Cluster.Builder clusterBuilder = newClusterBuilder();

-		builder.addContactPoints(StringUtils.commaDelimitedListToStringArray(contactPoints)).withPort(port);
+		clusterBuilder.addContactPoints(StringUtils.commaDelimitedListToStringArray(contactPoints)).withPort(port);

 		if (compressionType != null) {
-			builder.withCompression(convertCompressionType(compressionType));
+			clusterBuilder.withCompression(convertCompressionType(compressionType));
 		}

 		if (poolingOptions != null) {
-			builder.withPoolingOptions(poolingOptions);
+			clusterBuilder.withPoolingOptions(poolingOptions);
 		}

 		if (socketOptions != null) {
-			builder.withSocketOptions(socketOptions);
+			clusterBuilder.withSocketOptions(socketOptions);
 		}

 		if (queryOptions != null) {
-			builder.withQueryOptions(queryOptions);
+			clusterBuilder.withQueryOptions(queryOptions);
 		}

 		if (authProvider != null) {
-			builder.withAuthProvider(authProvider);
+			clusterBuilder.withAuthProvider(authProvider);
 		} else if (username != null) {
-			builder.withCredentials(username, password);
+			clusterBuilder.withCredentials(username, password);
 		}

 		if (nettyOptions != null) {
-			builder.withNettyOptions(nettyOptions);
+			clusterBuilder.withNettyOptions(nettyOptions);
 		}

 		if (loadBalancingPolicy != null) {
-			builder.withLoadBalancingPolicy(loadBalancingPolicy);
+			clusterBuilder.withLoadBalancingPolicy(loadBalancingPolicy);
 		}

 		if (reconnectionPolicy != null) {
-			builder.withReconnectionPolicy(reconnectionPolicy);
+			clusterBuilder.withReconnectionPolicy(reconnectionPolicy);
 		}

 		if (retryPolicy != null) {
-			builder.withRetryPolicy(retryPolicy);
+			clusterBuilder.withRetryPolicy(retryPolicy);
 		}

 		if (!metricsEnabled) {
-			builder.withoutMetrics();
+			clusterBuilder.withoutMetrics();
 		}

 		if (!jmxReportingEnabled) {
-			builder.withoutJMXReporting();
+			clusterBuilder.withoutJMXReporting();
 		}

 		if (sslEnabled) {
 			if (sslOptions == null) {
-				builder.withSSL();
+				clusterBuilder.withSSL();
 			} else {
-				builder.withSSL(sslOptions);
+				clusterBuilder.withSSL(sslOptions);
 			}
 		}

 		if (protocolVersion != null) {
-			builder.withProtocolVersion(protocolVersion);
+			clusterBuilder.withProtocolVersion(protocolVersion);
 		}

-		cluster = builder.build();
+		if (addressTranslator != null) {
+			clusterBuilder.withAddressTranslator(addressTranslator);
+		}
+
+		String clusterName = resolveClusterName();
+
+		if (StringUtils.hasText(clusterName)) {
+			clusterBuilder.withClusterName(clusterName);
+		}
+
+		clusterBuilder.withMaxSchemaAgreementWaitSeconds(maxSchemaAgreementWaitSeconds);
+
+		if (speculativeExecutionPolicy != null) {
+			clusterBuilder.withSpeculativeExecutionPolicy(speculativeExecutionPolicy);
+		}
+
+		if (timestampGenerator != null) {
+			clusterBuilder.withTimestampGenerator(timestampGenerator);
+		}
+
+		cluster = clusterBuilder.build();

 		if (hostStateListener != null) {
 			cluster.register(hostStateListener);
@@ -241,45 +252,94 @@ public class CassandraCqlClusterFactoryBean
 		executeSpecsAndScripts(keyspaceCreations, startupScripts);
 	}

-	/* (non-Javadoc)
+	/*
+	 * (non-Javadoc)
+	 * @see com.datastax.driver.core.Cluster#builder()
+	 */
+	Cluster.Builder newClusterBuilder() {
+		return Cluster.builder();
+	}
+
+	String resolveClusterName() {
+		return (StringUtils.hasText(clusterName) ? clusterName : beanName);
+	}
+
+	/*
+	 * (non-Javadoc)
 	 * @see org.springframework.beans.factory.DisposableBean#destroy()
 	 */
 	@Override
 	public void destroy() throws Exception {
-
 		executeSpecsAndScripts(keyspaceDrops, shutdownScripts);
 		cluster.close();
 	}

+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.beans.factory.FactoryBean#getObject()
+	 */
+	@Override
+	public Cluster getObject() {
+		return cluster;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.beans.factory.FactoryBean#getObjectType()
+	 */
+	@Override
+	public Class<? extends Cluster> getObjectType() {
+		return (cluster != null ? cluster.getClass() : Cluster.class);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.beans.factory.FactoryBean#isSingleton()
+	 */
+	@Override
+	public boolean isSingleton() {
+		return true;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.dao.support.PersistenceExceptionTranslator#translateExceptionIfPossible(java.lang.RuntimeException)
+	 */
+	@Override
+	public DataAccessException translateExceptionIfPossible(RuntimeException ex) {
+		return exceptionTranslator.translateExceptionIfPossible(ex);
+	}
+
 	/**
 	 * Examines the contents of all the KeyspaceSpecificationFactoryBeans and generates the proper KeyspaceSpecification
 	 * from them.
 	 */
 	private void generateSpecificationsFromFactoryBeans() {

-		for (KeyspaceActionSpecification<?> spec : keyspaceSpecifications) {
+		for (KeyspaceActionSpecification<?> keyspaceSpecification : keyspaceSpecifications) {

-			if (spec instanceof CreateKeyspaceSpecification) {
-				keyspaceCreations.add((CreateKeyspaceSpecification) spec);
+			if (keyspaceSpecification instanceof CreateKeyspaceSpecification) {
+				keyspaceCreations.add((CreateKeyspaceSpecification) keyspaceSpecification);
 			}
-			if (spec instanceof DropKeyspaceSpecification) {
-				keyspaceDrops.add((DropKeyspaceSpecification) spec);
+
+			if (keyspaceSpecification instanceof DropKeyspaceSpecification) {
+				keyspaceDrops.add((DropKeyspaceSpecification) keyspaceSpecification);
 			}
 		}
 	}

-	protected void executeSpecsAndScripts(List<? extends KeyspaceActionSpecification<?>> specifications, List<String> scripts) {
+	protected void executeSpecsAndScripts(List<? extends KeyspaceActionSpecification<?>> kepspaceActionSpecifications,
+			List<String> scripts) {

-		if (!CollectionUtils.isEmpty(specifications) || !CollectionUtils.isEmpty(scripts)) {
+		if (!CollectionUtils.isEmpty(kepspaceActionSpecifications) || !CollectionUtils.isEmpty(scripts)) {

 			Session session = cluster.connect();

 			try {
-
 				CqlTemplate template = new CqlTemplate(session);

-				for (KeyspaceActionSpecification<?> spec : specifications) {
-					template.execute(toCql(spec));
+				for (KeyspaceActionSpecification<?> keyspaceActionSpecification : kepspaceActionSpecifications) {
+					template.execute(toCql(keyspaceActionSpecification));
 				}

 				for (String script : scripts) {
@@ -293,18 +353,25 @@ public class CassandraCqlClusterFactoryBean
 		}
 	}

-	private String toCql(KeyspaceActionSpecification<?> spec) {
+	private String toCql(KeyspaceActionSpecification<?> keyspaceActionSpecification) {

-		if(spec instanceof CreateKeyspaceSpecification) {
-			return new CreateKeyspaceCqlGenerator((CreateKeyspaceSpecification) spec).toCql();
-		}
+		return (keyspaceActionSpecification instanceof CreateKeyspaceSpecification
+			? new CreateKeyspaceCqlGenerator((CreateKeyspaceSpecification) keyspaceActionSpecification).toCql()
+			: new DropKeyspaceCqlGenerator((DropKeyspaceSpecification) keyspaceActionSpecification).toCql());
+	}

-		return new DropKeyspaceCqlGenerator((DropKeyspaceSpecification) spec).toCql();
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.beans.factory.BeanNameAware#setBeanName(String)
+	 */
+	@Override
+	public void setBeanName(String beanName) {
+		this.beanName = beanName;
 	}

 	/**
-	 * Set a comma-delimited string of the contact points (hosts) to connect to. Default is {@code localhost}, see
-	 * {@link #DEFAULT_CONTACT_POINTS}.
+	 * Set a comma-delimited string of the contact points (hosts) to connect to. Default is {@code localhost};
+	 * see {@link #DEFAULT_CONTACT_POINTS}.
 	 */
 	public void setContactPoints(String contactPoints) {
 		this.contactPoints = contactPoints;
@@ -364,7 +431,6 @@ public class CassandraCqlClusterFactoryBean
 	/**
 	 * Set the {@link NettyOptions} used by a client to customize the driver's underlying Netty layer.
 	 *
-	 * @param nettyOptions
 	 * @since 1.5
 	 */
 	public void setNettyOptions(NettyOptions nettyOptions) {
@@ -458,13 +524,6 @@ public class CassandraCqlClusterFactoryBean
 		return shutdownScripts;
 	}

-	/**
-	 * @return Returns the keyspaceSpecifications.
-	 */
-	public Set<KeyspaceActionSpecification<?>> getKeyspaceSpecifications() {
-		return keyspaceSpecifications;
-	}
-
 	/**
 	 * @param keyspaceSpecifications The keyspaceSpecifications to set.
 	 */
@@ -472,6 +531,13 @@ public class CassandraCqlClusterFactoryBean
 		this.keyspaceSpecifications = keyspaceSpecifications;
 	}

+	/**
+	 * @return Returns the keyspaceSpecifications.
+	 */
+	public Set<KeyspaceActionSpecification<?>> getKeyspaceSpecifications() {
+		return keyspaceSpecifications;
+	}
+
 	/**
 	 * Set the username to use with {@link com.datastax.driver.core.PlainTextAuthProvider}.
 	 *
@@ -529,6 +595,62 @@ public class CassandraCqlClusterFactoryBean
 		this.latencyTracker = latencyTracker;
 	}

+	/**
+	 * Configures the address translator used by the new cluster to translate IP addresses received
+	 * from Cassandra nodes into locally query-able addresses.
+	 *
+	 * @param addressTranslator {@link AddressTranslator} used by the new cluster.
+	 * @see com.datastax.driver.core.Cluster.Builder#withAddressTranslator(AddressTranslator)
+	 * @see com.datastax.driver.core.policies.AddressTranslator
+	 */
+	public void setAddressTranslator(AddressTranslator addressTranslator) {
+		this.addressTranslator = addressTranslator;
+	}
+
+	/**
+	 * An optional name for the create cluster.
+	 * \
+	 * @param clusterName optional name for the cluster.
+	 * @see com.datastax.driver.core.Cluster.Builder#withClusterName(String)
+	 */
+	public void setClusterName(String clusterName) {
+		this.clusterName = clusterName;
+	}
+
+	/**
+	 * Sets the maximum time to wait for schema agreement before returning from a DDL query.  The timeout is used
+	 * to wait for all currently up hosts in the cluster to agree on the schema.
+	 *
+	 * @param seconds max schema agreement wait in seconds.
+	 * @see com.datastax.driver.core.Cluster.Builder#withMaxSchemaAgreementWaitSeconds(int)
+	 */
+	public void setMaxSchemaAgreementWaitSeconds(int seconds) {
+		this.maxSchemaAgreementWaitSeconds = seconds;
+	}
+
+	/**
+	 * Configures the speculative execution policy to use for the new cluster.
+	 *
+	 * @param speculativeExecutionPolicy {@link SpeculativeExecutionPolicy} to use with the new cluster.
+	 * @see com.datastax.driver.core.Cluster.Builder#withSpeculativeExecutionPolicy(SpeculativeExecutionPolicy)
+	 * @see com.datastax.driver.core.policies.SpeculativeExecutionPolicy
+	 */
+	public void setSpeculativeExecutionPolicy(SpeculativeExecutionPolicy speculativeExecutionPolicy) {
+		this.speculativeExecutionPolicy = speculativeExecutionPolicy;
+	}
+
+	/**
+	 * Configures the generator that will produce the client-side timestamp sent with each query.
+	 *
+	 * @param timestampGenerator {@link TimestampGenerator} used to produce a client-side timestamp
+	 * sent with each query.
+	 * @see com.datastax.driver.core.Cluster.Builder#withTimestampGenerator(TimestampGenerator)
+	 * @see com.datastax.driver.core.TimestampGenerator
+	 */
+	public void setTimestampGenerator(TimestampGenerator timestampGenerator) {
+		this.timestampGenerator = timestampGenerator;
+	}
+
 	private static Compression convertCompressionType(CompressionType type) {
 		switch (type) {
 			case NONE:
@@ -537,6 +659,6 @@ public class CassandraCqlClusterFactoryBean
 				return Compression.SNAPPY;
 		}

-		throw new IllegalArgumentException("unknown compression type " + type);
+		throw new IllegalArgumentException(String.format("Unknown compression type [%s]", type));
 	}
 }
--- a/spring-cql/src/main/java/org/springframework/cassandra/config/java/AbstractClusterConfiguration.java
+++ b/spring-cql/src/main/java/org/springframework/cassandra/config/java/AbstractClusterConfiguration.java
@@ -31,9 +31,12 @@ import com.datastax.driver.core.PoolingOptions;
 import com.datastax.driver.core.ProtocolVersion;
 import com.datastax.driver.core.QueryOptions;
 import com.datastax.driver.core.SocketOptions;
+import com.datastax.driver.core.TimestampGenerator;
+import com.datastax.driver.core.policies.AddressTranslator;
 import com.datastax.driver.core.policies.LoadBalancingPolicy;
 import com.datastax.driver.core.policies.ReconnectionPolicy;
 import com.datastax.driver.core.policies.RetryPolicy;
+import com.datastax.driver.core.policies.SpeculativeExecutionPolicy;

 /**
 * Base class for Spring Cassandra configuration that can handle creating namespaces, execute arbitrary CQL on startup &
@@ -51,22 +54,24 @@ public abstract class AbstractClusterConfiguration {

 		CassandraCqlClusterFactoryBean bean = new CassandraCqlClusterFactoryBean();

-		bean.setContactPoints(getContactPoints());
-		bean.setPort(getPort());
-
+		bean.setAddressTranslator(getAddressTranslator());
 		bean.setAuthProvider(getAuthProvider());
+		bean.setClusterName(getClusterName());
 		bean.setCompressionType(getCompressionType());
-		bean.setProtocolVersion(getProtocolVersion());
-
+		bean.setContactPoints(getContactPoints());
 		bean.setLoadBalancingPolicy(getLoadBalancingPolicy());
-		bean.setReconnectionPolicy(getReconnectionPolicy());
-		bean.setRetryPolicy(getRetryPolicy());
+		bean.setMaxSchemaAgreementWaitSeconds(getMaxSchemaAgreementWaitSeconds());
 		bean.setMetricsEnabled(getMetricsEnabled());
 		bean.setNettyOptions(getNettyOptions());
-
 		bean.setPoolingOptions(getPoolingOptions());
+		bean.setPort(getPort());
+		bean.setProtocolVersion(getProtocolVersion());
 		bean.setQueryOptions(getQueryOptions());
+		bean.setReconnectionPolicy(getReconnectionPolicy());
+		bean.setRetryPolicy(getRetryPolicy());
+		bean.setSpeculativeExecutionPolicy(getSpeculativeExecutionPolicy());
 		bean.setSocketOptions(getSocketOptions());
+		bean.setTimestampGenerator(getTimestampGenerator());

 		bean.setKeyspaceCreations(getKeyspaceCreations());
 		bean.setKeyspaceDrops(getKeyspaceDrops());
@@ -77,23 +82,13 @@ public abstract class AbstractClusterConfiguration {
 	}

 	/**
-	 * Returns the Cassandra port. Defaults to {@code 9042}
+	 * Returns the {@link AddressTranslator}.
 	 *
-	 * @return the Cassandra port
-	 * @see CassandraCqlClusterFactoryBean#DEFAULT_PORT
+	 * @return the {@link AddressTranslator}; may be {@literal null}.
+	 * @since 1.5
 	 */
-	protected int getPort() {
-		return CassandraCqlClusterFactoryBean.DEFAULT_PORT;
-	}
-
-	/**
-	 * Returns the Cassandra contact points. Defaults to {@code localhost}
-	 *
-	 * @return the Cassandra contact points
-	 * @see CassandraCqlClusterFactoryBean#DEFAULT_CONTACT_POINTS
-	 */
-	protected String getContactPoints() {
-		return CassandraCqlClusterFactoryBean.DEFAULT_CONTACT_POINTS;
+	protected AddressTranslator getAddressTranslator() {
+		return null;
 	}

 	/**
@@ -105,6 +100,16 @@ public abstract class AbstractClusterConfiguration {
 		return null;
 	}

+	/**
+	 * Returns the cluster name.
+	 *
+	 * @return the cluster name; may be {@literal null}.
+	 * @since 1.5
+	 */
+	protected String getClusterName() {
+		return null;
+	}
+
 	/**
 	 * Returns the {@link CompressionType}.
 	 *
@@ -115,12 +120,13 @@ public abstract class AbstractClusterConfiguration {
 	}

 	/**
-	 * Returns the {@link ProtocolVersion}.
+	 * Returns the Cassandra contact points. Defaults to {@code localhost}
 	 *
-	 * @return the {@link ProtocolVersion}, may be {@literal null}.
+	 * @return the Cassandra contact points
+	 * @see CassandraCqlClusterFactoryBean#DEFAULT_CONTACT_POINTS
 	 */
-	protected ProtocolVersion getProtocolVersion() {
-		return null;
+	protected String getContactPoints() {
+		return CassandraCqlClusterFactoryBean.DEFAULT_CONTACT_POINTS;
 	}

 	/**
@@ -133,21 +139,12 @@ public abstract class AbstractClusterConfiguration {
 	}

 	/**
-	 * Returns the {@link ReconnectionPolicy}.
+	 * Returns the maximum schema agreement wait in seconds.
 	 *
-	 * @return the {@link ReconnectionPolicy}, may be {@literal null}.
+	 * @return the maximum schema agreement wait in seconds; default to {@literal 10} seconds.
 	 */
-	protected ReconnectionPolicy getReconnectionPolicy() {
-		return null;
-	}
-
-	/**
-	 * Returns the {@link RetryPolicy}.
-	 *
-	 * @return the {@link RetryPolicy}, may be {@literal null}.
-	 */
-	protected RetryPolicy getRetryPolicy() {
-		return null;
+	protected int getMaxSchemaAgreementWaitSeconds() {
+		return CassandraCqlClusterFactoryBean.DEFAULT_MAX_SCHEMA_AGREEMENT_WAIT_SECONDS;
 	}

 	/**
@@ -179,6 +176,25 @@ public abstract class AbstractClusterConfiguration {
 		return null;
 	}

+	/**
+	 * Returns the Cassandra port. Defaults to {@code 9042}
+	 *
+	 * @return the Cassandra port
+	 * @see CassandraCqlClusterFactoryBean#DEFAULT_PORT
+	 */
+	protected int getPort() {
+		return CassandraCqlClusterFactoryBean.DEFAULT_PORT;
+	}
+
+	/**
+	 * Returns the {@link ProtocolVersion}.
+	 *
+	 * @return the {@link ProtocolVersion}, may be {@literal null}.
+	 */
+	protected ProtocolVersion getProtocolVersion() {
+		return null;
+	}
+
 	/**
 	 * Returns the {@link QueryOptions}.
 	 *
@@ -189,6 +205,34 @@ public abstract class AbstractClusterConfiguration {
 		return null;
 	}

+	/**
+	 * Returns the {@link ReconnectionPolicy}.
+	 *
+	 * @return the {@link ReconnectionPolicy}, may be {@literal null}.
+	 */
+	protected ReconnectionPolicy getReconnectionPolicy() {
+		return null;
+	}
+
+	/**
+	 * Returns the {@link RetryPolicy}.
+	 *
+	 * @return the {@link RetryPolicy}, may be {@literal null}.
+	 */
+	protected RetryPolicy getRetryPolicy() {
+		return null;
+	}
+
+	/**
+	 * Returns the {@link SpeculativeExecutionPolicy}.
+	 *
+	 * @return the {@link SpeculativeExecutionPolicy}; may be {@literal null}.
+	 * @since 1.5
+	 */
+	protected SpeculativeExecutionPolicy getSpeculativeExecutionPolicy() {
+		return null;
+	}
+
 	/**
 	 * Returns the {@link SocketOptions}.
 	 *
@@ -198,6 +242,16 @@ public abstract class AbstractClusterConfiguration {
 		return null;
 	}

+	/**
+	 * Returns the {@link TimestampGenerator}.
+	 *
+	 * @return the {@link TimestampGenerator}; may be {@literal null}.
+	 * @since 1.5
+	 */
+	protected TimestampGenerator getTimestampGenerator() {
+		return null;
+	}
+
 	/**
 	 * Returns the list of keyspace creations to be run right after {@link com.datastax.driver.core.Cluster}
 	 * initialization.
--- a/spring-cql/src/main/java/org/springframework/cassandra/config/xml/CassandraCqlClusterParser.java
+++ b/spring-cql/src/main/java/org/springframework/cassandra/config/xml/CassandraCqlClusterParser.java
@@ -89,17 +89,22 @@ public class CassandraCqlClusterParser extends AbstractBeanDefinitionParser {
 	 */
 	protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {

+		addOptionalPropertyReference(builder, "addressTranslator", element, "address-translator-ref");
 		addOptionalPropertyReference(builder, "authProvider", element, "auth-info-provider-ref");
 		addOptionalPropertyReference(builder, "hostStateListener", element, "host-state-listener-ref");
 		addOptionalPropertyReference(builder, "latencyTracker", element, "latency-tracker-ref");
 		addOptionalPropertyReference(builder, "loadBalancingPolicy", element, "load-balancing-policy-ref");
 		addOptionalPropertyReference(builder, "reconnectionPolicy", element, "reconnection-policy-ref");
 		addOptionalPropertyReference(builder, "retryPolicy", element, "retry-policy-ref");
+		addOptionalPropertyReference(builder, "speculativeExecutionPolicy", element, "speculative-execution-policy-ref");
 		addOptionalPropertyReference(builder, "sslOptions", element, "ssl-options-ref");
+		addOptionalPropertyReference(builder, "timestampGenerator", element, "timestamp-generator-ref");

+		addOptionalPropertyValue(builder, "clusterName", element, "cluster-name");
 		addOptionalPropertyValue(builder, "contactPoints", element, "contact-points");
 		addOptionalPropertyValue(builder, "compressionType", element, "compression");
 		addOptionalPropertyValue(builder, "jmxReportingEnabled", element, "jmx-reporting-enabled");
+		addOptionalPropertyValue(builder, "maxSchemaAgreementWaitSeconds", element, "max-schema-agreement-wait-seconds");
 		addOptionalPropertyValue(builder, "metricsEnabled", element, "metrics-enabled");
 		addOptionalPropertyValue(builder, "password", element, "password");
 		addOptionalPropertyValue(builder, "port", element, "port");
--- a/spring-cql/src/main/resources/org/springframework/cassandra/config/spring-cql-1.0.xsd
+++ b/spring-cql/src/main/resources/org/springframework/cassandra/config/spring-cql-1.0.xsd
@@ -105,6 +105,23 @@ The name of the Cassandra Cluster definition; default is "cassandra-cluster".
 				]]></xsd:documentation>
 			</xsd:annotation>
 		</xsd:attribute>
+		<xsd:attribute name="address-translator-ref" use="optional">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+Configures the address translator to use for the new cluster.
+				]]></xsd:documentation>
+			</xsd:annotation>
+			<xsd:simpleType>
+				<xsd:annotation>
+					<xsd:appinfo>
+						<tool:annotation kind="ref">
+							<tool:assignable-to type="com.datastax.driver.core.policies.AddressTranslator"/>
+						</tool:annotation>
+					</xsd:appinfo>
+				</xsd:annotation>
+				<xsd:union memberTypes="xsd:string"/>
+			</xsd:simpleType>
+		</xsd:attribute>
 		<xsd:attribute name="auth-info-provider-ref" use="optional">
 			<xsd:annotation>
 				<xsd:documentation><![CDATA[
@@ -122,6 +139,13 @@ AuthInfoProvider implementation.
 				<xsd:union memberTypes="xsd:string" />
 			</xsd:simpleType>
 		</xsd:attribute>
+		<xsd:attribute name="cluster-name" type="xsd:string" use="optional">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+An optional name for the create cluster.
+				]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
 		<xsd:attribute name="compression" type="xsd:string" default="NONE" use="optional">
 			<xsd:annotation>
 				<xsd:documentation><![CDATA[
@@ -221,6 +245,13 @@ LoadBalancingPolicy implementation.
 				<xsd:union memberTypes="xsd:string" />
 			</xsd:simpleType>
 		</xsd:attribute>
+		<xsd:attribute name="max-schema-agreement-wait-seconds" type="xsd:string" default="10" use="optional">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+Sets the maximum time to wait for schema agreement before returning from a DDL query. Defaults to 10 seconds.
+				]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
 		<xsd:attribute name="metrics-enabled" type="xsd:string"
 					   default="true">
 			<xsd:annotation>
@@ -289,6 +320,23 @@ RetryPolicy implementation.
 				<xsd:union memberTypes="xsd:string" />
 			</xsd:simpleType>
 		</xsd:attribute>
+		<xsd:attribute name="speculative-execution-policy-ref" use="optional">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+Configures the speculative execution policy to use for the new cluster.
+				]]></xsd:documentation>
+			</xsd:annotation>
+			<xsd:simpleType>
+				<xsd:annotation>
+					<xsd:appinfo>
+						<tool:annotation kind="ref">
+							<tool:assignable-to type="com.datastax.driver.core.policies.SpeculativeExecutionPolicy"/>
+						</tool:annotation>
+					</xsd:appinfo>
+				</xsd:annotation>
+				<xsd:union memberTypes="xsd:string"/>
+			</xsd:simpleType>
+		</xsd:attribute>
 		<xsd:attribute name="ssl-enabled" type="xsd:string" default="false">
 			<xsd:annotation>
 				<xsd:documentation><![CDATA[
@@ -313,6 +361,23 @@ Custom SSL Options. sslEnabled must be true for sslOptions to be used.
 				<xsd:union memberTypes="xsd:string" />
 			</xsd:simpleType>
 		</xsd:attribute>
+		<xsd:attribute name="timestamp-generator-ref" use="optional">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+Configures the generator that will produce the client-side timestamp sent with each query.
+				]]></xsd:documentation>
+			</xsd:annotation>
+			<xsd:simpleType>
+				<xsd:annotation>
+					<xsd:appinfo>
+						<tool:annotation kind="ref">
+							<tool:assignable-to type="com.datastax.driver.core.TimestampGenerator"/>
+						</tool:annotation>
+					</xsd:appinfo>
+				</xsd:annotation>
+				<xsd:union memberTypes="xsd:string"/>
+			</xsd:simpleType>
+		</xsd:attribute>
 		<xsd:attribute name="username" type="xsd:string" use="optional">
 			<xsd:annotation>
 				<xsd:documentation><![CDATA[
--- a/spring-cql/src/test/java/org/springframework/cassandra/config/CassandraCqlClusterFactoryBeanUnitTests.java
+++ b/spring-cql/src/test/java/org/springframework/cassandra/config/CassandraCqlClusterFactoryBeanUnitTests.java
@@ -18,35 +18,44 @@ package org.springframework.cassandra.config;

 import static org.hamcrest.MatcherAssert.*;
 import static org.hamcrest.Matchers.*;
+import static org.mockito.Mockito.*;

 import org.junit.Test;
+import org.mockito.Matchers;
 import org.springframework.test.util.ReflectionTestUtils;

 import com.datastax.driver.core.AuthProvider;
+import com.datastax.driver.core.Cluster;
 import com.datastax.driver.core.Configuration;
 import com.datastax.driver.core.JdkSSLOptions;
 import com.datastax.driver.core.PlainTextAuthProvider;
 import com.datastax.driver.core.PoolingOptions;
+import com.datastax.driver.core.ProtocolOptions;
 import com.datastax.driver.core.ProtocolOptions.Compression;
 import com.datastax.driver.core.ProtocolVersion;
 import com.datastax.driver.core.QueryOptions;
 import com.datastax.driver.core.SSLOptions;
 import com.datastax.driver.core.SocketOptions;
+import com.datastax.driver.core.TimestampGenerator;
+import com.datastax.driver.core.policies.AddressTranslator;
 import com.datastax.driver.core.policies.ExponentialReconnectionPolicy;
 import com.datastax.driver.core.policies.LoadBalancingPolicy;
+import com.datastax.driver.core.policies.Policies;
 import com.datastax.driver.core.policies.ReconnectionPolicy;
 import com.datastax.driver.core.policies.RoundRobinPolicy;
+import com.datastax.driver.core.policies.SpeculativeExecutionPolicy;

 /**
 * Unit tests for {@link CassandraCqlClusterFactoryBean}.
 *
- * @see DATACASS-226
 * @author Mark Paluch
+ * @author John Blum
+ * @see <a href="https://jira.spring.io/browse/DATACASS-226">DATACASS-226</a>
 */
 public class CassandraCqlClusterFactoryBeanUnitTests {

 	/**
-	 * @see DATACASS-226
+	 * @see <a href="https://jira.spring.io/browse/DATACASS-226">DATACASS-226</a>
 	 * @throws Exception
 	 */
 	@Test
@@ -62,7 +71,7 @@ public class CassandraCqlClusterFactoryBeanUnitTests {
 	}

 	/**
-	 * @see DATACASS-226
+	 * @see <a href="https://jira.spring.io/browse/DATACASS-226">DATACASS-226</a>
 	 * @throws Exception
 	 */
 	@Test
@@ -76,7 +85,7 @@ public class CassandraCqlClusterFactoryBeanUnitTests {
 	}

 	/**
-	 * @see DATACASS-226
+	 * @see <a href="https://jira.spring.io/browse/DATACASS-226">DATACASS-226</a>
 	 * @throws Exception
 	 */
 	@Test
@@ -92,7 +101,7 @@ public class CassandraCqlClusterFactoryBeanUnitTests {
 	}

 	/**
-	 * @see DATACASS-226
+	 * @see <a href="https://jira.spring.io/browse/DATACASS-226">DATACASS-226</a>
 	 * @throws Exception
 	 */
 	@Test
@@ -108,7 +117,7 @@ public class CassandraCqlClusterFactoryBeanUnitTests {
 	}

 	/**
-	 * @see DATACASS-226
+	 * @see <a href="https://jira.spring.io/browse/DATACASS-226">DATACASS-226</a>
 	 * @throws Exception
 	 */
 	@Test
@@ -124,7 +133,7 @@ public class CassandraCqlClusterFactoryBeanUnitTests {
 	}

 	/**
-	 * @see DATACASS-226
+	 * @see <a href="https://jira.spring.io/browse/DATACASS-226">DATACASS-226</a>
 	 * @throws Exception
 	 */
 	@Test
@@ -140,7 +149,7 @@ public class CassandraCqlClusterFactoryBeanUnitTests {
 	}

 	/**
-	 * @see DATACASS-226
+	 * @see <a href="https://jira.spring.io/browse/DATACASS-226">DATACASS-226</a>
 	 * @throws Exception
 	 */
 	@Test
@@ -156,7 +165,7 @@ public class CassandraCqlClusterFactoryBeanUnitTests {
 	}

 	/**
-	 * @see DATACASS-226
+	 * @see <a href="https://jira.spring.io/browse/DATACASS-226">DATACASS-226</a>
 	 * @throws Exception
 	 */
 	@Test
@@ -172,8 +181,8 @@ public class CassandraCqlClusterFactoryBeanUnitTests {
 	}

 	/**
-	 * @see DATACASS-226
-	 * @see DATACASS-263
+	 * @see <a href="https://jira.spring.io/browse/DATACASS-226">DATACASS-226</a>
+	 * @see <a href="https://jira.spring.io/browse/DATACASS-263">DATACASS-263</a>
 	 * @throws Exception
 	 */
 	@Test
@@ -190,8 +199,8 @@ public class CassandraCqlClusterFactoryBeanUnitTests {
 	}

 	/**
-	 * @see DATACASS-226
-	 * @see DATACASS-263
+	 * @see <a href="https://jira.spring.io/browse/DATACASS-226">DATACASS-226</a>
+	 * @see <a href="https://jira.spring.io/browse/DATACASS-263">DATACASS-263</a>
 	 * @throws Exception
 	 */
 	@Test
@@ -209,7 +218,7 @@ public class CassandraCqlClusterFactoryBeanUnitTests {
 	}

 	/**
-	 * @see DATACASS-226
+	 * @see <a href="https://jira.spring.io/browse/DATACASS-226">DATACASS-226</a>
 	 * @throws Exception
 	 */
 	@Test
@@ -225,7 +234,7 @@ public class CassandraCqlClusterFactoryBeanUnitTests {
 	}

 	/**
-	 * @see DATACASS-226
+	 * @see <a href="https://jira.spring.io/browse/DATACASS-226">DATACASS-226</a>
 	 * @throws Exception
 	 */
 	@Test
@@ -241,7 +250,7 @@ public class CassandraCqlClusterFactoryBeanUnitTests {
 	}

 	/**
-	 * @see DATACASS-226
+	 * @see <a href="https://jira.spring.io/browse/DATACASS-226">DATACASS-226</a>
 	 * @throws Exception
 	 */
 	@Test
@@ -256,7 +265,7 @@ public class CassandraCqlClusterFactoryBeanUnitTests {
 	}

 	/**
-	 * @see DATACASS-226
+	 * @see <a href="https://jira.spring.io/browse/DATACASS-226">DATACASS-226</a>
 	 * @throws Exception
 	 */
 	@Test
@@ -273,7 +282,7 @@ public class CassandraCqlClusterFactoryBeanUnitTests {
 	}

 	/**
-	 * @see DATACASS-226
+	 * @see <a href="https://jira.spring.io/browse/DATACASS-226">DATACASS-226</a>
 	 * @throws Exception
 	 */
 	@Test
@@ -287,7 +296,7 @@ public class CassandraCqlClusterFactoryBeanUnitTests {
 	}

 	/**
-	 * @see DATACASS-226
+	 * @see <a href="https://jira.spring.io/browse/DATACASS-226">DATACASS-226</a>
 	 * @throws Exception
 	 */
 	@Test
@@ -300,6 +309,119 @@ public class CassandraCqlClusterFactoryBeanUnitTests {
 		assertThat(getConfiguration(bean).getMetricsOptions().isJMXReportingEnabled(), is(false));
 	}

+	/**
+	 * @see <a href="https://jira.spring.io/browse/DATACASS-316">DATACASS-316</a>
+	 */
+	@Test
+	public void shouldSetAddressTranslator() throws Exception {
+
+		AddressTranslator mockAddressTranslator = mock(AddressTranslator.class);
+		CassandraCqlClusterFactoryBean bean = new CassandraCqlClusterFactoryBean();
+
+		bean.setAddressTranslator(mockAddressTranslator);
+		bean.afterPropertiesSet();
+
+		assertThat(getPolicies(bean).getAddressTranslator(), is(equalTo(mockAddressTranslator)));
+	}
+
+	/**
+	 * @see <a href="https://jira.spring.io/browse/DATACASS-317">DATACASS-317</a>
+	 */
+	@Test
+	public void shouldSetClusterNameWithBeanNameProperty() throws Exception {
+
+		final Cluster.Builder mockClusterBuilder = mock(Cluster.Builder.class);
+
+		when(mockClusterBuilder.addContactPoints(Matchers.<String[]>anyVararg())).thenReturn(mockClusterBuilder);
+
+		CassandraCqlClusterFactoryBean bean = new CassandraCqlClusterFactoryBean() {
+			@Override Cluster.Builder newClusterBuilder() {
+				return mockClusterBuilder;
+			}
+		};
+
+		bean.setBeanName("ABC");
+		bean.setClusterName(" ");
+		bean.afterPropertiesSet();
+
+		verify(mockClusterBuilder, times(1)).withClusterName(eq("ABC"));
+	}
+
+	/**
+	 * @see <a href="https://jira.spring.io/browse/DATACASS-317">DATACASS-317</a>
+	 */
+	@Test
+	public void shouldSetClusterNameWithClusterNameProperty() throws Exception {
+
+		final Cluster.Builder mockClusterBuilder = mock(Cluster.Builder.class);
+
+		when(mockClusterBuilder.addContactPoints(Matchers.<String[]>anyVararg())).thenReturn(mockClusterBuilder);
+
+		CassandraCqlClusterFactoryBean bean = new CassandraCqlClusterFactoryBean() {
+			@Override Cluster.Builder newClusterBuilder() {
+				return mockClusterBuilder;
+			}
+		};
+
+		bean.setBeanName("ABC");
+		bean.setClusterName("XYZ");
+		bean.afterPropertiesSet();
+
+		verify(mockClusterBuilder, times(1)).withClusterName(eq("XYZ"));
+	}
+
+	/**
+	 * @see <a href="https://jira.spring.io/browse/DATACASS-319">DATACASS-319</a>
+	 */
+	@Test
+	public void shouldSetMaxSchemaAgreementWaitSeconds() throws Exception {
+
+		CassandraCqlClusterFactoryBean bean = new CassandraCqlClusterFactoryBean();
+
+		bean.setMaxSchemaAgreementWaitSeconds(20);
+		bean.afterPropertiesSet();
+
+		assertThat(getProtocolOptions(bean).getMaxSchemaAgreementWaitSeconds(), is(equalTo(20)));
+	}
+
+	/**
+	 * @see <a href="https://jira.spring.io/browse/DATACASS-320">DATACASS-320</a>
+	 */
+	@Test
+	public void shouldSetSpeculativeExecutionPolicy() throws Exception {
+
+		SpeculativeExecutionPolicy mockSpeculativeExecutionPolicy = mock(SpeculativeExecutionPolicy.class);
+		CassandraCqlClusterFactoryBean bean = new CassandraCqlClusterFactoryBean();
+
+		bean.setSpeculativeExecutionPolicy(mockSpeculativeExecutionPolicy);
+		bean.afterPropertiesSet();
+
+		assertThat(getPolicies(bean).getSpeculativeExecutionPolicy(), is(equalTo(mockSpeculativeExecutionPolicy)));
+	}
+
+	/**
+	 * @see <a href="https://jira.spring.io/browse/DATACASS-238">DATACASS-238</a>
+	 */
+	@Test
+	public void shouldSetTimestampGenerator() throws Exception {
+
+		TimestampGenerator mockTimestampGenerator = mock(TimestampGenerator.class);
+		CassandraCqlClusterFactoryBean bean = new CassandraCqlClusterFactoryBean();
+
+		bean.setTimestampGenerator(mockTimestampGenerator);
+		bean.afterPropertiesSet();
+
+		assertThat(getPolicies(bean).getTimestampGenerator(), is(equalTo(mockTimestampGenerator)));
+	}
+
+	private Policies getPolicies(CassandraCqlClusterFactoryBean bean) throws Exception {
+		return getConfiguration(bean).getPolicies();
+	}
+
+	private ProtocolOptions getProtocolOptions(CassandraCqlClusterFactoryBean bean) throws Exception {
+		return getConfiguration(bean).getProtocolOptions();
+	}
+
 	private Configuration getConfiguration(CassandraCqlClusterFactoryBean bean) throws Exception {
 		return bean.getObject().getConfiguration();
 	}
--- a/spring-cql/src/test/java/org/springframework/cassandra/config/java/AbstractClusterConfigurationUnitTests.java
+++ b/spring-cql/src/test/java/org/springframework/cassandra/config/java/AbstractClusterConfigurationUnitTests.java
@@ -34,14 +34,19 @@ import com.datastax.driver.core.Cluster;
 import com.datastax.driver.core.Configuration;
 import com.datastax.driver.core.PlainTextAuthProvider;
 import com.datastax.driver.core.PoolingOptions;
+import com.datastax.driver.core.ProtocolOptions;
 import com.datastax.driver.core.ProtocolOptions.Compression;
 import com.datastax.driver.core.ProtocolVersion;
 import com.datastax.driver.core.QueryOptions;
 import com.datastax.driver.core.SocketOptions;
+import com.datastax.driver.core.TimestampGenerator;
+import com.datastax.driver.core.policies.AddressTranslator;
 import com.datastax.driver.core.policies.ExponentialReconnectionPolicy;
 import com.datastax.driver.core.policies.LoadBalancingPolicy;
+import com.datastax.driver.core.policies.Policies;
 import com.datastax.driver.core.policies.ReconnectionPolicy;
 import com.datastax.driver.core.policies.RoundRobinPolicy;
+import com.datastax.driver.core.policies.SpeculativeExecutionPolicy;

 /**
 * Unit tests for {@link AbstractClusterConfiguration}.
@@ -53,7 +58,7 @@ import com.datastax.driver.core.policies.RoundRobinPolicy;
 public class AbstractClusterConfigurationUnitTests {

 	/**
-	 * @see DATACASS-226
+	 * @see <a href="https://jira.spring.io/browse/DATACASS-226"></a>
 	 * @throws Exception
 	 */
 	@Test
@@ -72,7 +77,7 @@ public class AbstractClusterConfigurationUnitTests {
 	}

 	/**
-	 * @see DATACASS-226
+	 * @see <a href="https://jira.spring.io/browse/DATACASS-226"></a>
 	 * @throws Exception
 	 */
 	@Test
@@ -92,7 +97,7 @@ public class AbstractClusterConfigurationUnitTests {
 	}

 	/**
-	 * @see DATACASS-226
+	 * @see <a href="https://jira.spring.io/browse/DATACASS-226"></a>
 	 * @throws Exception
 	 */
 	@Test
@@ -112,7 +117,7 @@ public class AbstractClusterConfigurationUnitTests {
 	}

 	/**
-	 * @see DATACASS-226
+	 * @see <a href="https://jira.spring.io/browse/DATACASS-226"></a>
 	 * @throws Exception
 	 */
 	@Test
@@ -132,7 +137,7 @@ public class AbstractClusterConfigurationUnitTests {
 	}

 	/**
-	 * @see DATACASS-226
+	 * @see <a href="https://jira.spring.io/browse/DATACASS-226"></a>
 	 * @throws Exception
 	 */
 	@Test
@@ -152,7 +157,7 @@ public class AbstractClusterConfigurationUnitTests {
 	}

 	/**
-	 * @see DATACASS-226
+	 * @see <a href="https://jira.spring.io/browse/DATACASS-226"></a>
 	 * @throws Exception
 	 */
 	@Test
@@ -172,7 +177,7 @@ public class AbstractClusterConfigurationUnitTests {
 	}

 	/**
-	 * @see DATACASS-226
+	 * @see <a href="https://jira.spring.io/browse/DATACASS-226"></a>
 	 * @throws Exception
 	 */
 	@Test
@@ -192,7 +197,7 @@ public class AbstractClusterConfigurationUnitTests {
 	}

 	/**
-	 * @see DATACASS-226
+	 * @see <a href="https://jira.spring.io/browse/DATACASS-226"></a>
 	 * @throws Exception
 	 */
 	@Test
@@ -212,7 +217,7 @@ public class AbstractClusterConfigurationUnitTests {
 	}

 	/**
-	 * @see DATACASS-226
+	 * @see <a href="https://jira.spring.io/browse/DATACASS-226"></a>
 	 * @throws Exception
 	 */
 	@Test
@@ -231,7 +236,7 @@ public class AbstractClusterConfigurationUnitTests {
 	}

 	/**
-	 * @see DATACASS-226
+	 * @see <a href="https://jira.spring.io/browse/DATACASS-226"></a>
 	 * @throws Exception
 	 */
 	@Test
@@ -249,7 +254,7 @@ public class AbstractClusterConfigurationUnitTests {
 	}

 	/**
-	 * @see DATACASS-226
+	 * @see <a href="https://jira.spring.io/browse/DATACASS-226"></a>
 	 * @throws Exception
 	 */
 	@Test
@@ -268,7 +273,7 @@ public class AbstractClusterConfigurationUnitTests {
 	}

 	/**
-	 * @see DATACASS-226
+	 * @see <a href="https://jira.spring.io/browse/DATACASS-226"></a>
 	 * @throws Exception
 	 */
 	@Test
@@ -287,7 +292,7 @@ public class AbstractClusterConfigurationUnitTests {
 	}

 	/**
-	 * @see DATACASS-226
+	 * @see <a href="https://jira.spring.io/browse/DATACASS-226"></a>
 	 * @throws Exception
 	 */
 	@Test
@@ -305,7 +310,7 @@ public class AbstractClusterConfigurationUnitTests {
 	}

 	/**
-	 * @see DATACASS-226
+	 * @see <a href="https://jira.spring.io/browse/DATACASS-226"></a>
 	 * @throws Exception
 	 */
 	@Test
@@ -322,12 +327,100 @@ public class AbstractClusterConfigurationUnitTests {
 		assertThat(clusterConfiguration.cluster().getShutdownScripts(), is(equalTo(scripts)));
 	}

+	/**
+	 * <a href="https://jira.spring.io/browse/DATACASS-316">DATACASS-316</a>
+	 */
+	@Test
+	public void shouldSetAddressTranslator() throws Exception {
+		final AddressTranslator mockAddressTranslator = mock(AddressTranslator.class);
+
+		AbstractClusterConfiguration clusterConfiguration = new AbstractClusterConfiguration() {
+			@Override protected AddressTranslator getAddressTranslator() {
+				return mockAddressTranslator;
+			}
+		};
+
+		assertThat(getPolicies(getCluster(clusterConfiguration)).getAddressTranslator(),
+			is(equalTo(mockAddressTranslator)));
+	}
+
+	/**
+	 * <a href="https://jira.spring.io/browse/DATACASS-120">DATACASS-120</a>
+	 * <a href="https://jira.spring.io/browse/DATACASS-317">DATACASS-317</a>
+	 */
+	@Test
+	public void shouldSetClusterName() throws Exception {
+		AbstractClusterConfiguration clusterConfiguration = new AbstractClusterConfiguration() {
+			@Override protected String getClusterName() {
+				return "testCluster";
+			}
+		};
+
+		assertThat(getCluster(clusterConfiguration).getClusterName(), is(equalTo("testCluster")));
+	}
+
+	/**
+	 * <a href="https://jira.spring.io/browse/DATACASS-319">DATACASS-319</a>
+	 */
+	@Test
+	public void shouldSetMaxSchemaAgreementWaitInSeconds() throws Exception {
+		AbstractClusterConfiguration clusterConfiguration = new AbstractClusterConfiguration() {
+			@Override protected int getMaxSchemaAgreementWaitSeconds() {
+				return 30;
+			}
+		};
+
+		assertThat(getProtocolOptions(getCluster(clusterConfiguration)).getMaxSchemaAgreementWaitSeconds(),
+			is(equalTo(30)));
+	}
+
+	/**
+	 * <a href="https://jira.spring.io/browse/DATACASS-320">DATACASS-320</a>
+	 */
+	@Test
+	public void shouldSetSpeculativeExecutionPolicy() throws Exception {
+		final SpeculativeExecutionPolicy mockSpeculativeExecutionPolicy = mock(SpeculativeExecutionPolicy.class);
+
+		AbstractClusterConfiguration clusterConfiguration = new AbstractClusterConfiguration() {
+			@Override protected SpeculativeExecutionPolicy getSpeculativeExecutionPolicy() {
+				return mockSpeculativeExecutionPolicy;
+			}
+		};
+
+		assertThat(getPolicies(getCluster(clusterConfiguration)).getSpeculativeExecutionPolicy(),
+			is(equalTo(mockSpeculativeExecutionPolicy)));
+	}
+
+	/**
+	 * <a href="https://jira.spring.io/browse/DATACASS-238">DATACASS-238</a>
+	 */
+	@Test
+	public void shouldSetTimestampGenerator() throws Exception {
+		final TimestampGenerator mockTimestampGenerator = mock(TimestampGenerator.class);
+
+		AbstractClusterConfiguration clusterConfiguration = new AbstractClusterConfiguration() {
+			@Override protected TimestampGenerator getTimestampGenerator() {
+				return mockTimestampGenerator;
+			}
+		};
+
+		assertThat(getPolicies(getCluster(clusterConfiguration)).getTimestampGenerator(),
+			is(equalTo(mockTimestampGenerator)));
+	}
+
+	private Policies getPolicies(Cluster cluster) throws Exception {
+		return getConfiguration(cluster).getPolicies();
+	}
+
+	private ProtocolOptions getProtocolOptions(Cluster cluster) throws Exception {
+		return getConfiguration(cluster).getProtocolOptions();
+	}
+
 	private Configuration getConfiguration(Cluster cluster) throws Exception {
 		return cluster.getConfiguration();
 	}

 	private Cluster getCluster(AbstractClusterConfiguration clusterConfiguration) throws Exception {
-
 		CassandraCqlClusterFactoryBean cluster = clusterConfiguration.cluster();
 		cluster.afterPropertiesSet();
 		return cluster.getObject();
--- a/spring-cql/src/test/java/org/springframework/cassandra/config/xml/CassandraCqlClusterParserUnitTests.java
+++ b/spring-cql/src/test/java/org/springframework/cassandra/config/xml/CassandraCqlClusterParserUnitTests.java
@@ -15,6 +15,7 @@
 */
 package org.springframework.cassandra.config.xml;

+import static org.hamcrest.Matchers.contains;
 import static org.hamcrest.Matchers.*;
 import static org.junit.Assert.*;
 import static org.mockito.Mockito.*;
@@ -85,16 +86,21 @@ public class CassandraCqlClusterParserUnitTests {
 		BeanDefinition mockContainingBeanDefinition = mock(BeanDefinition.class);

 		when(mockContainingBeanDefinition.getScope()).thenReturn("Singleton");
+		when(mockElement.getAttribute("address-translator-ref")).thenReturn("testAddressTranslator");
 		when(mockElement.getAttribute("auth-info-provider-ref")).thenReturn("testAuthInfoProvider");
 		when(mockElement.getAttribute("host-state-listener-ref")).thenReturn("testHostStateListener");
 		when(mockElement.getAttribute("latency-tracker-ref")).thenReturn("testLatencyTracker");
 		when(mockElement.getAttribute("load-balancing-policy-ref")).thenReturn("testLoadBalancingPolicy");
 		when(mockElement.getAttribute("reconnection-policy-ref")).thenReturn("testReconnectionPolicy");
 		when(mockElement.getAttribute("retry-policy-ref")).thenReturn("testRetryPolicy");
+		when(mockElement.getAttribute("speculative-execution-policy-ref")).thenReturn("testSpeculativeExecutionPolicy");
 		when(mockElement.getAttribute("ssl-options-ref")).thenReturn("testSslOptions");
+		when(mockElement.getAttribute("timestamp-generator-ref")).thenReturn("testTimestampGenerator");
+		when(mockElement.getAttribute("cluster-name")).thenReturn("testCluster");
 		when(mockElement.getAttribute("contact-points")).thenReturn("skullbox");
 		when(mockElement.getAttribute("compression")).thenReturn("SNAPPY");
 		when(mockElement.getAttribute("jmx-reporting-enabled")).thenReturn("true");
+		when(mockElement.getAttribute("max-schema-agreement-wait-seconds")).thenReturn("30");
 		when(mockElement.getAttribute("metrics-enabled")).thenReturn("true");
 		when(mockElement.getAttribute("password")).thenReturn("p@55w0rd");
 		when(mockElement.getAttribute("port")).thenReturn("12345");
@@ -116,16 +122,21 @@ public class CassandraCqlClusterParserUnitTests {
 		assertThat(beanDefinition.getDestroyMethodName(), is(equalTo("destroy")));
 		assertThat((Element) beanDefinition.getSource(), is(equalTo(mockElement)));
 		assertThat(beanDefinition.isLazyInit(), is(false));
+		assertThat(getPropertyValueAsString(beanDefinition, "addressTranslator"), is(equalTo("testAddressTranslator")));
 		assertThat(getPropertyValueAsString(beanDefinition, "authProvider"), is(equalTo("testAuthInfoProvider")));
 		assertThat(getPropertyValueAsString(beanDefinition, "hostStateListener"), is(equalTo("testHostStateListener")));
 		assertThat(getPropertyValueAsString(beanDefinition, "latencyTracker"), is(equalTo("testLatencyTracker")));
 		assertThat(getPropertyValueAsString(beanDefinition, "loadBalancingPolicy"), is(equalTo("testLoadBalancingPolicy")));
 		assertThat(getPropertyValueAsString(beanDefinition, "reconnectionPolicy"), is(equalTo("testReconnectionPolicy")));
 		assertThat(getPropertyValueAsString(beanDefinition, "retryPolicy"), is(equalTo("testRetryPolicy")));
+		assertThat(getPropertyValueAsString(beanDefinition, "speculativeExecutionPolicy"), is(equalTo("testSpeculativeExecutionPolicy")));
 		assertThat(getPropertyValueAsString(beanDefinition, "sslOptions"), is(equalTo("testSslOptions")));
+		assertThat(getPropertyValueAsString(beanDefinition, "timestampGenerator"), is(equalTo("testTimestampGenerator")));
+		assertThat(getPropertyValueAsString(beanDefinition, "clusterName"), is(equalTo("testCluster")));
 		assertThat(getPropertyValueAsString(beanDefinition, "contactPoints"), is(equalTo("skullbox")));
 		assertThat(getPropertyValueAsString(beanDefinition, "compressionType"), is(equalTo("SNAPPY")));
 		assertThat(getPropertyValueAsString(beanDefinition, "jmxReportingEnabled"), is(equalTo("true")));
+		assertThat(getPropertyValueAsString(beanDefinition, "maxSchemaAgreementWaitSeconds"), is(equalTo("30")));
 		assertThat(getPropertyValueAsString(beanDefinition, "metricsEnabled"), is(equalTo("true")));
 		assertThat(getPropertyValueAsString(beanDefinition, "password"), is(equalTo("p@55w0rd")));
 		assertThat(getPropertyValueAsString(beanDefinition, "port"), is(equalTo("12345")));
@@ -133,16 +144,21 @@ public class CassandraCqlClusterParserUnitTests {
 		assertThat(getPropertyValueAsString(beanDefinition, "username"), is(equalTo("jonDoe")));

 		verify(mockContainingBeanDefinition).getScope();
+		verify(mockElement).getAttribute(eq("address-translator-ref"));
 		verify(mockElement).getAttribute(eq("auth-info-provider-ref"));
 		verify(mockElement).getAttribute(eq("host-state-listener-ref"));
 		verify(mockElement).getAttribute(eq("latency-tracker-ref"));
 		verify(mockElement).getAttribute(eq("load-balancing-policy-ref"));
 		verify(mockElement).getAttribute(eq("reconnection-policy-ref"));
 		verify(mockElement).getAttribute(eq("retry-policy-ref"));
+		verify(mockElement).getAttribute(eq("speculative-execution-policy-ref"));
+		verify(mockElement).getAttribute(eq("timestamp-generator-ref"));
 		verify(mockElement).getAttribute(eq("ssl-options-ref"));
+		verify(mockElement).getAttribute(eq("cluster-name"));
 		verify(mockElement).getAttribute(eq("contact-points"));
 		verify(mockElement).getAttribute(eq("compression"));
 		verify(mockElement).getAttribute(eq("jmx-reporting-enabled"));
+		verify(mockElement).getAttribute(eq("max-schema-agreement-wait-seconds"));
 		verify(mockElement).getAttribute(eq("metrics-enabled"));
 		verify(mockElement).getAttribute(eq("password"));
 		verify(mockElement).getAttribute(eq("port"));
--- a/spring-cql/src/test/java/org/springframework/cassandra/support/BeanDefinitionTestUtils.java
+++ b/spring-cql/src/test/java/org/springframework/cassandra/support/BeanDefinitionTestUtils.java
@@ -23,7 +23,7 @@ import org.springframework.util.Assert;
 /**
 * {@code BeanDefinitionTestUtils} is a collection of {@link org.springframework.beans.factory.config.BeanDefinition}
 * -based utility methods for use in unit and integration testing scenarios.
- * 
+ *
 * @author Mark Paluch
 */
 public abstract class BeanDefinitionTestUtils {
@@ -35,7 +35,7 @@ public abstract class BeanDefinitionTestUtils {

 	/**
 	 * Retrieve the {@code propertyValue} from a {@link BeanDefinition} by its {@code propertyName}.
-	 * 
+	 *
 	 * @param beanDefinition must not be {@literal null}.
 	 * @param propertyName must not be {@literal null} or empty.
 	 * @return the property value, may be {@literal null}.
@@ -47,12 +47,13 @@ public abstract class BeanDefinitionTestUtils {
 		Assert.notNull(propertyName, "Property name must not be empty");

 		PropertyValue propertyValue = beanDefinition.getPropertyValues().getPropertyValue(propertyName);
+
 		return (T) (propertyValue != null ? propertyValue.getValue() : null);
 	}

 	/**
 	 * Retrieve the {@code propertyValue} as {@literal String} from a {@link BeanDefinition} by its {@code propertyName}.
-	 * 
+	 *
 	 * @param beanDefinition must not be {@literal null}.
 	 * @param propertyName must not be {@literal null} or empty.
 	 * @return the property value, may be {@literal null}.
@@ -63,8 +64,8 @@ public abstract class BeanDefinitionTestUtils {
 		Assert.notNull(propertyName, "Property name must not be empty");

 		Object value = getPropertyValue(beanDefinition, propertyName);
-		return (value instanceof RuntimeBeanReference ? ((RuntimeBeanReference) value).getBeanName()
-				: (value != null ? String.valueOf(value) : null));
-	}

+		return (value instanceof RuntimeBeanReference ? ((RuntimeBeanReference) value).getBeanName()
+			: (value != null ? String.valueOf(value) : null));
+	}
 }
--- a/spring-cql/src/test/java/org/springframework/cassandra/test/integration/config/xml/XmlConfigIntegrationTests.java
+++ b/spring-cql/src/test/java/org/springframework/cassandra/test/integration/config/xml/XmlConfigIntegrationTests.java
@@ -35,6 +35,9 @@ import com.datastax.driver.core.HostDistance;
 import com.datastax.driver.core.PoolingOptions;
 import com.datastax.driver.core.Session;
 import com.datastax.driver.core.SocketOptions;
+import com.datastax.driver.core.TimestampGenerator;
+import com.datastax.driver.core.policies.AddressTranslator;
+import com.datastax.driver.core.policies.SpeculativeExecutionPolicy;

 /**
 * Test XML namespace configuration using the spring-cql-1.0.xsd.
@@ -49,21 +52,29 @@ public class XmlConfigIntegrationTests extends AbstractEmbeddedCassandraIntegrat

 	public static final String KEYSPACE = "xmlconfigtest";

-	@Rule public KeyspaceRule keyspaceRule = new KeyspaceRule(cassandraEnvironment, KEYSPACE);
+	@Rule
+	public KeyspaceRule keyspaceRule = new KeyspaceRule(cassandraEnvironment, KEYSPACE);

 	private ConfigurableApplicationContext applicationContext;
+
+	private AddressTranslator addressTranslator;
 	private Cluster cluster;
 	private Executor executor;
 	private Session session;
+	private SpeculativeExecutionPolicy speculativeExecutionPolicy;
+	private TimestampGenerator timestampGenerator;

 	@Before
 	public void setUp() {
 		this.applicationContext = new ClassPathXmlApplicationContext(
 			"XmlConfigIntegrationTests-context.xml", getClass());

+		this.addressTranslator = applicationContext.getBean(AddressTranslator.class);
 		this.cluster = applicationContext.getBean(Cluster.class);
 		this.executor = applicationContext.getBean(Executor.class);
 		this.session = applicationContext.getBean(Session.class);
+		this.speculativeExecutionPolicy = applicationContext.getBean(SpeculativeExecutionPolicy.class);
+		this.timestampGenerator = applicationContext.getBean(TimestampGenerator.class);
 	}

 	@After
@@ -78,6 +89,18 @@ public class XmlConfigIntegrationTests extends AbstractEmbeddedCassandraIntegrat
 		IntegrationTestUtils.assertKeyspaceExists(KEYSPACE, session);
 	}

+	@Test
+	public void clusterConfigurationIsCorrect() {
+		assertThat(cluster.getConfiguration().getPolicies().getAddressTranslator(), is(equalTo(addressTranslator)));
+		assertThat(cluster.getClusterName(), is(equalTo("skynet")));
+		assertThat(cluster.getConfiguration().getProtocolOptions().getMaxSchemaAgreementWaitSeconds(), is(equalTo(30)));
+
+		assertThat(cluster.getConfiguration().getPolicies().getSpeculativeExecutionPolicy(),
+			is(equalTo(speculativeExecutionPolicy)));
+
+		assertThat(cluster.getConfiguration().getPolicies().getTimestampGenerator(), is(equalTo(timestampGenerator)));
+	}
+
 	/**
 	 * @see <a href="https://jira.spring.io/browse/DATACASS-298">DATACASS-298</a>
 	 */
--- a/spring-cql/src/test/resources/org/springframework/cassandra/test/integration/config/xml/XmlConfigIntegrationTests-context.xml
+++ b/spring-cql/src/test/resources/org/springframework/cassandra/test/integration/config/xml/XmlConfigIntegrationTests-context.xml
@@ -15,12 +15,25 @@

 	<task:executor id="testExecutor" pool-size="10"/>

-	<cassandra:cluster contact-points="localhost" port="${build.cassandra.native_transport_port}"
-			heartbeat-interval-seconds="60"
-			initialization-executor-ref="testExecutor"
-			idle-timeout-seconds="300"
-			pool-timeout-milliseconds="15000">
-		<cassandra:local-pooling-options core-connections="2"
+	<bean name="testAddressTranslator" class="com.datastax.driver.core.policies.IdentityTranslator"/>
+
+	<bean name="testSpeculativeExecutionPolicy" class="com.datastax.driver.core.policies.NoSpeculativeExecutionPolicy"/>
+
+	<bean name="testTimestampGenerator" class="com.datastax.driver.core.AtomicMonotonicTimestampGenerator"/>
+
+	<cassandra:cluster
+		cluster-name="skynet" contact-points="localhost" port="${build.cassandra.native_transport_port}"
+		address-translator-ref="testAddressTranslator"
+		heartbeat-interval-seconds="60"
+		initialization-executor-ref="testExecutor"
+		idle-timeout-seconds="300"
+		max-schema-agreement-wait-seconds="30"
+		pool-timeout-milliseconds="15000"
+		speculative-execution-policy-ref="testSpeculativeExecutionPolicy"
+		timestamp-generator-ref="testTimestampGenerator">
+
+		<cassandra:local-pooling-options
+			core-connections="2"
 			max-connections="8"
 			max-simultaneous-requests="100"
 			min-simultaneous-requests="25"/>
@@ -38,6 +51,7 @@
 			send-buffer-size="65536"
 			so-linger="60"
 			tcp-no-delay="true"/>
+
 	</cassandra:cluster>

 	<cassandra:session keyspace-name="xmlconfigtest"/>
--- a/spring-data-cassandra/src/main/resources/org/springframework/data/cassandra/config/spring-cassandra-1.0.xsd
+++ b/spring-data-cassandra/src/main/resources/org/springframework/data/cassandra/config/spring-cassandra-1.0.xsd
@@ -114,6 +114,23 @@ The name of the Cassandra Cluster definition; default is "cassandra-cluster".
 				]]></xsd:documentation>
 			</xsd:annotation>
 		</xsd:attribute>
+		<xsd:attribute name="address-translator-ref" use="optional">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+Configures the address translator to use for the new cluster.
+				]]></xsd:documentation>
+			</xsd:annotation>
+			<xsd:simpleType>
+				<xsd:annotation>
+					<xsd:appinfo>
+						<tool:annotation kind="ref">
+							<tool:assignable-to type="com.datastax.driver.core.policies.AddressTranslator"/>
+						</tool:annotation>
+					</xsd:appinfo>
+				</xsd:annotation>
+				<xsd:union memberTypes="xsd:string"/>
+			</xsd:simpleType>
+		</xsd:attribute>
 		<xsd:attribute name="auth-info-provider-ref" use="optional">
 			<xsd:annotation>
 				<xsd:documentation><![CDATA[
@@ -131,6 +148,13 @@ AuthInfoProvider implementation.
 				<xsd:union memberTypes="xsd:string" />
 			</xsd:simpleType>
 		</xsd:attribute>
+		<xsd:attribute name="cluster-name" type="xsd:string" use="optional">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+An optional name for the create cluster.
+				]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
 		<xsd:attribute name="compression" type="xsd:string" default="NONE" use="optional">
 			<xsd:annotation>
 				<xsd:documentation><![CDATA[
@@ -230,6 +254,13 @@ LoadBalancingPolicy implementation.
 				<xsd:union memberTypes="xsd:string" />
 			</xsd:simpleType>
 		</xsd:attribute>
+		<xsd:attribute name="max-schema-agreement-wait-seconds" type="xsd:string" default="10" use="optional">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+Sets the maximum time to wait for schema agreement before returning from a DDL query. Defaults to 10 seconds.
+				]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
 		<xsd:attribute name="metrics-enabled" type="xsd:string"
 			default="true">
 			<xsd:annotation>
@@ -298,6 +329,23 @@ RetryPolicy implementation.
 				<xsd:union memberTypes="xsd:string" />
 			</xsd:simpleType>
 		</xsd:attribute>
+		<xsd:attribute name="speculative-execution-policy-ref" use="optional">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+Configures the speculative execution policy to use for the new cluster.
+				]]></xsd:documentation>
+			</xsd:annotation>
+			<xsd:simpleType>
+				<xsd:annotation>
+					<xsd:appinfo>
+						<tool:annotation kind="ref">
+							<tool:assignable-to type="com.datastax.driver.core.policies.SpeculativeExecutionPolicy"/>
+						</tool:annotation>
+					</xsd:appinfo>
+				</xsd:annotation>
+				<xsd:union memberTypes="xsd:string"/>
+			</xsd:simpleType>
+		</xsd:attribute>
 		<xsd:attribute name="ssl-enabled" type="xsd:string" default="false">
 			<xsd:annotation>
 				<xsd:documentation><![CDATA[
@@ -322,6 +370,23 @@ Custom SSL Options. sslEnabled must be true for sslOptions to be used.
 				<xsd:union memberTypes="xsd:string" />
 			</xsd:simpleType>
 		</xsd:attribute>
+		<xsd:attribute name="timestamp-generator-ref" use="optional">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+Configures the generator that will produce the client-side timestamp sent with each query.
+				]]></xsd:documentation>
+			</xsd:annotation>
+			<xsd:simpleType>
+				<xsd:annotation>
+					<xsd:appinfo>
+						<tool:annotation kind="ref">
+							<tool:assignable-to type="com.datastax.driver.core.TimestampGenerator"/>
+						</tool:annotation>
+					</xsd:appinfo>
+				</xsd:annotation>
+				<xsd:union memberTypes="xsd:string"/>
+			</xsd:simpleType>
+		</xsd:attribute>
 		<xsd:attribute name="username" type="xsd:string" use="optional">
 			<xsd:annotation>
 				<xsd:documentation><![CDATA[