DATACASS-335 - Reuse Cluster connection resources during tests where possible.

High client churn seems to affect Cassandra in a negative way (connection timeouts, driver considers hosts as down). Almost all integration tests bootstrap their own Cluster instance and dispose it once the tests has finished. Identify tests where reuse of a global Cluster instance provided by CassandraRule is possible and switch from context bootstrapping to reuse. This helps to prevent integration test failures.

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

@@ -16,13 +16,13 @@
 
 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;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.TimeUnit;
 
 import org.junit.rules.ExternalResource;
 import org.springframework.cassandra.core.SessionCallback;
@@ -36,6 +36,7 @@ import org.springframework.util.SocketUtils;
 import com.datastax.driver.core.Cluster;
 import com.datastax.driver.core.QueryOptions;
 import com.datastax.driver.core.Session;
+import com.datastax.driver.core.SocketOptions;
 
 /**
  * Rule to provide a Cassandra context for integration tests. This rule can use/spin up either an embedded Cassandra
@@ -55,6 +56,8 @@ import com.datastax.driver.core.Session;
  */
 public class CassandraRule extends ExternalResource {
 
+	private static ResourceHolder resourceHolder;
+
 	private final CassandraConnectionProperties properties = new CassandraConnectionProperties();
 	private final String configurationFileName;
 	private final long startUpTimeout;
@@ -318,28 +321,51 @@ public class CassandraRule extends ExternalResource {
 			QueryOptions queryOptions = new QueryOptions();
 			queryOptions.setRefreshSchemaIntervalMillis(0);
 
-			cluster = new Cluster.Builder().addContactPoints(hostIp) //
-					.withPort(port) //
-					.withMaxSchemaAgreementWaitSeconds(3) //
-					.withQueryOptions(queryOptions) //
-					.withNettyOptions(FastShutdownNettyOptions.INSTANCE) //
-					.build();
+			SocketOptions socketOptions = new SocketOptions();
+			socketOptions.setConnectTimeoutMillis((int) TimeUnit.SECONDS.toMillis(15));
+			socketOptions.setReadTimeoutMillis((int) TimeUnit.SECONDS.toMillis(15));
+
+			if (resourceHolder == null) {
+
+				cluster = new Cluster.Builder().addContactPoints(hostIp) //
+						.withPort(port) //
+						.withQueryOptions(queryOptions) //
+						.withMaxSchemaAgreementWaitSeconds(3) //
+						.withSocketOptions(socketOptions) //
+						.withNettyOptions(FastShutdownNettyOptions.INSTANCE) //
+						.build();
+
+				if (properties.getBoolean("build.cassandra.reuse-cluster")) {
+					resourceHolder = new ResourceHolder(cluster, cluster.connect());
+				}
+			} else {
+				cluster = resourceHolder.cluster;
+			}
+
 		} else {
 			cluster = parent.cluster;
 			cassandraPort = parent.cassandraPort;
 		}
 
-		session = cluster.connect();
+		if (parent != null) {
+			session = parent.getSession();
+		} else if (resourceHolder == null) {
+			session = cluster.connect();
+		} else {
+			session = resourceHolder.session;
+		}
 	}
 
 	private void cleanupConnection() {
 
-		if (parent == null) {
-			session.close();
-			cluster.closeAsync();
-			cluster = null;
-		} else {
-			session.closeAsync();
+		if (resourceHolder == null) {
+			if (parent == null) {
+				session.close();
+				cluster.closeAsync();
+				cluster = null;
+			} else {
+				session.closeAsync();
+			}
 		}
 
 		session = null;
@@ -396,4 +422,24 @@ public class CassandraRule extends ExternalResource {
 
 		}
 	}
+
+	private static class ResourceHolder {
+
+		private Cluster cluster;
+		private Session session;
+
+		public ResourceHolder(final Cluster cluster, final Session session) {
+			this.cluster = cluster;
+			this.session = session;
+
+			Runtime.getRuntime().addShutdownHook(new Thread() {
+
+				@Override
+				public void run() {
+					session.close();
+					cluster.close();
+				}
+			});
+		}
+	}
 }

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

@@ -40,7 +40,7 @@ import com.datastax.driver.core.Session;
  */
 public class KeyspaceRule extends ExternalResource {
 
-	private Cluster cluster;
+	private final CassandraRule cassandraRule;
 	private Session session;
 	private final String keyspaceName;
 
@@ -65,43 +65,26 @@ public class KeyspaceRule extends ExternalResource {
 		Assert.notNull(cassandraRule, "CassandraRule must not be null!");
 		Assert.hasText(keyspaceName, "KeyspaceName must not be empty!");
 
-		// Support initialized and initializing CassandraRule.
-		if (cassandraRule.getCluster() != null) {
-			this.cluster = cassandraRule.getCluster();
-			this.session = cluster.connect();
-		} else {
-			cassandraRule.before(new SessionCallback<Object>() {
-				@Override
-				public Object doInSession(Session s) throws DataAccessException {
-					KeyspaceRule.this.cluster = s.getCluster();
-					KeyspaceRule.this.session = cluster.connect();
-					return null;
-				}
-			});
-		}
-		this.keyspaceName = keyspaceName;
-	}
-
-	/**
-	 * Create a {@link KeyspaceRule} initialized with a {@link Cluster} for creating a keyspace using the given
-	 * {@code keyspaceName}.
-	 *
-	 * @param cluster
-	 * @param keyspaceName
-	 */
-	public KeyspaceRule(Cluster cluster, String keyspaceName) {
-
-		Assert.notNull(cluster, "Cluster must not be null!");
-		Assert.hasText(keyspaceName, "KeyspaceName must not be empty!");
-
-		this.cluster = cluster;
-		this.session = cluster.connect();
 		this.keyspaceName = keyspaceName;
+		this.cassandraRule = cassandraRule;
 	}
 
 	@Override
 	protected void before() throws Throwable {
 
+		// Support initialized and initializing CassandraRule.
+		if (cassandraRule.getCluster() != null) {
+			this.session = cassandraRule.getSession();
+		} else {
+			cassandraRule.before(new SessionCallback<Object>() {
+				@Override
+				public Object doInSession(Session s) throws DataAccessException {
+					KeyspaceRule.this.session = cassandraRule.getSession();
+					return null;
+				}
+			});
+		}
+
 		Assert.state(session != null, "Session was not initialized");
 
 		session.execute(String.format("CREATE KEYSPACE %s WITH durable_writes = false AND "

spring-cql/src/test/resources/config/cassandra-connection.properties

@@ -6,3 +6,4 @@ build.cassandra.storage_port=@build.cassandra.storage_port@
 build.cassandra.ssl_storage_port=@build.cassandra.ssl_storage_port@
 build.cassandra.mode=@build.cassandra.mode@
 build.cassandra.host=@build.cassandra.host@
+build.cassandra.reuse-cluster=true