From 2aafc7daa9198b523a9b39e7ba1fa64cb91b75c0 Mon Sep 17 00:00:00 2001 From: Vladimir Tsanev Date: Wed, 11 Jan 2017 02:10:33 +0200 Subject: [PATCH 1/2] Add PoolingOptions to CasandraProperties This change allows users to configure some basic pooling options for cassandra driver via configuration properties. See gh-7946 --- .../cassandra/CassandraAutoConfiguration.java | 13 ++++++ .../cassandra/CassandraProperties.java | 44 +++++++++++++++++++ .../CassandraAutoConfigurationTests.java | 25 +++++++++++ .../appendix-application-properties.adoc | 3 ++ 4 files changed, 85 insertions(+) diff --git a/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/cassandra/CassandraAutoConfiguration.java b/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/cassandra/CassandraAutoConfiguration.java index 0571276196..eeda52ef47 100644 --- a/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/cassandra/CassandraAutoConfiguration.java +++ b/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 entry : this.properties.getMaxRequestsPerConnection().entrySet()) { + options.setMaxRequestsPerConnection(entry.getKey(), entry.getValue()); + } + return options; + } } diff --git a/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/cassandra/CassandraProperties.java b/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/cassandra/CassandraProperties.java index ddcab2bc74..56aa3a431a 100644 --- a/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/cassandra/CassandraProperties.java +++ b/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 maxRequestsPerConnection = new HashMap(); + /** * 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 getMaxRequestsPerConnection() { + return this.maxRequestsPerConnection; + } + + public void setMaxRequestsPerConnection(Map maxRequestsPerConnection) { + this.maxRequestsPerConnection = maxRequestsPerConnection; + } + public boolean isSsl() { return this.ssl; } diff --git a/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/cassandra/CassandraAutoConfigurationTests.java b/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/cassandra/CassandraAutoConfigurationTests.java index e3cda12e07..5de72f7b4c 100644 --- a/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/cassandra/CassandraAutoConfigurationTests.java +++ b/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); } diff --git a/spring-boot-docs/src/main/asciidoc/appendix-application-properties.adoc b/spring-boot-docs/src/main/asciidoc/appendix-application-properties.adoc index fc3d6e422e..0d4203efc3 100644 --- a/spring-boot-docs/src/main/asciidoc/appendix-application-properties.adoc +++ b/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. From ba1bc45a53bac02c75fef328c9e43b070174dfab Mon Sep 17 00:00:00 2001 From: Stephane Nicoll Date: Thu, 1 Jun 2017 16:04:49 +0200 Subject: [PATCH 2/2] Polish "Add PoolingOptions to CasandraProperties" Closes gh-7946 --- .../cassandra/CassandraAutoConfiguration.java | 13 +- .../cassandra/CassandraProperties.java | 117 +++++++++++------- .../CassandraAutoConfigurationTests.java | 41 +++--- .../appendix-application-properties.adoc | 4 + 4 files changed, 106 insertions(+), 69 deletions(-) diff --git a/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/cassandra/CassandraAutoConfiguration.java b/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/cassandra/CassandraAutoConfiguration.java index eeda52ef47..ef308e84d8 100644 --- a/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/cassandra/CassandraAutoConfiguration.java +++ b/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/cassandra/CassandraAutoConfiguration.java @@ -17,10 +17,8 @@ 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; @@ -133,12 +131,13 @@ public class CassandraAutoConfiguration { } private PoolingOptions getPoolingOptions() { + CassandraProperties.Pool pool = this.properties.getPool(); PoolingOptions options = new PoolingOptions(); - options.setHeartbeatIntervalSeconds(this.properties.getHeartbeatIntervalSeconds()); - options.setMaxQueueSize(this.properties.getMaxQueueSize()); - for (Map.Entry entry : this.properties.getMaxRequestsPerConnection().entrySet()) { - options.setMaxRequestsPerConnection(entry.getKey(), entry.getValue()); - } + options.setIdleTimeoutSeconds(pool.getIdleTimeout()); + options.setPoolTimeoutMillis(pool.getPoolTimeout()); + options.setHeartbeatIntervalSeconds(pool.getHeartbeatInterval()); + options.setMaxQueueSize(pool.getMaxQueueSize()); return options; } + } diff --git a/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/cassandra/CassandraProperties.java b/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/cassandra/CassandraProperties.java index 56aa3a431a..1cd717e39d 100644 --- a/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/cassandra/CassandraProperties.java +++ b/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/cassandra/CassandraProperties.java @@ -1,5 +1,5 @@ /* - * Copyright 2012-2016 the original author or authors. + * Copyright 2012-2017 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -16,12 +16,7 @@ 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; @@ -38,6 +33,7 @@ import org.springframework.boot.context.properties.ConfigurationProperties; * @author Julien Dubois * @author Phillip Webb * @author Mark Paluch + * @author Stephane Nicoll * @since 1.3.0 */ @ConfigurationProperties(prefix = "spring.data.cassandra") @@ -118,21 +114,6 @@ 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 maxRequestsPerConnection = new HashMap(); - /** * Schema action to take at startup. */ @@ -143,6 +124,11 @@ public class CassandraProperties { */ private boolean ssl = false; + /** + * Pool configuration. + */ + private final Pool pool = new Pool(); + public String getKeyspaceName() { return this.keyspaceName; } @@ -265,30 +251,6 @@ 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 getMaxRequestsPerConnection() { - return this.maxRequestsPerConnection; - } - - public void setMaxRequestsPerConnection(Map maxRequestsPerConnection) { - this.maxRequestsPerConnection = maxRequestsPerConnection; - } - public boolean isSsl() { return this.ssl; } @@ -305,4 +267,69 @@ public class CassandraProperties { this.schemaAction = schemaAction; } + public Pool getPool() { + return this.pool; + } + + /** + * Pool properties. + */ + public static class Pool { + + /** + * Idle timeout (in seconds) before an idle connection is removed. + */ + private int idleTimeout = 120; + + /** + * Pool timeout (in milliseconds) when trying to acquire a connection from a + * host's pool. + */ + private int poolTimeout = 5000; + + /** + * Heartbeat interval (in seconds) after which a message is sent on an idle + * connection to make sure it's still alive. + */ + private int heartbeatInterval = 30; + + /** + * Maximum number of requests that get enqueued if no connection is available. + */ + private int maxQueueSize = 256; + + public int getIdleTimeout() { + return this.idleTimeout; + } + + public void setIdleTimeout(int idleTimeout) { + this.idleTimeout = idleTimeout; + } + + public int getPoolTimeout() { + return this.poolTimeout; + } + + public void setPoolTimeout(int poolTimeout) { + this.poolTimeout = poolTimeout; + } + + public int getHeartbeatInterval() { + return this.heartbeatInterval; + } + + public void setHeartbeatInterval(int heartbeatInterval) { + this.heartbeatInterval = heartbeatInterval; + } + + public int getMaxQueueSize() { + return this.maxQueueSize; + } + + public void setMaxQueueSize(int maxQueueSize) { + this.maxQueueSize = maxQueueSize; + } + + } + } diff --git a/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/cassandra/CassandraAutoConfigurationTests.java b/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/cassandra/CassandraAutoConfigurationTests.java index 5de72f7b4c..54cad5ae89 100644 --- a/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/cassandra/CassandraAutoConfigurationTests.java +++ b/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/cassandra/CassandraAutoConfigurationTests.java @@ -17,7 +17,7 @@ package org.springframework.boot.autoconfigure.cassandra; import com.datastax.driver.core.Cluster; -import com.datastax.driver.core.HostDistance; +import com.datastax.driver.core.PoolingOptions; import org.junit.After; import org.junit.Test; @@ -82,27 +82,34 @@ public class CassandraAutoConfigurationTests { } @Test - public void heartbeatInterval() { - load("spring.data.cassandra.heartbeat-interval-seconds=60"); + public void defaultPoolOptions() { + load(); assertThat(this.context.getBeanNamesForType(Cluster.class).length).isEqualTo(1); - Cluster cluster = this.context.getBean(Cluster.class); - assertThat(cluster.getConfiguration().getPoolingOptions().getHeartbeatIntervalSeconds()).isEqualTo(60); + PoolingOptions poolingOptions = this.context.getBean(Cluster.class) + .getConfiguration().getPoolingOptions(); + assertThat(poolingOptions.getIdleTimeoutSeconds()) + .isEqualTo(PoolingOptions.DEFAULT_IDLE_TIMEOUT_SECONDS); + assertThat(poolingOptions.getPoolTimeoutMillis()) + .isEqualTo(PoolingOptions.DEFAULT_POOL_TIMEOUT_MILLIS); + assertThat(poolingOptions.getHeartbeatIntervalSeconds()) + .isEqualTo(PoolingOptions.DEFAULT_HEARTBEAT_INTERVAL_SECONDS); + assertThat(poolingOptions.getMaxQueueSize()) + .isEqualTo(PoolingOptions.DEFAULT_MAX_QUEUE_SIZE); } @Test - public void maxQueueSize() { - load("spring.data.cassandra.max-queue-size=1024"); + public void customizePoolOptions() { + load("spring.data.cassandra.pool.idle-timeout=42", + "spring.data.cassandra.pool.pool-timeout=52", + "spring.data.cassandra.pool.heartbeat-interval=62", + "spring.data.cassandra.pool.max-queue-size=72"); 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); + PoolingOptions poolingOptions = this.context.getBean(Cluster.class) + .getConfiguration().getPoolingOptions(); + assertThat(poolingOptions.getIdleTimeoutSeconds()).isEqualTo(42); + assertThat(poolingOptions.getPoolTimeoutMillis()).isEqualTo(52); + assertThat(poolingOptions.getHeartbeatIntervalSeconds()).isEqualTo(62); + assertThat(poolingOptions.getMaxQueueSize()).isEqualTo(72); } private void load(String... environment) { diff --git a/spring-boot-docs/src/main/asciidoc/appendix-application-properties.adoc b/spring-boot-docs/src/main/asciidoc/appendix-application-properties.adoc index 0d4203efc3..96bad8ed71 100644 --- a/spring-boot-docs/src/main/asciidoc/appendix-application-properties.adoc +++ b/spring-boot-docs/src/main/asciidoc/appendix-application-properties.adoc @@ -595,6 +595,10 @@ content into your application; rather pick only the properties that you need. 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.pool.heartbeat-interval=30 # Heartbeat interval (in seconds) after which a message is sent on an idle connection to make sure it's still alive. + spring.data.cassandra.pool.idle-timeout=120 # Idle timeout (in seconds) before an idle connection is removed. + spring.data.cassandra.pool.max-queue-size=256 # Maximum number of requests that get enqueued if no connection is available. + spring.data.cassandra.pool.pool-timeout=5000 # Pool timeout (in milliseconds) when trying to acquire a connection from a host's pool. spring.data.cassandra.reactive-repositories.enabled=true # Enable Cassandra reactive repositories. spring.data.cassandra.read-timeout-millis= # Socket option: read time out. spring.data.cassandra.reconnection-policy= # Reconnection policy class.