DATACASS-344 - Upgrade Cassandra Driver to 3.1.1.

We are now compatible with Datastax’ Cassandra Driver 3.1.1 and support the newly introduced configuration options MaxQueueSize for PoolingOptions.

This change also removes test assertions for PoolTimeoutMillis as this option was deprecated and made unusable with 3.1.1.

.travis.yml

@@ -4,6 +4,9 @@ jdk:
 env:
   matrix:
     - PROFILE=ci
+    - PROFILE=ci CASSANDRA_DRIVER_VERSION=3.0.3
+    - PROFILE=ci CASSANDRA_DRIVER_VERSION=3.1.0
+    - PROFILE=ci CASSANDRA_DRIVER_VERSION=3.1.1
     - PROFILE=spring41-next
     - PROFILE=spring42
     - PROFILE=spring42-next
@@ -29,4 +32,4 @@ sudo: false
 before_install:
   - sed -i.bak -e 's|https://nexus.codehaus.org/snapshots/|https://oss.sonatype.org/content/repositories/codehaus-snapshots/|g' ~/.m2/settings.xml
 install: if [ ! -z ${CASSANDRA_VERSION} ]; then ./setup-cassandra.sh; fi;
-script: mvn clean install -P${PROFILE} -Dmaven.javadoc.skip=true
+script: mvn clean install -P${PROFILE} -Dcassandra-driver.version=${CASSANDRA_DRIVER_VERSION:-3.1.1} -Dmaven.javadoc.skip=true

pom.xml

@@ -70,7 +70,7 @@
 		<build.cassandra.ssl_storage_port>17001</build.cassandra.ssl_storage_port>
 		<build.cassandra.storage_port>17000</build.cassandra.storage_port>
 		<cassandra.version>3.9</cassandra.version>
-		<cassandra-driver.version>3.1.0</cassandra-driver.version>
+		<cassandra-driver.version>3.1.1</cassandra-driver.version>
 		<dist.id>spring-data-cassandra</dist.id>
 		<el.version>1.0</el.version>
 		<failsafe.version>2.16</failsafe.version>

spring-cql/src/main/java/org/springframework/cassandra/config/PoolingOptionsFactoryBean.java

@@ -15,6 +15,10 @@
  */
 package org.springframework.cassandra.config;
 
+
+import static org.springframework.util.ReflectionUtils.invokeMethod;
+
+import java.lang.reflect.Method;
 import java.util.concurrent.Executor;
 
 import org.springframework.beans.factory.FactoryBean;
@@ -22,6 +26,7 @@ import org.springframework.beans.factory.InitializingBean;
 
 import com.datastax.driver.core.HostDistance;
 import com.datastax.driver.core.PoolingOptions;
