Add PoolingOptions to CasandraProperties

This change allows users to configure some basic pooling
options for cassandra driver via configuration properties.

See gh-7946

spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/cassandra/CassandraAutoConfiguration.java

@@ -17,8 +17,11 @@
 package org.springframework.boot.autoconfigure.cassandra;
 
 import java.util.List;
+import java.util.Map;
 
 import com.datastax.driver.core.Cluster;
+import com.datastax.driver.core.HostDistance;
+import com.datastax.driver.core.PoolingOptions;
 import com.datastax.driver.core.QueryOptions;
 import com.datastax.driver.core.SocketOptions;
 import com.datastax.driver.core.policies.LoadBalancingPolicy;
@@ -89,6 +92,7 @@ public class CassandraAutoConfiguration {
 		if (properties.isSsl()) {
 			builder.withSSL();
 		}
+		builder.withPoolingOptions(getPoolingOptions());
 		String points = properties.getContactPoints();
 		builder.addContactPoints(StringUtils.commaDelimitedListToStringArray(points));
 
@@ -128,4 +132,13 @@ public class CassandraAutoConfiguration {
 		return options;
 	}
 
+	private PoolingOptions getPoolingOptions() {
+		PoolingOptions options = new PoolingOptions();
+		options.setHeartbeatIntervalSeconds(this.properties.getHeartbeatIntervalSeconds());
+		options.setMaxQueueSize(this.properties.getMaxQueueSize());
+		for (Map.Entry<HostDistance, Integer> entry : this.properties.getMaxRequestsPerConnection().entrySet()) {
+			options.setMaxRequestsPerConnection(entry.getKey(), entry.getValue());
+		}
+		return options;
+	}
 }

spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/cassandra/CassandraProperties.java

@@ -16,7 +16,12 @@
 
 package org.springframework.boot.autoconfigure.cassandra;
 
+import java.util.HashMap;
+import java.util.Map;
+
 import com.datastax.driver.core.ConsistencyLevel;
+import com.datastax.driver.core.HostDistance;
+import com.datastax.driver.core.PoolingOptions;
 import com.datastax.driver.core.ProtocolOptions;
 import com.datastax.driver.core.ProtocolOptions.Compression;
 import com.datastax.driver.core.QueryOptions;
@@ -113,6 +118,21 @@ public class CassandraProperties {
 	 */
 	private int readTimeoutMillis = SocketOptions.DEFAULT_READ_TIMEOUT_MILLIS;
 
+	/**
+	 * Pooling option: heartbeat interval.
+	 */
+	private int heartbeatIntervalSeconds = PoolingOptions.DEFAULT_HEARTBEAT_INTERVAL_SECONDS;
+
+	/**
+	 * Pooling option: max queue size.
+	 */
+	private int maxQueueSize = PoolingOptions.DEFAULT_MAX_QUEUE_SIZE;
+
+	/**
+	 * Pooling option: max requests per connection.
+	 */
+	private Map<HostDistance, Integer> maxRequestsPerConnection = new HashMap<HostDistance, Integer>();
+
 	/**
 	 * Schema action to take at startup.
 	 */
@@ -245,6 +265,30 @@ public class CassandraProperties {
 		this.readTimeoutMillis = readTimeoutMillis;
 	}
 
+	public int getHeartbeatIntervalSeconds() {
+		return this.heartbeatIntervalSeconds;
+	}
+
+	public void setHeartbeatIntervalSeconds(int heartbeatIntervalSeconds) {
+		this.heartbeatIntervalSeconds = heartbeatIntervalSeconds;
+	}
+
+	public int getMaxQueueSize() {
+		return this.maxQueueSize;
+	}
+
+	public void setMaxQueueSize(int maxQueueSize) {
+		this.maxQueueSize = maxQueueSize;
+	}
+
+	public Map<HostDistance, Integer> getMaxRequestsPerConnection() {
+		return this.maxRequestsPerConnection;
+	}
+
+	public void setMaxRequestsPerConnection(Map<HostDistance, Integer> maxRequestsPerConnection) {
+		this.maxRequestsPerConnection = maxRequestsPerConnection;
+	}
+
 	public boolean isSsl() {
 		return this.ssl;
 	}

spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/cassandra/CassandraAutoConfigurationTests.java

@@ -17,6 +17,7 @@
 package org.springframework.boot.autoconfigure.cassandra;
 
 import com.datastax.driver.core.Cluster;
+import com.datastax.driver.core.HostDistance;
 import org.junit.After;
 import org.junit.Test;
 
@@ -80,6 +81,30 @@ public class CassandraAutoConfigurationTests {
 		assertThat(cluster.getClusterName()).isEqualTo("overridden-name");
 	}
 
+	@Test
+	public void heartbeatInterval() {
+		load("spring.data.cassandra.heartbeat-interval-seconds=60");
+		assertThat(this.context.getBeanNamesForType(Cluster.class).length).isEqualTo(1);
+		Cluster cluster = this.context.getBean(Cluster.class);
+		assertThat(cluster.getConfiguration().getPoolingOptions().getHeartbeatIntervalSeconds()).isEqualTo(60);
+	}
+
+	@Test
+	public void maxQueueSize() {
+		load("spring.data.cassandra.max-queue-size=1024");
+		assertThat(this.context.getBeanNamesForType(Cluster.class).length).isEqualTo(1);
+		Cluster cluster = this.context.getBean(Cluster.class);
+		assertThat(cluster.getConfiguration().getPoolingOptions().getMaxQueueSize()).isEqualTo(1024);
+	}
+
+	@Test
+	public void maxRequestsPerConnection() {
+		load("spring.data.cassandra.max-requests-per-connection.local=100");
+		assertThat(this.context.getBeanNamesForType(Cluster.class).length).isEqualTo(1);
+		Cluster cluster = this.context.getBean(Cluster.class);
+		assertThat(cluster.getConfiguration().getPoolingOptions().getMaxRequestsPerConnection(HostDistance.LOCAL)).isEqualTo(100);
+	}
+
 	private void load(String... environment) {
 		load(null, environment);
 	}

spring-boot-docs/src/main/asciidoc/appendix-application-properties.adoc

@@ -588,8 +588,11 @@ content into your application; rather pick only the properties that you need.
 	spring.data.cassandra.consistency-level= # Queries consistency level.
 	spring.data.cassandra.contact-points=localhost # Comma-separated list of cluster node addresses.
 	spring.data.cassandra.fetch-size= # Queries default fetch size.
+	spring.data.cassandra.heartbeat-interval-seconds= # Pooling option: heartbeat interval.
 	spring.data.cassandra.keyspace-name= # Keyspace name to use.
 	spring.data.cassandra.load-balancing-policy= # Class name of the load balancing policy.
+	spring.data.cassandra.max-requests-per-connection.*= # Pooling option: max requests per connection.
+	spring.data.cassandra.max-queue-size= # Pooling option: max queue size.
 	spring.data.cassandra.port= # Port of the Cassandra server.
 	spring.data.cassandra.password= # Login password of the server.
 	spring.data.cassandra.reactive-repositories.enabled=true # Enable Cassandra reactive repositories.