+import org.springframework.util.ReflectionUtils;
 
 /**
  * Spring {@link FactoryBean} for the Cassandra Java driver {@link PoolingOptions}.
@@ -37,6 +42,18 @@ import com.datastax.driver.core.PoolingOptions;
 @SuppressWarnings("unused")
 public class PoolingOptionsFactoryBean implements FactoryBean<PoolingOptions>, InitializingBean {
 
+	private static final PoolingOptions DEFAULT = new PoolingOptions();
+
+	private static final Method SET_MAX_QUEUE_SIZE;
+	private static final Method GET_MAX_QUEUE_SIZE;
+
+	static {
+		SET_MAX_QUEUE_SIZE = ReflectionUtils
+				.findMethod(PoolingOptions.class, "setMaxQueueSize", int.class);
+		GET_MAX_QUEUE_SIZE = ReflectionUtils
+				.findMethod(PoolingOptions.class, "getMaxQueueSize");
+	}
+
 	private Executor initializationExecutor;
 
 	private Integer heartbeatIntervalSeconds;
@@ -45,7 +62,12 @@ public class PoolingOptionsFactoryBean implements FactoryBean<PoolingOptions>, I
 	private Integer localMaxConnections;
 	private Integer localMaxSimultaneousRequests;
 	private Integer localMinSimultaneousRequests;
+
+	// Deprecated since Cassandra Driver 3.1.1
 	private Integer poolTimeoutMilliseconds;
+
+	// Available since Cassandra Driver 3.1.1
+	private int maxQueueSize;
 	private Integer remoteCoreConnections;
 	private Integer remoteMaxConnections;
 	private Integer remoteMaxSimultaneousRequests;
@@ -78,6 +100,23 @@ public class PoolingOptionsFactoryBean implements FactoryBean<PoolingOptions>, I
 		if (poolTimeoutMilliseconds != null) {
 			poolingOptions.setPoolTimeoutMillis(poolTimeoutMilliseconds);
 		}
+
+		if (!isDefaultMaxQueueSize() && SET_MAX_QUEUE_SIZE != null) {
+			invokeMethod(SET_MAX_QUEUE_SIZE, poolingOptions, maxQueueSize);
+		}
+	}
+
+	private boolean isDefaultMaxQueueSize() {
+
+		if(GET_MAX_QUEUE_SIZE != null){
+
+			Integer defaultMaxQueueSize = (Integer) invokeMethod(GET_MAX_QUEUE_SIZE, poolingOptions);
+			if(defaultMaxQueueSize.intValue() == maxQueueSize){
+				return true;
+			}
+		}
+
+		return false;
 	}
 
 	/*
@@ -247,6 +286,24 @@ public class PoolingOptionsFactoryBean implements FactoryBean<PoolingOptions>, I
 		return poolTimeoutMilliseconds;
 	}
 
+	/**
+	 * Sets the maximum number of requests that get enqueued if no connection is available.
+	 *
+	 * @param maxQueueSize maximum number of requests that get enqueued if no connection is available.
+	 */
+	public void setMaxQueueSize(Integer maxQueueSize) {
+		this.maxQueueSize = maxQueueSize;
+	}
+
+	/**
+	 * Gets the maximum number of requests that get enqueued if no connection is available.
+	 *
+	 * @return the {@code maxQueueSize}.
+	 */
+	public Integer getMaxQueueSize() {
+		return maxQueueSize;
+	}
+
 	/**
 	 * Sets the core number of connections per host for the {@link HostDistance#LOCAL} scope.
 	 *

spring-cql/src/main/java/org/springframework/cassandra/config/xml/CassandraCqlClusterParser.java

@@ -48,6 +48,7 @@ import com.datastax.driver.core.SocketOptions;
  * @author Matthew T. Adams
  * @author David Webb
  * @author John Blum
+ * @author Mark Paluch
  */
 public class CassandraCqlClusterParser extends AbstractBeanDefinitionParser {
 
@@ -95,6 +96,7 @@ public class CassandraCqlClusterParser extends AbstractBeanDefinitionParser {
 		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, "nettyOptions", element, "netty-options-ref");
 		addOptionalPropertyReference(builder, "reconnectionPolicy", element, "reconnection-policy-ref");
 		addOptionalPropertyReference(builder, "retryPolicy", element, "retry-policy-ref");
 		addOptionalPropertyReference(builder, "speculativeExecutionPolicy", element, "speculative-execution-policy-ref");
@@ -138,6 +140,7 @@ public class CassandraCqlClusterParser extends AbstractBeanDefinitionParser {
 		addOptionalPropertyValue(poolingOptionsBuilder, "heartbeatIntervalSeconds", element, "heartbeat-interval-seconds");
 		addOptionalPropertyValue(poolingOptionsBuilder, "idleTimeoutSeconds", element, "idle-timeout-seconds");
 		addOptionalPropertyValue(poolingOptionsBuilder, "poolTimeoutMilliseconds", element, "pool-timeout-milliseconds");
+		addOptionalPropertyValue(poolingOptionsBuilder, "maxQueueSize", element, "max-queue-size");
 
 		// parse child elements
 		for (Element subElement : DomUtils.getChildElements(element)) {

spring-cql/src/main/resources/org/springframework/cassandra/config/spring-cql-1.0.xsd

@@ -263,6 +263,16 @@ LoadBalancingPolicy implementation.
 				<xsd:union memberTypes="xsd:string" />
 			</xsd:simpleType>
 		</xsd:attribute>
+		<xsd:attribute name="max-queue-size" type="xsd:string" default="256" use="optional">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+Sets the maximum number of requests that get enqueued if no connection is available.
+If the queue grows past this value, new requests will be rejected immediately (and the driver will move to the
+next host in the query plan). This limit is per connection pool, not global to the driver.
+See com.datastax.driver.core.PoolingOption for more details. Available since Cassandra Driver 3.1.1.
+				]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
 		<xsd:attribute name="max-schema-agreement-wait-seconds" type="xsd:string" default="10" use="optional">
 			<xsd:annotation>
 				<xsd:documentation><![CDATA[

spring-cql/src/test/java/org/springframework/cassandra/config/PoolingOptionsFactoryBeanUnitTests.java

@@ -17,12 +17,15 @@ package org.springframework.cassandra.config;
 
 import static org.hamcrest.Matchers.*;
 import static org.junit.Assert.*;
+import static org.junit.Assume.assumeNotNull;
 import static org.mockito.Matchers.eq;
 import static org.mockito.Mockito.*;
 import static org.mockito.Mockito.any;
 import static org.mockito.Mockito.anyInt;
 import static org.mockito.Mockito.same;
+import static org.springframework.util.ReflectionUtils.invokeMethod;
 
+import java.lang.reflect.Method;
 import java.util.concurrent.Executor;
 
 import org.junit.Before;
@@ -36,6 +39,7 @@ import org.mockito.stubbing.Answer;
 
 import com.datastax.driver.core.HostDistance;
 import com.datastax.driver.core.PoolingOptions;
+import org.springframework.util.ReflectionUtils;
 
 /**
  * Unit tests for {@link PoolingOptionsFactoryBean}.
@@ -43,6 +47,7 @@ import com.datastax.driver.core.PoolingOptions;
  * @author Sumit Kumar
  * @author David Webb
  * @author John Blum
+ * @author Mark Paluch
  * @see <a href="https://jira.spring.io/browse/DATACASS-176">DATACASS-176</a>
  * @see <a href="https://jira.spring.io/browse/DATACASS-298">DATACASS-298</a>
  */
@@ -77,6 +82,7 @@ public class PoolingOptionsFactoryBeanUnitTests {
 
 	/**
 	 * @see <a href="https://jira.spring.io/browse/DATACASS-298">DATACASS-298</a>
+	 * @see <a href="https://jira.spring.io/browse/DATACASS-344">DATACASS-344</a>
 	 */
 	@Test
 	public void setAndGetFactoryBeanProperties() {
@@ -196,6 +202,36 @@ public class PoolingOptionsFactoryBeanUnitTests {
 		verify(poolingOptionsSpy, never()).setNewConnectionThreshold(eq(HostDistance.REMOTE), eq(5));
 	}
 
+	/**
+	 * @see <a href="https://jira.spring.io/browse/DATACASS-344">DATACASS-344</a>
+	 */
+	@Test
+	public void afterPropertiesSetInitializesMaxQueueSize() throws Exception {
+
+		Method setMaxQueueSize = ReflectionUtils
+				.findMethod(PoolingOptions.class, "setMaxQueueSize", int.class);
+
+		Method getMaxQueueSize = ReflectionUtils
+				.findMethod(PoolingOptions.class, "getMaxQueueSize");
+
+		assumeNotNull(setMaxQueueSize);
+
+		PoolingOptionsFactoryBean poolingOptionsFactoryBean = new PoolingOptionsFactoryBean() {
+			@Override
+			PoolingOptions newPoolingOptions() {
+				return poolingOptionsSpy;
+			}
+		};
+
+		poolingOptionsFactoryBean.setMaxQueueSize(1234);
+
+		poolingOptionsFactoryBean.afterPropertiesSet();
+
+		assertThat(poolingOptionsFactoryBean.getObject(), is(sameInstance(poolingOptionsSpy)));
+		assertThat(poolingOptionsFactoryBean.getObjectType(), is(equalTo((Class) poolingOptionsSpy.getClass())));
+		assertThat(invokeMethod(getMaxQueueSize, poolingOptionsSpy), is(equalTo((Object) 1234)));
+	}
+
 	/**
 	 * This particular test case is technically an integration test since it uses an actual instance of a DataStax Java
 	 * driver class type... {@link PoolingOptions}! The max values should be set before setting core values. Otherwise the

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

@@ -16,6 +16,7 @@
 
 package org.springframework.cassandra.test.integration;
 
+import static org.apache.cassandra.db.marshal.CompositeType.build;
 import static org.springframework.cassandra.test.integration.CassandraRule.InvocationMode.*;
 
 import java.util.ArrayList;
@@ -317,11 +318,12 @@ public class CassandraRule extends ExternalResource {
 			QueryOptions queryOptions = new QueryOptions();
 			queryOptions.setRefreshSchemaIntervalMillis(0);
 
-			cluster = new Cluster.Builder().addContactPoints(hostIp).//
-					withPort(port).//
-					withQueryOptions(queryOptions).//
-					withNettyOptions(FastShutdownNettyOptions.INSTANCE).//
-					build();
+			cluster = new Cluster.Builder().addContactPoints(hostIp) //
+					.withPort(port) //
+					.withMaxSchemaAgreementWaitSeconds(3) //
+					.withQueryOptions(queryOptions) //
+					.withNettyOptions(FastShutdownNettyOptions.INSTANCE) //
+					.build();
 		} else {
 			cluster = parent.cluster;
 			cassandraPort = parent.cassandraPort;

spring-cql/src/test/java/org/springframework/cassandra/test/integration/config/xml/PropertyPlaceholderNamespaceCreatingXmlConfigIntegrationTests.java

@@ -69,7 +69,6 @@ public class PropertyPlaceholderNamespaceCreatingXmlConfigIntegrationTests
 		assertThat(poolingOptions, is(notNullValue(PoolingOptions.class)));
 		assertThat(poolingOptions.getHeartbeatIntervalSeconds(), is(equalTo(60)));
 		assertThat(poolingOptions.getIdleTimeoutSeconds(), is(equalTo(180)));
-		assertThat(poolingOptions.getPoolTimeoutMillis(), is(equalTo(30000)));
 		assertThat(poolingOptions.getCoreConnectionsPerHost(HostDistance.LOCAL), is(equalTo(4)));
 		assertThat(poolingOptions.getMaxConnectionsPerHost(HostDistance.LOCAL), is(equalTo(8)));
 		assertThat(poolingOptions.getMaxRequestsPerConnection(HostDistance.LOCAL), is(equalTo(20)));

spring-cql/src/test/java/org/springframework/cassandra/test/integration/config/xml/XmlConfigIntegrationTests.java

@@ -97,7 +97,7 @@ public class XmlConfigIntegrationTests extends AbstractEmbeddedCassandraIntegrat
 	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().getProtocolOptions().getMaxSchemaAgreementWaitSeconds(), is(equalTo(2)));
 
 		assertThat(cluster.getConfiguration().getPolicies().getSpeculativeExecutionPolicy(),
 			is(equalTo(speculativeExecutionPolicy)));
@@ -123,7 +123,6 @@ public class XmlConfigIntegrationTests extends AbstractEmbeddedCassandraIntegrat
 		assertThat(poolingOptions.getHeartbeatIntervalSeconds(), is(equalTo(60)));
 		assertThat(poolingOptions.getIdleTimeoutSeconds(), is(equalTo(300)));
 		assertThat(poolingOptions.getInitializationExecutor(), is(equalTo(executor)));
-		assertThat(poolingOptions.getPoolTimeoutMillis(), is(equalTo(15000)));
 		assertThat(poolingOptions.getCoreConnectionsPerHost(HostDistance.LOCAL), is(equalTo(2)));
 		assertThat(poolingOptions.getMaxConnectionsPerHost(HostDistance.LOCAL), is(equalTo(8)));
 		assertThat(poolingOptions.getMaxRequestsPerConnection(HostDistance.LOCAL), is(equalTo(100)));

spring-cql/src/test/resources/org/springframework/cassandra/test/integration/config/xml/XmlConfigIntegrationTests-context.xml

@@ -30,7 +30,7 @@
 		heartbeat-interval-seconds="60"
 		initialization-executor-ref="testExecutor"
 		idle-timeout-seconds="300"
-		max-schema-agreement-wait-seconds="30"
+		max-schema-agreement-wait-seconds="2"
 		pool-timeout-milliseconds="15000"
 		speculative-execution-policy-ref="testSpeculativeExecutionPolicy"
 		timestamp-generator-ref="testTimestampGenerator">

spring-data-cassandra/src/main/resources/org/springframework/data/cassandra/config/spring-cassandra-1.0.xsd

@@ -272,6 +272,16 @@ LoadBalancingPolicy implementation.
 				<xsd:union memberTypes="xsd:string" />
 			</xsd:simpleType>
 		</xsd:attribute>
+		<xsd:attribute name="max-queue-size" type="xsd:string" default="256" use="optional">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+Sets the maximum number of requests that get enqueued if no connection is available.
+If the queue grows past this value, new requests will be rejected immediately (and the driver will move to the
+next host in the query plan). This limit is per connection pool, not global to the driver.
+See com.datastax.driver.core.PoolingOption for more details. Available since Cassandra Driver 3.1.1.
+				]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
 		<xsd:attribute name="max-schema-agreement-wait-seconds" type="xsd:string" default="10" use="optional">
 			<xsd:annotation>
 				<xsd:documentation><![CDATA[

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/test/integration/mapping/types/CassandraTypeMappingIntegrationTest.java

@@ -717,21 +717,6 @@ public class CassandraTypeMappingIntegrationTest extends AbstractSpringDataEmbed
 		assertThat(loaded.getBpZoneId(), is(equalTo(entity.getBpZoneId())));
 	}
 
-	/**
-	 * @see DATACASS-271
-	 */
-	@Test(expected = InvalidQueryException.class)
-	public void insertFailsOnWriteTime() {
-
-		// writing of time is not supported with Insert/Update statements as they mix up types.
-		// The only way to insert a time right now seems a PreparedStatement
-		String id = "1";
-		long time = 21312214L;
-
-		Insert insert = QueryBuilder.insertInto("timeentity").value("id", id).value("time", time);
-		cassandraOperations.getSession().execute(insert);
-	}
-
 	/**
 	 * @see DATACASS-285
 	 */